agent_session_manager

defmodule ASM.Extensions.PubSub.Adapters.Local do
  @moduledoc """
  Local in-node PubSub adapter built on `Registry` duplicate keys.

  Broadcast message contract:
  - each subscriber receives `{:asm_pubsub, topic, payload}`
  """

  @behaviour ASM.Extensions.PubSub.Adapter

  alias ASM.Error

  @default_registry :asm_ext_pubsub

  @type registry_ref :: atom() | pid()
  @type state :: %{registry: registry_ref()}

  @spec start_link(keyword()) :: GenServer.on_start() | {:error, Error.t()}
  def start_link(opts \\ []) when is_list(opts) do
    with {:ok, registry} <- normalize_registry(Keyword.get(opts, :registry, @default_registry)) do
      case registry do
        name when is_atom(name) ->
          Registry.start_link(keys: :duplicate, name: name)

        _pid ->
          {:error, config_error("local adapter start_link/1 requires atom :registry name")}
      end
    end
  end

  @impl true
  def init(opts) when is_list(opts) do
    auto_start? = Keyword.get(opts, :auto_start, true)

    with {:ok, registry} <- normalize_registry(Keyword.get(opts, :registry, @default_registry)),
         :ok <- ensure_registry(registry, auto_start?) do
      {:ok, %{registry: registry}}
    end
  end

  @impl true
  def broadcast(%{registry: registry}, topic, message) when is_binary(topic) do
    Registry.dispatch(registry, topic, fn entries ->
      Enum.each(entries, fn {pid, _meta} -> send(pid, message) end)
    end)

    :ok
  rescue
    error in [ArgumentError] ->
      {:error, runtime_error("local pubsub broadcast failed", error)}
  catch
    :exit, reason ->
      {:error, runtime_error("local pubsub broadcast failed", reason)}
  end

  @impl true
  def subscribe(%{registry: registry}, topic) when is_binary(topic) do
    case Registry.register(registry, topic, :ok) do
      {:ok, _owner} ->
        :ok

      {:error, {:already_registered, _owner}} ->
        :ok
    end
  rescue
    error in [ArgumentError] ->
      {:error, runtime_error("local pubsub subscribe failed", error)}
  catch
    :exit, reason ->
      {:error, runtime_error("local pubsub subscribe failed", reason)}
  end

  defp ensure_registry(registry, true) do
    if is_pid(registry) do
      if Process.alive?(registry) do
        :ok
      else
        {:error, config_error("local pubsub registry pid is not alive: #{inspect(registry)}")}
      end
    else
      do_ensure_registry_named(registry)
    end
  end

  defp ensure_registry(registry, false) do
    cond do
      is_pid(registry) and Process.alive?(registry) ->
        :ok

      is_atom(registry) and Process.whereis(registry) ->
        :ok

      is_pid(registry) ->
        {:error, config_error("local pubsub registry pid is not alive: #{inspect(registry)}")}

      true ->
        {:error, config_error("local pubsub registry is not started: #{inspect(registry)}")}
    end
  end

  defp do_ensure_registry_named(registry) do
    case Process.whereis(registry) do
      pid when is_pid(pid) ->
        :ok

      nil ->
        case Registry.start_link(keys: :duplicate, name: registry) do
          {:ok, _pid} ->
            :ok

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

          {:error, reason} ->
            {:error, runtime_error("failed to start local pubsub registry", reason)}
        end
    end
  end

  defp normalize_registry(registry) when is_atom(registry), do: {:ok, registry}

  defp normalize_registry(registry) when is_pid(registry) do
    if Process.alive?(registry) do
      {:ok, registry}
    else
      {:error, config_error("local adapter :registry pid is not alive: #{inspect(registry)}")}
    end
  end

  defp normalize_registry(other) do
    {:error,
     config_error("local adapter :registry must be an atom or pid, got: #{inspect(other)}")}
  end

  defp config_error(message) do
    Error.new(:config_invalid, :config, message)
  end

  defp runtime_error(message, cause) do
    Error.new(:unknown, :runtime, message, cause: cause)
  end
end