docket

if Code.ensure_loaded?(Ecto.Adapters.SQL) and Code.ensure_loaded?(Postgrex) do
  defmodule Docket.Postgres.Notifier do
    @moduledoc """
    LISTEN fast path turning committed immediate-wake notifications into
    dispatcher polls.

    The Postgres RunStore announces every committed due wake with `pg_notify`
    on the `docket_wake` channel, carrying the schema prefix (empty string
    when unprefixed) as payload. The notifier holds one dedicated LISTEN
    connection outside the Repo pool, and each notification whose payload
    matches its own context prefix requests one immediate dispatcher poll.
    Notification bursts collapse inside
    the dispatcher.

    Polling remains the correctness mechanism. A lost notification, a dead
    listener, or an absent notifier only delays claiming until the
    dispatcher's next scheduled poll; omitting this child entirely is
    poll-only operation, where the dispatcher's poll interval alone bounds
    immediate-wake latency.

    The LISTEN connection derives its options from the configured Repo,
    reconnects on its own, and re-subscribes after reconnecting.
    Notifications sent while it is disconnected are dropped. `pg_notify`
    emission is ordinary SQL inside the writer's transaction and works
    through any pooler, but LISTEN requires a session-scoped connection:
    behind PgBouncer in transaction or statement pooling mode, point
    `:connection` at a direct or session-pooled endpoint, or run poll-only.
    """

    use GenServer

    alias Docket.Postgres.{Dispatcher, RunStore, Storage}

    @type option ::
            {:name, GenServer.name()}
            | {:context, Docket.Backend.ctx()}
            | {:dispatcher, GenServer.server()}
            | {:connection, keyword()}

    @repo_only_options [
      :adapter,
      :log,
      :loggers,
      :name,
      :otp_app,
      :pool,
      :pool_count,
      :pool_size,
      :priv,
      :stacktrace,
      :telemetry_prefix
    ]

    @spec start_link([option()]) :: GenServer.on_start()
    def start_link(opts) do
      GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name))
    end

    def child_spec(opts) do
      %{
        id: Keyword.get(opts, :name, __MODULE__),
        start: {__MODULE__, :start_link, [opts]}
      }
    end

    @impl true
    def init(opts) do
      {repo, prefix} = Storage.context!(Keyword.fetch!(opts, :context))
      dispatcher = Keyword.fetch!(opts, :dispatcher)
      connection = connection_opts(repo, Keyword.get(opts, :connection, []))

      {:ok, listener} = Postgrex.Notifications.start_link(connection)

      case Postgrex.Notifications.listen(listener, RunStore.wake_channel()) do
        {:ok, _reference} -> :ok
        {:eventually, _reference} -> :ok
      end

      {:ok, %{payload: prefix || "", dispatcher: dispatcher, listener: listener}}
    end

    @impl true
    def handle_info({:notification, _pid, _ref, channel, payload}, state) do
      matched? = channel == RunStore.wake_channel() and payload == state.payload

      if matched? do
        Dispatcher.request_poll(state.dispatcher)
      end

      :telemetry.execute(
        [:docket, :postgres, :notification],
        %{count: 1},
        %{result: if(matched?, do: :received, else: :ignored)}
      )

      {:noreply, state}
    end

    def handle_info(_message, state), do: {:noreply, state}

    defp connection_opts(repo, overrides) do
      repo.config()
      |> Keyword.drop(@repo_only_options)
      |> Keyword.put(:auto_reconnect, true)
      |> Keyword.merge(overrides)
    end
  end
end