docket

if Code.ensure_loaded?(Ecto.Adapters.SQL) and Code.ensure_loaded?(Postgrex) do
  defmodule Docket.Postgres.AdmissionPhase do
    @moduledoc """
    Coordinates the admission preference shared by one PostgreSQL backend instance.

    The legacy claim policy gives single-run claims (`demand == 1`) a choice between
    ready work and expired leases. To prevent one class from being preferred
    indefinitely, successful single-run attempts alternate their first choice:
    `:ready`, then `:expired`, then `:ready` again. Here, successful means any
    `{:ok, _}` result, including an empty or poison-only batch; it does not mean that
    a lease was returned. A query may also fall through to the non-preferred class,
    so this alternates the first choice rather than guaranteeing alternating result
    classes.

    Every admission receives the current preference. The legacy policy ignores it
    for multi-run ordering because that query already makes progress across both
    classes, and multi-run attempts do not advance the phase. Other policy
    implementations remain free to interpret the supplied value.

    This process owns exactly three pieces of transient, instance-local state:

      * `preference` — the next first-choice class, initially `:ready`;
      * `owner` — the current attempt's token, demand, and caller monitor;
      * `queue` — callers waiting in FIFO order for the current owner to finish.

    `run/3` executes the callback in the calling process while holding exclusive
    ownership, supplies it with the current preference, and advances the preference
    only after a successful single-run attempt. Error results, exceptions, caller
    exits, and multi-run attempts release ownership without advancing it. Caller
    monitoring ensures a process that exits cannot strand the queue.

    Checkout and completion intentionally have no timeout: claim serialization must
    outlive ordinary call timeouts. Consequently, a live callback that never returns
    blocks the queue, and a callback must not call `run/3` recursively for the same
    phase. The owning claim path is responsible for bounding its database operation;
    process death is the phase's recovery boundary.

    Production dispatcher mode and synchronous testing/drain mode both use this
    abstraction, with one phase process for the active mode of each backend instance.
    Keeping the phase here, rather than in a caller or per-call options, gives that
    mode one serialized ordering history. In production it belongs to the Runner's
    `:one_for_all` accounting unit alongside the dispatcher and vehicle supervisor;
    restarting that unit resets the phase to `:ready`. In testing mode it is the
    backend's standalone child. The state is scheduling coordination only and is
    deliberately not durable.
    """

    use GenServer

    def start_link(opts) do
      GenServer.start_link(__MODULE__, :ready, Keyword.take(opts, [:name]))
    end

    def run(server, demand, fun) when is_integer(demand) and demand > 0 and is_function(fun, 1) do
      {token, preference} = GenServer.call(server, {:checkout, demand}, :infinity)

      try do
        result = fun.(preference)
        :ok = GenServer.call(server, {:complete, token, result}, :infinity)
        result
      catch
        kind, reason ->
          GenServer.cast(server, {:abandon, token})
          :erlang.raise(kind, reason, __STACKTRACE__)
      end
    end

    @impl true
    def init(preference), do: {:ok, %{preference: preference, owner: nil, queue: :queue.new()}}

    @impl true
    def handle_call({:checkout, demand}, from, %{owner: nil} = state) do
      {reply, state} = take_ownership(state, from, demand)
      {:reply, reply, state}
    end

    def handle_call({:checkout, demand}, from, state) do
      {:noreply, %{state | queue: :queue.in({from, demand}, state.queue)}}
    end

    def handle_call({:complete, token, result}, _from, %{owner: %{token: token}} = state) do
      state = maybe_advance(state, result)
      {:reply, :ok, release(state)}
    end

    def handle_call({:complete, _token, _result}, _from, state), do: {:reply, :stale, state}

    @impl true
    def handle_cast({:abandon, token}, %{owner: %{token: token}} = state),
      do: {:noreply, release(state)}

    def handle_cast({:abandon, _token}, state), do: {:noreply, state}

    @impl true
    def handle_info(
          {:DOWN, monitor, :process, _pid, _reason},
          %{owner: %{monitor: monitor}} = state
        ),
        do: {:noreply, release(state)}

    def handle_info({:DOWN, _monitor, :process, _pid, _reason}, state), do: {:noreply, state}

    defp take_ownership(state, {pid, _tag}, demand) do
      token = make_ref()
      monitor = Process.monitor(pid)
      reply = {token, state.preference}
      {reply, %{state | owner: %{token: token, monitor: monitor, demand: demand}}}
    end

    defp release(%{owner: %{monitor: monitor}} = state) do
      Process.demonitor(monitor, [:flush])
      state = %{state | owner: nil}

      case :queue.out(state.queue) do
        {{:value, {from, demand}}, queue} ->
          {reply, state} = take_ownership(%{state | queue: queue}, from, demand)
          GenServer.reply(from, reply)
          state

        {:empty, _queue} ->
          state
      end
    end

    defp maybe_advance(%{owner: %{demand: 1}, preference: :ready} = state, {:ok, _}),
      do: %{state | preference: :expired}

    defp maybe_advance(%{owner: %{demand: 1}, preference: :expired} = state, {:ok, _}),
      do: %{state | preference: :ready}

    defp maybe_advance(state, _result), do: state
  end
end