defmodule AudioProxy.RenderCoordinator do
  @moduledoc """
  One render per in-flight cache key, shared by everyone who asks for it.

  API doc §5: concurrent requests for the same not-yet-cached variant must not
  each spawn an encoder. This is the single-flight in front of
  `AudioProxy.Ffmpeg.Render` — a `Registry` of unique cache keys, a coordinator
  process per key, and a `DynamicSupervisor` holding them.

  ## Subscribing

  `subscribe/2` either starts the render or attaches to the running one:

      {:ok, :miss, render, []}          # this caller started it
      {:ok, :coalesced, render, chunks} # it was already running; `chunks` is the catch-up

  `render` is the coordinator's pid and stands where `AudioProxy.Ffmpeg.Render`'s
  did. It broadcasts the pipeline's own contract, with itself as the handle:

    * `{:chunk, render, binary}`
    * `{:done, render, %{exit_status: 0}}`
    * `{:error, render, failure}`

  So a consumer written against the pipeline works here unchanged, except for
  the two calls that name a handle: `Render.ack/2` has no counterpart (see
  *Backlog* below) and `Render.cancel/1` becomes `unsubscribe/1`, because a
  subscriber leaving must not cancel a render other subscribers are still
  reading.

  ## The slot comes first

  A coordinator does not spawn its render until `AudioProxy.Semaphore` grants it
  a slot, and releases that slot from `terminate/2` — so the cap counts renders
  rather than requests, and a variant twenty clients are waiting on costs one.

  It asks with `Semaphore.request/2` rather than the blocking `acquire/1`,
  because this process must keep answering joins while it waits: the requests
  coalescing onto a queued render are exactly the ones that should not each take
  a slot of their own. So a coordinator has one phase more than the render does
  — `:queued`, before `:rendering` — and a subscriber cannot tell them apart,
  which is the point. It is waiting for bytes either way.

  The slot is asked for in `init/1`, which is what makes a full queue a *start*
  failure rather than a render failure: `subscribe/2` answers
  `{:error, {:queue_full, retry_after}}` and no coordinator is left registered
  under that key, so the next request asks the semaphore again instead of
  joining something that is never going to render. `AudioProxy.ErrorJSON`
  already renders that tuple as §5's 429, `Retry-After` and all.

  ## The start race

  Starting and joining are the same call. `DynamicSupervisor.start_child/2`
  either succeeds — this caller is the MISS, and its pid is already in the
  subscriber list before `init/1` returns, so no chunk can be produced before
  it is listening — or answers `{:error, {:already_started, pid}}`, which is
  the COALESCED path and a join.

  The join is a `call` that returns the backlog *and* registers the subscriber
  in one callback. That atomicity is the whole seam: a joiner cannot be handed
  a backlog and then miss the chunk that was broadcast while it was being
  handed one, and cannot be registered first and receive a chunk it will also
  find in its backlog.

  Losing the race to a coordinator that is stopping — a finished render past
  its linger, a failed one that has already unregistered — is a caught exit and
  a retry, which starts a fresh render.

  ## Backlog, and the memory bound

  Everything the render produces is retained, in order, so a subscriber that
  joins at any point receives the whole stream. Retaining it is not a cost the
  coalescing imposes: the variant-cache slice tees the same bytes to storage.

  Because the bytes are held anyway, the pipeline is acknowledged the moment a
  chunk arrives, and the pipeline's own high-water mark stops being the bound.
  What bounds this instead is `AP_MAX_VARIANT_BYTES`: a render whose output
  passes it fails for every subscriber rather than growing without limit. It is
  a ceiling on the *variant*, separate from `AP_MAX_SRC_BYTES`'s ceiling on the
  source, so a deployment can accept two-hour masters and still bound one
  render's retention to a preview's worth of bytes. It defaults to the
  effective `AP_MAX_SRC_BYTES`, so a deployment that sets neither is bounded
  exactly where it was before the two were separated.

  The breach is only detectable once the response has committed to `200` and
  begun streaming, so it is a failed request rather than a `413` — the source
  ceiling stays the only one enforced before a render starts. That is the
  reason the escalation is written down: the named-pipe (FIFO) pattern in
  `AudioProxy.Ffmpeg.Render`'s moduledoc spools instead of retaining, and
  nothing in the contract above changes when it lands.

  Raising the ceiling does not buy capacity. It bounds one render while the
  bill is `AP_MAX_CONCURRENCY × backlog`, so raising it licenses every slot to
  reach the larger figure — which turns one killed render into an exhausted
  container. `docs/capacity.md` has the arithmetic.

  ## Teardown

  Three ways this ends, and they differ in what happens to the key:

    * **Done** — the completion is broadcast, then the coordinator lingers
      briefly, still registered, so a request that passed its cache check a
      moment too late still gets the finished bytes instead of re-rendering.
    * **Failure** — broadcast to every current subscriber, then the key is
      unregistered *immediately*, so the next request retries rather than
      attaching to a corpse.
    * **Last subscriber gone** — nobody is listening, so the subprocess is
      cancelled. `unsubscribe/1` returns only once that has happened, which is
      what makes it usable as a barrier the way `Render.cancel/1` is.

  One subscriber dying never affects the others: it is removed and the render
  continues.

  ## The write-back tee is a subscriber

  When a variant store is configured and the spec carries `:metadata`, the
  coordinator starts an `AudioProxy.VariantStore.Tee` and registers it like
  any other subscriber — before the first chunk can exist, so the store
  receives the whole stream. That one fact is the disconnect policy: with a
  store, the last *client* leaving still leaves the tee counted, so the render
  completes into the store and the next request is a HIT; without one, the
  subscriber count reaches zero and the render is cancelled as before. A tee
  that dies is forgotten like any subscriber — and if no client remains
  either, the render is cancelled, because nothing is left that could profit
  from it.

  It starts with the *render*, not with the coordinator, which is only
  visible while a slot is being waited for. A queued coordinator has no tee,
  so a queued render every client has abandoned stops rather than holding its
  place to encode for the cache alone — which would spend a slot, under
  exactly the contention that made it queue, on a render no client wants.
  """

  use GenServer

  require Logger

  alias AudioProxy.{Config, Semaphore}
  alias AudioProxy.Ffmpeg.{Render, RenderSupervisor}
  alias AudioProxy.VariantStore
  alias AudioProxy.VariantStore.Tee

  @registry __MODULE__.Registry
  @supervisor __MODULE__.Supervisor

  # How long a finished coordinator stays registered. Long enough to catch a
  # request that was already in flight when the render completed, short enough
  # that a hot key does not keep its bytes resident. A straggler that misses it
  # starts a fresh render — wasteful, never wrong.
  @linger 1_000

  # Joining is a message to a live process that does no I/O in the callback, so
  # this only ever expires on a coordinator that is wedged.
  @join_timeout 5_000

  # Leaving can block on the pipeline's kill discipline, which is `Render`'s
  # own worst case plus margin. A timeout here would hand back `:ok` for a
  # subprocess still being killed, so it is set above that worst case rather
  # than at something request-shaped.
  @unsubscribe_timeout 15_000

  # Added to what `terminate/2` can block on for this process' shutdown budget,
  # so it can always finish the work it started. See `child_spec/1`.
  @shutdown_margin 2_000

  # Retries of the whole start-or-join loop. Each one either starts a render or
  # joins a live coordinator; only a coordinator stopping in the window between
  # the two costs an attempt, so exhausting five means something is wrong that
  # retrying will not fix.
  @attempts 5

  @typedoc "The coordinator, which is also the handle in the messages above."
  @type t :: pid()

  @typedoc "Whether this subscriber started the render or attached to it."
  @type status :: :miss | :coalesced

  @typedoc """
  How to render, when this subscriber turns out to be the one starting it.

  The options of `AudioProxy.Ffmpeg.Render.start_link/1` less `:consumer`,
  which is the coordinator itself — plus `:metadata`
  (`t:AudioProxy.VariantStore.metadata/0`), which never reaches the pipeline:
  it is what the write-back tee stores alongside the bytes, and without it no
  tee is started.
  """
  @type render_spec :: keyword()

  @typedoc """
  What a subscriber receives.

  The pipeline's own three, plus `{:rendering, _}` — sent only to subscribers
  that were attached while this coordinator was waiting for a render slot, and
  meaning "bytes are now possible". A consumer that measures how long the
  render has been silent needs it, because until then there was no render. One
  that does not may ignore it.
  """
  @type message ::
          {:rendering, t()}
          | {:chunk, t(), binary()}
          | {:done, t(), %{exit_status: 0}}
          | {:error, t(), Render.failure()}

  defstruct [
    :key,
    :spec,
    :metadata,
    :render,
    :render_monitor,
    :info,
    :linger_timer,
    phase: :rendering,
    subscribers: %{},
    backlog: [],
    bytes: 0
  ]

  @doc """
  The registry and supervisor this module needs, for the application tree.

  Start them after `AudioProxy.Ffmpeg.RenderSupervisor`: a coordinator spawns a
  render in `init/1`, and shutdown then tears the coordinators down first.
  """
  @spec children() :: [Supervisor.child_spec() | {module(), term()}]
  def children do
    [
      {Registry, keys: :unique, name: @registry},
      {DynamicSupervisor, strategy: :one_for_one, name: @supervisor}
    ]
  end

  @doc """
  How many renders are in flight right now.

  One per registered cache key, which is the same thing `AP_MAX_CONCURRENCY`
  counts: the twenty requests coalesced onto one encode are one render here.
  `AudioProxy.Metrics` samples it per scrape rather than maintaining a gauge
  from the render events, because a registry entry cannot outlive the process
  that holds it — an abnormal exit corrects this number for free, where an
  event-driven counter would be permanently one too high.

  Briefly includes a coordinator that has finished and not yet unregistered,
  which is the same window that lets a late request coalesce onto its backlog.
  """
  @spec in_flight() :: non_neg_integer()
  def in_flight, do: Registry.count(@registry)

  @doc """
  Subscribes the calling process to the render for `cache_key`, starting it
  from `spec` if it is not already running.

  Returns `{:ok, status, render, backlog}`. `backlog` is the bytes produced
  before this subscriber attached, in order, and is always empty for a `:miss`;
  a consumer must deliver it before the chunks that follow.

  Errors are the pipeline's own — `{:error, :ffmpeg_not_found}` and friends —
  plus `{:error, {:queue_full, retry_after}}` when `AudioProxy.Semaphore` had
  neither a slot nor room to wait, and `{:error, :coordinator_unavailable}` if
  the start-or-join loop could not settle, which means something is repeatedly
  killing coordinators.
  """
  @spec subscribe(String.t(), render_spec()) ::
          {:ok, status(), t(), [binary()]} | {:error, term()}
  def subscribe(cache_key, spec) when is_binary(cache_key) and is_list(spec) do
    start_or_join(cache_key, spec, @attempts)
  end

  @doc """
  Detaches the calling process from `render`.

  The render continues for whoever is left. If nobody is, it is cancelled and
  its subprocess killed before this returns — so, like
  `AudioProxy.Ffmpeg.Render.cancel/1`, this is a barrier and not a request.

  Detaching from a coordinator that has already stopped is `:ok`: that is the
  outcome asked for.
  """
  @spec unsubscribe(t()) :: :ok
  def unsubscribe(render) do
    GenServer.call(render, :unsubscribe, @unsubscribe_timeout)
  catch
    :exit, _reason -> :ok
  end

  @doc false
  def child_spec(opts) do
    %{
      id: __MODULE__,
      start: {__MODULE__, :start_link, [opts]},
      # Never restarted, for the same reason a render is not: the byte stream
      # is gone and every subscriber has already been told.
      restart: :temporary,
      # Derived, not chosen, and derived from *both* calls `terminate/2` makes.
      # A budget below `Render.cancel/1`'s own timeout means the supervisor
      # brutal-kills this process mid-cancel on exactly the renders that need
      # cancelling most — the ones whose subprocess is ignoring SIGTERM. The
      # subprocess still dies (the pipeline monitors its consumer), but the
      # kill would be racing the guarantee instead of implementing it. The
      # release that follows the cancel has a bound of its own, and leaving it
      # out would put the sum back over the budget in the one case where both
      # are slow.
      shutdown: Render.cancel_timeout() + Semaphore.release_timeout() + @shutdown_margin
    }
  end

  @doc false
  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: via(Keyword.fetch!(opts, :key)))
  end

  defp via(key), do: {:via, Registry, {@registry, key}}

  defp start_or_join(_key, _spec, 0), do: {:error, :coordinator_unavailable}

  defp start_or_join(key, spec, attempts) do
    child = {__MODULE__, key: key, spec: spec, subscriber: self()}

    case DynamicSupervisor.start_child(@supervisor, child) do
      {:ok, pid} ->
        {:ok, :miss, pid, []}

      {:error, {:already_started, pid}} ->
        case join(pid) do
          {:ok, backlog} -> {:ok, :coalesced, pid, backlog}
          :gone -> start_or_join(key, spec, attempts - 1)
        end

      # `init/1` refuses a render it could not start as a `:shutdown` stop, so
      # a missing ffmpeg is an error tuple rather than a crash report on every
      # request. Unwrapped here, so callers see the pipeline's own reason.
      {:error, {:shutdown, reason}} ->
        {:error, reason}

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

  defp join(pid) do
    GenServer.call(pid, {:join, self()}, @join_timeout)
  catch
    # Stopping between the registry lookup and this call. Retrying starts a
    # fresh render, which is correct — the key is on its way out.
    :exit, _reason -> :gone
  end

  ## Server

  @impl true
  def init(opts) do
    # So that the supervisor's shutdown is a stop like any other and reaches
    # `terminate/2`, where the subprocess is killed.
    Process.flag(:trap_exit, true)

    key = Keyword.fetch!(opts, :key)
    {metadata, spec} = Keyword.pop(Keyword.fetch!(opts, :spec), :metadata)
    subscriber = Keyword.fetch!(opts, :subscriber)

    state = %__MODULE__{
      key: key,
      spec: spec,
      metadata: metadata,
      # Registered before the first chunk can exist, so the subscriber that
      # started this render needs no backlog. The tee is registered the same
      # way, but not here — see `spawn_render/1`.
      subscribers: %{subscriber => Process.monitor(subscriber)}
    }

    # Asked for here rather than after `init/1` returns, so that a full queue is
    # a `start_child` error the requesting process gets back directly — a 429
    # before a coordinator exists, rather than a coordinator whose only act is
    # to broadcast a failure. `request/2` answers immediately whichever way it
    # goes; only the *waiting* would block, and that is what `:queued` is.
    #
    # Before the tee, too: a coordinator that cannot get a slot stops without
    # ever running `terminate/2`, and it should not have opened a staging file
    # for a render that will not happen.
    case Semaphore.request() do
      :granted ->
        start_render(state)

      :queued ->
        {:ok, %{state | phase: :queued}}

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

  # Started from `init/1`, so a start failure can still be a `{:stop,
  # {:shutdown, _}}` that `start_or_join/3` unwraps into the caller's own
  # `{:error, reason}`: no ffmpeg on `PATH` is a 500 with the reason in the log
  # and no crash report per request, exactly as it was before slots existed.
  defp start_render(state) do
    case spawn_render(state) do
      {:ok, state} -> {:ok, state}
      {:error, reason} -> {:stop, {:shutdown, reason}}
    end
  end

  # Started from a grant that arrived after `init/1` returned. The caller has
  # long since been handed a coordinator, so the only way left to report a
  # start failure is the failure broadcast every other failure uses.
  defp start_granted_render(state) do
    case spawn_render(state) do
      {:ok, state} ->
        {:noreply, state}

      {:error, reason} ->
        fail(state, %{
          class: :render_failed,
          exit_status: nil,
          detail: "could not start render: #{inspect(reason)}"
        })
    end
  end

  # Announced on every transition into `:rendering`, including the one that
  # happens inside `init/1` where there was no wait at all. A consumer's budget
  # for "the render has said nothing" cannot start when it subscribed, because
  # between subscribing and a byte being *possible* there may have been a queue;
  # this is the moment it becomes possible, and it has to be reported the same
  # way whether the queue was empty or not — a consumer that only heard about
  # the slow case would have to guess about the fast one.
  #
  # The tee is started here rather than in `init/1`, which is where the slot and
  # the write-back meet. Registering it before the render is spawned keeps the
  # guarantee it was given — no chunk can exist yet, so it needs no backlog —
  # while keeping its lifetime the render's rather than this process'. Starting
  # it while queued would open a staging file for a render that has not begun,
  # against a deadline of its own that assumes one has; worse, it would count as
  # a subscriber, so a queued render every client had abandoned would hold its
  # place, take a slot, and encode for the cache alone. Nothing designed that,
  # and under contention it is a slot taken from a request that has a client.
  defp spawn_render(state) do
    state = %{state | subscribers: maybe_tee(state.subscribers, state.key, state.metadata)}

    case RenderSupervisor.start_render(Keyword.put(state.spec, :consumer, self())) do
      {:ok, render} ->
        state = %{
          state
          | phase: :rendering,
            render: render,
            render_monitor: Process.monitor(render)
        }

        broadcast(state, {:rendering, self()})

        {:ok, state}

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

  # The tee exists exactly when there is somewhere to write and something to
  # write it with. Failing to start one is logged, not fatal: the render and
  # its clients owe nothing to the cache.
  defp maybe_tee(subscribers, key, metadata) do
    if metadata != nil and VariantStore.configured?() do
      case Tee.start(self(), key, metadata) do
        {:ok, tee} ->
          Map.put(subscribers, tee, Process.monitor(tee))

        {:error, reason} ->
          Logger.warning("could not start variant store tee: #{inspect(reason)}")
          subscribers
      end
    else
      subscribers
    end
  end

  # `:queued` and `:rendering` are one thing from a subscriber's side: bytes are
  # coming, and none have arrived that this joiner is missing. They differ in
  # one respect only, and it is not about the bytes — a joiner that arrives
  # while the render is already under way never queued, so it is told so here
  # rather than waiting for the announcement it has already missed. Sent before
  # the reply, so it is in the joiner's mailbox by the time it starts reading.
  @impl true
  def handle_call({:join, pid}, _from, %__MODULE__{phase: :rendering} = state) do
    send(pid, {:rendering, self()})

    {:reply, {:ok, backlog(state)}, monitor_subscriber(state, pid)}
  end

  def handle_call({:join, pid}, _from, %__MODULE__{phase: :queued} = state) do
    {:reply, {:ok, backlog(state)}, monitor_subscriber(state, pid)}
  end

  # The render is over and every byte is in the backlog, so this joiner is
  # complete the moment it has it. The terminal message is sent before the
  # reply and therefore lands in the caller's mailbox first; `GenServer.call/3`
  # selects on its own ref, so the joiner still gets the backlog first and
  # finds `{:done, _, _}` waiting when it starts consuming.
  def handle_call({:join, pid}, _from, %__MODULE__{phase: :completed} = state) do
    send(pid, {:done, self(), state.info})

    {:reply, {:ok, backlog(state)}, state}
  end

  def handle_call(:unsubscribe, {pid, _tag}, state) do
    case forget_subscriber(state, pid) do
      %__MODULE__{phase: phase, subscribers: subscribers} = state
      when phase in [:queued, :rendering] and map_size(subscribers) == 0 ->
        # Nobody is reading. Replying from `terminate/2`'s far side is what
        # makes this a barrier: the subprocess is gone before the caller runs.
        {:stop, :normal, :ok, state}

      state ->
        {:reply, :ok, state}
    end
  end

  @impl true
  def handle_info({:chunk, render, data}, %__MODULE__{render: render} = state) do
    # Acknowledged immediately: the bytes are retained here regardless, so the
    # pipeline's high-water mark would only stall a stream this process has
    # already paid the memory for. The backlog cap is the bound instead.
    Render.ack(render, byte_size(data))

    case retain(state, data) do
      {:ok, state} ->
        broadcast(state, {:chunk, self(), data})
        {:noreply, state}

      {:error, failure} ->
        fail(state, failure)
    end
  end

  def handle_info({:done, render, info}, %__MODULE__{render: render} = state) do
    broadcast(state, {:done, self(), info})

    Process.demonitor(state.render_monitor, [:flush])

    # Here, not in `terminate/2`. A slot caps *encoders*, and this one's is
    # gone: `{:done, _, _}` arrives after the subprocess has exited and every
    # byte it wrote has been forwarded. What the linger below keeps alive is a
    # buffer being served from memory, which costs no CPU and must not cost a
    # slot — holding one across it would shrink effective concurrency by the
    # linger over the render's own duration, which for the preview-sized
    # renders v1 targets is most of it. `terminate/2` still releases, for the
    # paths that never get here; the second call is a no-op.
    Semaphore.release()

    {:noreply,
     %{
       state
       | phase: :completed,
         info: info,
         render: nil,
         render_monitor: nil,
         linger_timer: Process.send_after(self(), :linger_expired, @linger)
     }}
  end

  def handle_info({:error, render, failure}, %__MODULE__{render: render} = state) do
    fail(state, failure)
  end

  # The pipeline died without saying anything — it cannot report its own crash.
  def handle_info(
        {:DOWN, monitor, :process, _pid, reason},
        %__MODULE__{render_monitor: monitor} = state
      ) do
    fail(state, %{
      class: :render_failed,
      exit_status: nil,
      detail: "render died without reporting: #{inspect(reason)}"
    })
  end

  def handle_info({:DOWN, _monitor, :process, pid, _reason}, state) do
    case forget_subscriber(state, pid) do
      %__MODULE__{phase: phase, subscribers: subscribers} = state
      when phase in [:queued, :rendering] and map_size(subscribers) == 0 ->
        {:stop, :normal, state}

      state ->
        {:noreply, state}
    end
  end

  # The slot came free. Nothing about this is visible to a subscriber: it has
  # been waiting for bytes, and still is.
  def handle_info({Semaphore, :granted}, %__MODULE__{phase: :queued} = state) do
    start_granted_render(state)
  end

  def handle_info(:linger_expired, state), do: {:stop, :normal, state}

  # The only link is the supervisor's, so this is the shutdown path. Stopping
  # rather than ignoring it is what lets `terminate/2` kill the subprocess.
  def handle_info({:EXIT, _from, reason}, state), do: {:stop, reason, state}

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

  @impl true
  def terminate(_reason, %__MODULE__{} = state) do
    # `cancel/1` returns once the subprocess is gone and tolerates a render that
    # already finished, so this is safe on every stop path.
    if state.render, do: Render.cancel(state.render)

    # After the cancel, not before: releasing first would hand the slot to a
    # waiter while this render's ffmpeg was still being killed, which is the
    # cap being exceeded by exactly the margin the kill discipline takes.
    #
    # Idempotent, and it covers a `:queued` coordinator that never held a slot
    # as well as one that did. The semaphore's own monitor is the backstop for
    # the stop paths that never reach here.
    Semaphore.release()
    :ok
  end

  ## Subscribers

  defp monitor_subscriber(state, pid) do
    if Map.has_key?(state.subscribers, pid) do
      state
    else
      %{state | subscribers: Map.put(state.subscribers, pid, Process.monitor(pid))}
    end
  end

  defp forget_subscriber(state, pid) do
    case Map.pop(state.subscribers, pid) do
      {nil, _subscribers} ->
        state

      {monitor, subscribers} ->
        Process.demonitor(monitor, [:flush])
        %{state | subscribers: subscribers}
    end
  end

  defp broadcast(state, message) do
    Enum.each(state.subscribers, fn {pid, _monitor} -> send(pid, message) end)
  end

  ## Backlog

  defp backlog(state), do: Enum.reverse(state.backlog)

  defp retain(state, data) do
    bytes = state.bytes + byte_size(data)
    limit = Config.get(:max_variant_bytes)

    if bytes > limit do
      {:error,
       %{
         class: :render_failed,
         exit_status: nil,
         detail: "render output exceeded the #{limit}-byte retention cap (AP_MAX_VARIANT_BYTES)"
       }}
    else
      {:ok, %{state | backlog: [data | state.backlog], bytes: bytes}}
    end
  end

  ## Failure

  # Told once to everyone currently attached, then the key goes immediately so
  # the next request renders again instead of joining something that is gone.
  defp fail(state, failure) do
    broadcast(state, {:error, self(), failure})
    Registry.unregister(@registry, state.key)

    {:stop, :normal, state}
  end
end