defmodule DustEcto.Phoenix.Broadcaster do
  @moduledoc """
  One broadcaster process per `{schema, pubsub, topic}` triple. Holds
  a single `DustEcto.Repo.subscribe/2` registration whose callback
  broadcasts every event through Phoenix.PubSub on `topic`.

  Lifecycle:
    * Started lazily by `DustEcto.Phoenix.subscribe_to_pubsub/3`.
    * Registered in `DustEcto.Phoenix.Registry` keyed by
      `{schema, pubsub, topic}` so concurrent callers reuse the same
      broadcaster.
    * Stays alive for the application lifetime. The cost is one
      process and one entry in the SDK callback registry — both
      negligible. Cleanup is opt-in (caller can `unsubscribe/3`).

  The broadcast message shape is `{:dust_event, event}` where `event`
  is whatever `DustEcto.Repo.subscribe/2` would deliver:
  `{:upserted, struct}` or `{:deleted, slug}`.
  """

  use GenServer

  require Logger

  @doc false
  def start_link({schema, pubsub, topic} = key)
      when is_atom(schema) and is_atom(pubsub) and is_binary(topic) do
    GenServer.start_link(__MODULE__, key, name: via(key))
  end

  @doc false
  def via({schema, pubsub, topic}) do
    {:via, Registry, {DustEcto.Phoenix.Registry, {schema, pubsub, topic}}}
  end

  @impl true
  def init({schema, pubsub, topic}) do
    case DustEcto.Repo.subscribe(schema, broadcast_callback(pubsub, topic)) do
      {:ok, ref} ->
        {:ok, %{schema: schema, pubsub: pubsub, topic: topic, ref: ref}}

      {:error, %DustEcto.Error{} = err} ->
        {:stop, {:dust_subscribe_failed, err}}
    end
  end

  @impl true
  def terminate(_reason, %{ref: ref}) when is_reference(ref) do
    DustEcto.Repo.unsubscribe(ref)
    :ok
  end

  def terminate(_, _), do: :ok

  # Returns a 1-arity function suitable for DustEcto.Repo.subscribe/2.
  # We deliberately use apply/3 rather than a direct Phoenix.PubSub
  # call so this module compiles even when phoenix_pubsub isn't a
  # transitive dep — the call only ever runs if the broadcaster
  # started, which in turn requires phoenix_pubsub at runtime.
  defp broadcast_callback(pubsub, topic) do
    fn event ->
      apply(Phoenix.PubSub, :broadcast, [pubsub, topic, {:dust_event, event}])
    end
  end
end