pulsar_elixir

defmodule Pulsar.Client do
  @moduledoc """
  A client represents an isolated Pulsar connection context.

  Each client maintains:
  - Separate broker connections
  - Independent consumer/producer registries
  - Isolated broker configuration

  ## Usage

  ### Single Client (Implicit)

  When using `Pulsar.start/1`, a default client is automatically created:

      # config.exs
      config :pulsar,
        host: "pulsar://localhost:6650",
        consumers: [...]

      # Uses implicit :default client
      {:ok, consumer} = Pulsar.start_consumer(topic, subscription, MyCallback)

  ### Multiple Clients (Explicit)

  You can start multiple clients in your supervision tree:

      children = [
        {Pulsar.Client, name: :analytics_client, host: "pulsar://analytics:6650"},
        {Pulsar.Client, name: :events_client, host: "pulsar://events:6650"}
      ]

      # Explicit client usage
      {:ok, consumer} = Pulsar.start_consumer(
        topic, subscription, MyCallback,
        client: :analytics_client
      )
  """

  use Supervisor

  @supported_broker_opts [
    :auth,
    :conn_timeout,
    :socket_opts
  ]

  ## Public API

  @doc """
  Starts a client with the given options.

  ## Options

  - `:name` - Required. The name of the client (atom)
  - `:host` - Required. Bootstrap broker URL
  - `:auth` - Optional. Authentication configuration
  - `:conn_timeout` - Optional. Connection timeout (default: 5000)
  - `:socket_opts` - Optional. Socket options
  """
  def start_link(opts) do
    name = Keyword.fetch!(opts, :name)
    bootstrap_host = Keyword.fetch!(opts, :host)

    case Supervisor.start_link(__MODULE__, opts, name: name) do
      {:ok, pid} = result ->
        # Start the bootstrap broker after supervisor is running
        case start_broker(bootstrap_host, client: name) do
          {:ok, _broker_pid} ->
            result

          {:error, reason} ->
            Supervisor.stop(pid)
            {:error, {:broker_startup_failed, reason}}
        end

      error ->
        error
    end
  end

  @impl true
  def init(opts) do
    client_name = Keyword.fetch!(opts, :name)
    broker_opts = build_broker_opts(opts)

    # Store broker opts in client state (passed to children via registry metadata)
    :persistent_term.put({__MODULE__, client_name, :broker_opts}, broker_opts)

    Pulsar.ProducerEpochStore.init(client_name)

    children = [
      {Registry, keys: :unique, name: broker_registry(client_name)},
      {Registry, keys: :unique, name: consumer_registry(client_name)},
      {Registry, keys: :unique, name: producer_registry(client_name)},
      {DynamicSupervisor, strategy: :one_for_one, name: broker_supervisor(client_name)},
      {DynamicSupervisor, strategy: :one_for_one, name: consumer_supervisor(client_name)},
      {DynamicSupervisor, strategy: :one_for_one, name: producer_supervisor(client_name)}
    ]

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

  ## Registry and Supervisor Name Helpers

  @doc false
  def broker_registry(client_name) do
    Module.concat([__MODULE__, client_name, BrokerRegistry])
  end

  @doc false
  def consumer_registry(client_name) do
    Module.concat([__MODULE__, client_name, ConsumerRegistry])
  end

  @doc false
  def producer_registry(client_name) do
    Module.concat([__MODULE__, client_name, ProducerRegistry])
  end

  @doc false
  def broker_supervisor(client_name) do
    Module.concat([__MODULE__, client_name, BrokerSupervisor])
  end

  @doc false
  def consumer_supervisor(client_name) do
    Module.concat([__MODULE__, client_name, ConsumerSupervisor])
  end

  @doc false
  def producer_supervisor(client_name) do
    Module.concat([__MODULE__, client_name, ProducerSupervisor])
  end

  @doc false
  def get_broker_opts(client_name) do
    :persistent_term.get({__MODULE__, client_name, :broker_opts}, [])
  end

  @doc """
  Returns a random broker process from the specified client's broker supervisor.

  Defaults to the `:default` client if no client is specified.

  This is useful for operations that need any broker from a client (e.g., service discovery).
  """
  @spec random_broker(atom()) :: pid() | nil
  def random_broker(client_name \\ :default) do
    broker_supervisor = broker_supervisor(client_name)

    case Supervisor.which_children(broker_supervisor) do
      [] ->
        nil

      children ->
        {_id, pid, _, _} = Enum.random(children)
        pid
    end
  end

  @doc """
  Starts a broker connection.

  If a broker for the given URL already exists, returns the existing broker.
  Otherwise, starts a new broker connection with the provided options.

  Returns `{:ok, broker_pid}` if successful, `{:error, reason}` otherwise.
  """
  @spec start_broker(String.t(), keyword()) :: {:ok, pid()} | {:error, term()}
  def start_broker(broker_url, opts \\ []) do
    client = Keyword.get(opts, :client, :default)
    broker_registry = broker_registry(client)
    broker_supervisor = broker_supervisor(client)

    case lookup_broker(broker_url, client: client) do
      {:ok, broker_pid} ->
        {:ok, broker_pid}

      {:error, :not_found} ->
        global_opts = get_broker_opts(client)
        merged_opts = Keyword.merge(global_opts, Keyword.delete(opts, :client))
        registry_opts = [{:name, {:via, Registry, {broker_registry, broker_url}}} | merged_opts]

        child_spec = %{
          id: broker_url,
          start: {Pulsar.Broker, :start_link, [broker_url, registry_opts]},
          restart: :permanent
        }

        case DynamicSupervisor.start_child(broker_supervisor, child_spec) do
          {:ok, broker_pid} ->
            {:ok, broker_pid}

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

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

  @doc """
  Looks up an existing broker connection by broker URL.

  Returns `{:ok, broker_pid}` if found, `{:error, :not_found}` otherwise.
  """
  @spec lookup_broker(String.t(), keyword()) :: {:ok, pid()} | {:error, :not_found}
  def lookup_broker(broker_url, opts \\ []) do
    client = Keyword.get(opts, :client, :default)
    broker_registry = broker_registry(client)

    case Registry.lookup(broker_registry, broker_url) do
      [{pid, _value}] -> {:ok, pid}
      [] -> {:error, :not_found}
    end
  end

  @doc """
  Stops a broker connection by broker URL.
  """
  @spec stop_broker(String.t(), keyword()) :: :ok | {:error, :not_found}
  def stop_broker(broker_url, opts \\ []) do
    case lookup_broker(broker_url, opts) do
      {:ok, broker_pid} ->
        Pulsar.Broker.stop(broker_pid)
        :ok

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

  @doc """
  Stops a client and all its resources gracefully.

  This stops all producers, consumers, brokers, and the client supervisor.

  ## Options

  - `:timeout` - Maximum time to wait for shutdown (default: 5000ms)

  ## Examples

      Pulsar.Client.stop(:my_client)

  """
  @spec stop(atom(), keyword()) :: :ok
  def stop(client_name, opts \\ []) when is_atom(client_name) do
    timeout = Keyword.get(opts, :timeout, 5000)

    try do
      Supervisor.stop(client_name, :normal, timeout)
    catch
      :exit, _ -> :ok
    end

    :persistent_term.erase({__MODULE__, client_name, :broker_opts})
    :ok
  end

  ## Private Functions

  defp build_broker_opts(opts) do
    app_opts =
      @supported_broker_opts
      |> Enum.map(fn key -> {key, Application.get_env(:pulsar, key)} end)
      |> Enum.reject(fn {_, v} -> is_nil(v) end)

    passed_opts =
      opts
      |> Keyword.take(@supported_broker_opts)
      |> Enum.reject(fn {_, v} -> is_nil(v) end)

    Keyword.merge(app_opts, passed_opts)
  end
end