poolder

defmodule Poolder.PubSub.Monitor do
  @moduledoc """
  Lightweight monitor process (not a GenServer) that:
    - Creates monitors for subscribed pids
    - Demonitor pids that no longer have subscriptions
    - Cleans up subscriptions upon receiving :DOWN
  """

  @doc "Starts the monitor process linked to the given ETS."
  def start_link(table) do
    pid = spawn_link(__MODULE__, :loop, [table])
    {:ok, pid}
  end

  @doc false
  def loop(table) do
    receive do
      {:ensure_monitor, ^table, pid} when is_pid(pid) ->
        ensure_monitor_entry(table, pid)
        loop(table)

      {:maybe_demonitor, ^table, pid} when is_pid(pid) ->
        maybe_demonitor_entry(table, pid)
        loop(table)

      {:DOWN, ref, :process, pid, _reason} ->
        # Verify that the :DOWN corresponds to a monitor that we manage
        case :ets.lookup(table, {:mon, pid}) do
          [{{:mon, ^pid}, ^ref}] ->
            cleanup_pid(table, pid)
            loop(table)

          _ ->
            loop(table)
        end

      :stop ->
        :ok
    after
      300_000 ->
        :erlang.hibernate(__MODULE__, :loop, [table])
    end
  end

  # --- internal helpers ---

  defp ensure_monitor_entry(table, pid) do
    case :ets.lookup(table, {:mon, pid}) do
      [] ->
        ref = Process.monitor(pid)
        :ets.insert(table, {{:mon, pid}, ref})

      _ ->
        :ok
    end
  end

  defp maybe_demonitor_entry(table, pid) do
    case has_any_subscription?(table, pid) do
      true ->
        :ok

      false ->
        case :ets.lookup(table, {:mon, pid}) do
          [{{:mon, ^pid}, ref}] ->
            Process.demonitor(ref, [:flush])
            :ets.delete(table, {:mon, pid})

          _ ->
            :ok
        end
    end
  end

  defp has_any_subscription?(table, pid) do
    case :ets.select(table, [
           {{{:sub, :_, pid}, :_}, [], [true]}
         ]) do
      [] -> false
      _ -> true
    end
  end

  defp cleanup_pid(table, pid) do
    # Removes all subscriptions {topic, pid}
    subs =
      :ets.select(table, [
        {{{:sub, :"$1", pid}, :_}, [], [:"$1"]}
      ])

    Enum.each(subs, fn topic ->
      :ets.delete(table, {:sub, topic, pid})
    end)

    :ets.delete(table, {:mon, pid})
  end
end