finitomata

defmodule Infinitomata do
  @moduledoc """
  The sibling of `Finitomata`, but runs transparently in the cluster.

  If you want to use a _stateful consistent hash ring_ like [`libring`](https://hexdocs.pm/libring),
    implement the behaviour `Finitomata.ClusterInfo` wrapping calls to it and 
    invoke `Finitomata.ClusterInfo.init(Impl)` before using `Infinitomata.start_fsm/4`.

  The example of such an implementation for `libring` (assuming the named ring `@ring`
    has been started in the supervision tree) follows.

  ```elixir
  defmodule MyApp.ClusterInfo do
    @moduledoc false
    @behaviour Finitomata.ClusterInfo

    @impl Finitomata.ClusterInfo
    def nodes, do: HashRing.nodes(@ring)

    @impl Finitomata.ClusterInfo
    def whois(id), do: HashRing.key_to_node(@ring, id) 
  end
  ```
  """
  @moduledoc since: "0.15.0"

  @max_attempts Application.compile_env(:finitomata, :infinitomata_attempts, 1_000)

  require Logger

  alias Finitomata.{ClusterInfo, State, Transition}
  alias Finitomata.Distributed.GroupMonitor, as: InfMon
  alias Finitomata.Distributed.Supervisor, as: InfSup
  alias Finitomata.Supervisor, as: FinSup

  @behaviour Finitomata.Supervisor

  @doc since: "0.16.0"
  def start_link(id \\ nil, nodes \\ Node.list()) do
    InfSup.start_link(id, nodes)
  end

  @doc since: "0.16.0"
  @impl Finitomata.Supervisor
  def child_spec(id \\ nil) do
    Supervisor.child_spec({InfSup, id}, id: {InfSup, id})
  end

  defp distributed_call(fun, id, target, args \\ []) do
    do_distributed_call(node(), fun, FinSup.infinitomata_name(id), target, args, @max_attempts)
  end

  defp do_distributed_call(:nonode@nohost, fun, id, target, args, _attempts) do
    apply(Finitomata, fun, [id, target | args])
  end

  defp do_distributed_call(_this, _fun, _id, _target, _args, 0), do: {:error, :exhausted}

  defp do_distributed_call(this, fun, id, target, args, attempts) do
    case InfSup.get(id, target) do
      %{node: node} ->
        with {:badrpc, error} <-
               :rpc.call(node, Finitomata, fun, [id, target | args]) do
          Logger.error(
            "[♻️] Distributed: " <> inspect(id: id, node: node, target: target, error: error)
          )

          :ok = synch(id)
          do_distributed_call(this, fun, id, target, args, attempts - 1)
        end

      nil ->
        Process.sleep(1)
        do_distributed_call(this, fun, id, target, args, attempts - 1)

      _ ->
        {:error, :not_started}
    end
  end

  @doc since: "0.16.0"
  @doc "Count of children"
  def count(id \\ nil) do
    id
    |> FinSup.infinitomata_name()
    |> InfMon.count()
  end

  @doc since: "0.16.0"
  @impl Finitomata.Supervisor
  @spec all(Finitomata.id()) :: %{
          optional(Finitomata.fsm_name()) => %{pid: pid(), node: node(), reference: reference()}
        }
  def all(id \\ nil) do
    id
    |> FinSup.infinitomata_name()
    |> InfSup.all()
  end

  @doc since: "0.18.0"
  @doc "Returns the random _FSM_ from the pool"
  def random(id \\ nil) do
    id
    |> all()
    |> Map.keys()
    |> case do
      [] ->
        Process.sleep(200)
        random(id)

      [_ | _] = nodes ->
        Enum.random(nodes)
    end
  end

  @doc since: "0.19.0"
  @doc "Synchronizes the local `Infinitomata` instance with the cluster"
  def synch(id \\ nil) do
    InfSup.synch(id, FinSup.infinitomata_name(id))
  end

  @doc since: "0.15.0"
  @doc """
  Starts the _FSM_ in the distributed environment. See `Finitomata.start_fsm/4` for docs and options
  """
  @impl Finitomata.Supervisor
  @spec start_fsm(Finitomata.id(), Finitomata.fsm_name(), module(), any()) ::
          DynamicSupervisor.on_start_child()
  def start_fsm(id \\ nil, target, implementation, payload) do
    id = FinSup.infinitomata_name(id)

    case InfSup.get(id, target) do
      nil ->
        node = ClusterInfo.whois({id, target})
        do_start_fsm(node == node(), node, id, target, implementation, payload)

      %{node: node, pid: pid} ->
        {:error, {:already_started, {node, pid}}}
    end
  end

  @spec do_start_fsm(boolean(), node(), Finitomata.id(), Finitomata.fsm_name(), module(), any()) ::
          DynamicSupervisor.on_start_child()
  defp do_start_fsm(false, node, id, target, implementation, payload) do
    case :rpc.block_call(node, Finitomata, :start_fsm, [id, target, implementation, payload]) do
      {:ok, pid} ->
        # local_pid = :rpc.call(node, :erlang, :list_to_pid, [:erlang.pid_to_list(pid)]).
        :ok = :rpc.block_call(node, :pg, :join, [InfSup.group(id), pid])
        {:ok, pid}

      {:error, {:already_started, pid}} ->
        {:error, {:already_started, {node, pid}}}

      {error, reason} when error in [:error, :badrpc] ->
        {:error, reason}
    end
  end

  defp do_start_fsm(true, node, id, target, implementation, payload) do
    case Finitomata.start_fsm(id, target, implementation, payload) do
      {:ok, pid} ->
        :ok = :pg.join(InfSup.group(id), pid)
        {:ok, pid}

      {:error, {:already_started, pid}} ->
        {:error, {:already_started, {node, pid}}}

      {:error, reason} ->
        {:error, reason}
    end
  end

  @doc since: "0.19.0"
  @impl Finitomata.Supervisor
  @spec timer_tick(Finitomata.id(), Finitomata.fsm_name()) :: :ok
  def timer_tick(id \\ nil, target) do
    distributed_call(:timer_tick, id, target)
  end

  @doc since: "0.15.0"
  @impl Finitomata.Supervisor
  @spec transition(
          Finitomata.id(),
          Finitomata.fsm_name(),
          Transition.event() | {Transition.event(), State.payload()},
          non_neg_integer()
        ) ::
          :ok
  def transition(id \\ nil, target, event_payload, delay \\ 0) do
    distributed_call(:transition, id, target, [event_payload, delay])
  end

  @doc since: "0.15.0"
  @impl Finitomata.Supervisor
  def state(id \\ nil, target, reload? \\ :full) do
    distributed_call(:state, id, target, [reload?])
  end

  @doc since: "0.26.0"
  @impl Finitomata.Supervisor
  def alive?(id \\ nil, target) do
    distributed_call(:alive?, id, target)
  end
end