finitomata

defmodule Finitomata.Distributed.Supervisor do
  @moduledoc false

  require Logger

  use Supervisor

  def start_link(id, nodes \\ Node.list()) do
    __MODULE__
    |> Supervisor.start_link(id, name: sup_name(id))
    |> then(fn
      {:ok, pid} when is_pid(pid) ->
        Enum.each(nodes, fn node ->
          with {:badrpc, error} <- :rpc.block_call(node, __MODULE__, :start_link, [id, []]) do
            Logger.error(
              "[♻️] Remote start: " <>
                inspect(id: id, name: sup_name(id), node: node, error: error)
            )
          end
        end)

        {:ok, _task_pid} = Task.start(fn -> synch(id, nodes) end)
        {:ok, pid}

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

      other ->
        other
    end)
  end

  @impl true
  def init(id) do
    id = Finitomata.Supervisor.infinitomata_name(id)

    children = [
      %{id: {:pg, id}, start: {__MODULE__, :start_pg, []}},
      {Finitomata, id},
      %{id: {Agent, id}, start: {Agent, :start_link, [fn -> %{} end, [name: agent(id)]]}},
      {Finitomata.Distributed.GroupMonitor, id}
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end

  defp sup_name(id), do: id |> Finitomata.Supervisor.infinitomata_name() |> Module.concat(Sup)

  defp agent(nil), do: agent(__MODULE__)
  defp agent(id), do: Module.concat(id, IdLookup)

  def group(nil), do: group(__MODULE__)
  def group(id), do: Module.concat(id, Group)

  def ungroup(id), do: id |> Module.split() |> Enum.slice(0..-2//1) |> Module.concat()

  def synch(id, nodes \\ Node.list(), fqn_id \\ nil) do
    fqn_id = if is_nil(fqn_id), do: Finitomata.Supervisor.infinitomata_name(id), else: fqn_id
    known_processes_alive = fqn_id |> group() |> :pg.get_members()
    domestic = Map.new(Infinitomata.all(id), &fix_pid(&1, known_processes_alive))

    known_fsms_alive = fn _ ->
      merger =
        fn
          _k, %{node: node, pid: pid}, %{node: node, pid: pid} ->
            %{node: node, pid: pid, ref: make_ref()}

          _k, %{node: _node, pid: nil}, %{node: node, pid: pid} ->
            %{node: node, pid: pid, ref: make_ref()}

          _k, %{node: node, pid: pid}, %{node: _node, pid: _pid} ->
            %{node: node, pid: pid, ref: make_ref()}
        end

      call_handler = fn result, id, node ->
        with {:badrpc, error} <- result do
          Logger.warning("[♻️] Synch Error: " <> inspect(id: id, node: node, error: error))
          []
        end
      end

      Enum.reduce(nodes, domestic, fn node, acc ->
        node
        |> :rpc.block_call(Infinitomata, :all, [id])
        |> call_handler.(id, node)
        |> Map.new(&fix_pid(&1, known_processes_alive))
        |> Map.merge(acc, merger)
      end)
    end

    fqn_id |> agent() |> Agent.update(known_fsms_alive)
  end

  def all(id) do
    if is_nil(GenServer.whereis(agent(id))) do
      Logger.debug("[♻️] Not healthy ‹#{inspect(id)}›, probably it’s terminating")
      %{}
    else
      with empty when empty == %{} <- Agent.get(agent(id), & &1) do
        Logger.debug("[♻️] Empty pool for ‹#{inspect(id)}›")
        %{}
      end
    end
  end

  def del(id, name), do: Agent.update(agent(id), &Map.delete(&1, name))

  def get(id, name), do: Agent.get(agent(id), &Map.get(&1, name))

  def select_by_pids(id, pids) when is_list(pids) do
    Agent.get(
      agent(id),
      &Map.filter(&1, fn {_, %{pid: pid}} -> pid in pids end)
    )
  end

  def delete_by_pids(id, pids) when is_list(pids) do
    {match, mismatch} =
      Agent.get(agent(id), &split_with(&1, fn {_, %{pid: pid}} -> pid in pids end))

    Agent.update(agent(id), fn _ -> mismatch end)
    match
  end

  def delete_by_pids(id, %MapSet{} = pids) do
    {match, mismatch} =
      Agent.get(agent(id), &split_with(&1, fn {_, %{pid: pid}} -> MapSet.member?(pids, pid) end))

    Agent.update(agent(id), fn _ -> mismatch end)
    match
  end

  def put(id, name, %{pid: pid} = data) when is_pid(pid) do
    Agent.update(agent(id), &Map.put(&1, name, data))
  end

  defp skip_node(nil), do: nil
  defp skip_node(pid) when is_pid(pid), do: pid |> :erlang.pid_to_list() |> skip_node()
  defp skip_node([?. | tail]), do: tail
  defp skip_node([_ | tail]), do: skip_node(tail)

  defp fix_pid({name, %{pid: pid} = value}, pids),
    do: {name, %{value | pid: fix_pid(pid, pids)}}

  defp fix_pid(pid, pids) do
    pid_tail = skip_node(pid)

    Enum.find(pids, &(skip_node(&1) == pid_tail))
  end

  if Version.compare(System.version(), "1.15.0") == :lt do
    defp split_with(%{} = map, fun) when is_function(fun, 1) do
      {truthy, falsey} = Enum.split_with(map, fun)
      {Map.new(truthy), Map.new(falsey)}
    end
  else
    defdelegate split_with(map, fun), to: Map
  end

  @doc false
  def start_pg do
    with {:error, {:already_started, _pid}} <- :pg.start_link(), do: :ignore
  end
end