poolder

defmodule Poolder.FactoryPool do
  @moduledoc """
  Dynamic pool of grouped processes with fast lookup by group or pid.

  ## Usage examples

    # Start a worker under group :http
    Poolder.FactoryPool.start_child(:http, {MyWorker, args})

    # Synchronous call to specific pid
    Poolder.FactoryPool.call(pid, :ping)

    # Async message to a specific pid
    Poolder.FactoryPool.cast(pid, :refresh)

    # Broadcast to all workers in a group
    Poolder.FactoryPool.broadcast(:http, :reload)

    # Count active workers in a group
    Poolder.FactoryPool.count(:http)
  """

  defmacro __using__(opts) do
    name = Keyword.get(opts, :name)
    monitor = Keyword.get(opts, :monitor, Poolder.Monitor.Default)
    table = Keyword.get(opts, :table)
    supervisor = Keyword.get(opts, :supervisor, Poolder.DynamicSupervisor)
    restart = Keyword.get(opts, :restart, :transient)
    caller = Keyword.get(opts, :caller, &GenServer.call/2)

    quote bind_quoted: [
            name: name,
            monitor: monitor,
            table: table,
            caller: caller,
            supervisor: supervisor,
            restart: restart
          ] do
      @name name || __MODULE__
      @table table || __MODULE__
      # :bag — {group, pid}
      @group_table table || Module.concat(table, GroupTable)
      # :set — {pid, group}
      @pid_table Module.concat(table, PidTable)
      @caller caller
      @supervisor supervisor
      @restart restart
      @monitor monitor
      @default_group :default

      @behaviour Poolder.FactoryPool
      alias Poolder.Ets

      def child_spec(_opts) do
        %{
          id: @name,
          start: {__MODULE__, :start_link, []},
          type: :worker,
          restart: @restart,
          shutdown: 500
        }
      end

      @doc "Initializes ETS tables for group and pid tracking"
      def start_link do
        Ets.new(@group_table, [:bag, :public, :named_table, read_concurrency: true])
        Ets.new(@pid_table, [:set, :public, :named_table, read_concurrency: true])

        case @monitor.start_link({@group_table, @pid_table}) do
          {:ok, pid} ->
            {:ok, pid}

          {:error, {:already_started, _pid}} ->
            :ignore
        end
      end

      @doc "Starts a worker process under a group and tracks it in ETS"
      def start_child(group, {mod, args}) do
        group = group || @default_group
        sup_name = via_supervisor(group)

        spec = %{
          id: make_ref(),
          start: {mod, :start_link, [args]},
          restart: :temporary
        }

        with {:ok, pid} <- DynamicSupervisor.start_child(sup_name, spec) do
          :ets.insert(@group_table, {group, pid})
          :ets.insert(@pid_table, {pid, group})
          @monitor.monitor(pid)
          {:ok, pid}
        end
      end

      @doc "Terminates a child process and removes its ETS references"
      def terminate(pid) when is_pid(pid) do
        case :ets.lookup(@pid_table, pid) do
          [{^pid, group}] ->
            :ets.delete_object(@group_table, {group, pid})
            :ets.delete(@pid_table, pid)
            DynamicSupervisor.terminate_child(via_supervisor(group), pid)

          [] ->
            {:error, :not_found}
        end
      end

      @doc "Terminates all child processes in a group and removes their ETS references"
      def terminate_group(group) do
        sup = via_supervisor(group)

        :ets.lookup(@group_table, group)
        |> Enum.each(fn {^group, pid} ->
          :ets.delete(@pid_table, pid)
          DynamicSupervisor.terminate_child(sup, pid)
        end)

        :ets.delete(@group_table, group)
      end

      @doc "Returns a list of all active workers in the given group"
      def list(group \\ @default_group) do
        :ets.foldl(fn {^group, pid}, acc -> [pid | acc] end, [], @group_table)
      end

      @doc "Returns a list of all active workers and their groups"
      def list_all do
        :ets.foldl(fn {pid, group}, acc -> [{pid, group} | acc] end, [], @pid_table)
      end

      @doc "Returns the number of active workers in the given group"
      def count(group \\ @default_group) do
        :ets.lookup(@group_table, group)
        |> Enum.count()
      end

      @doc "Returns the number of active workers in all groups"
      def count_all do
        :ets.info(@group_table, :size)
      end

      @doc "Sends a synchronous GenServer call to a specific pid"
      def call(pid, msg) when is_pid(pid) do
        @caller.(pid, msg)
      end

      @doc "Performs a guarded GenServer call to a pid under specific group"
      def call(group, pid, msg) when is_pid(pid) do
        case :ets.lookup(@pid_table, pid) do
          [{^pid, ^group}] -> @caller.(pid, msg)
          _ -> {:error, :not_found}
        end
      end

      @doc "Sends an asynchronous message to a specific pid"
      def cast(pid, msg) when is_pid(pid) do
        send(pid, msg)
        :ok
      end

      @doc "Sends a guarded asynchronous message to a pid in a group"
      def cast(group, pid, msg) when is_pid(pid) do
        case :ets.lookup(@pid_table, pid) do
          [{^pid, ^group}] -> send(pid, msg)
          _ -> {:error, :not_found}
        end
      end

      @doc "Sends a message to all registered pids in a group"
      def broadcast(group, msg) do
        :ets.lookup(@group_table, group)
        |> Enum.each(fn {^group, pid} -> send(pid, msg) end)
      end

      # Internal: creates or retrieves named DynamicSupervisor per group
      defp via_supervisor(group) do
        global_name = {:sup, group}

        case :global.whereis_name(global_name) do
          :undefined ->
            {:ok, _pid} =
              @supervisor.start_link({:global, global_name}, [])

            {:global, global_name}

          _pid ->
            {:global, global_name}
        end
      end
    end
  end

  @callback start_link() :: {:ok, pid} | :ignore
  @callback start_child(group :: atom, {module, args :: any}) :: {:ok, pid} | {:error, any}
  @callback terminate(pid :: pid) :: :ok | {:error, any}
  @callback terminate_group(group :: atom) :: :ok | {:error, any}
  @callback list(group :: atom) :: [pid]
  @callback list_all() :: [{pid, atom}]
  @callback count(group :: atom) :: non_neg_integer
  @callback count_all() :: non_neg_integer
  @callback call(pid :: pid, msg :: any) :: any
  @callback call(group :: atom, pid :: pid, msg :: any) :: any
  @callback cast(pid :: pid, msg :: any) :: :ok
  @callback cast(group :: atom, pid :: pid, msg :: any) :: :ok
  @callback broadcast(group :: atom, msg :: any) :: :ok
end