one_flight

defmodule OneFlight do
  @moduledoc """
  Collapses concurrent duplicate calls into one supervised execution per key.

  `one_flight` is not a cache. It only shares work that is already in progress;
  after a flight completes, the result is discarded.

  ## Example

      iex> {:ok, _pid} = start_supervised({OneFlight, name: ExampleFlights})
      iex> OneFlight.run(:key, fn -> :value end, name: ExampleFlights)
      {:ok, :value}
  """

  @type key :: term()
  @type option ::
          {:name, GenServer.server()}
          | {:timeout, timeout()}
          | {:on_abandoned, :cancel | :complete}
          | {:scope, :node}

  @doc false
  def child_spec(opts) do
    id = Keyword.get(opts, :name, __MODULE__)
    %{id: id, start: {OneFlight.Supervisor, :start_link, [opts]}}
  end

  @doc false
  def start_link(opts), do: OneFlight.Supervisor.start_link(opts)

  @doc """
  Runs `fun` once for concurrent callers using the same key.
  """
  @spec run(key(), (-> term()), [option()]) :: {:ok, term()} | {:error, OneFlight.Error.t()}
  def run(key, fun, opts \\ []) when is_function(fun, 0) do
    with :ok <- validate_scope(Keyword.get(opts, :scope, :node)) do
      timeout = Keyword.get(opts, :timeout, 5_000)
      key |> async(fun, opts) |> await(timeout)
    end
  end

  @doc """
  Runs `fun` once and unwraps a successful result.

  Raises `OneFlight.Error` when the non-bang API would return an error tuple.
  """
  @spec run!(key(), (-> term()), [option()]) :: term()
  def run!(key, fun, opts \\ []) when is_function(fun, 0) do
    case run(key, fun, opts) do
      {:ok, result} -> result
      {:error, error} -> raise error
    end
  end

  @doc """
  Starts a flight and returns a handle that can be awaited.
  """
  @spec async(key(), (-> term()), [option()]) :: OneFlight.Flight.t()
  def async(key, fun, opts \\ []) when is_function(fun, 0) do
    name = Keyword.get(opts, :name, __MODULE__)
    coordinator = OneFlight.Supervisor.coordinator_name(name)
    {:wait, ref} = OneFlight.Coordinator.run(coordinator, key, fun, opts)
    %OneFlight.Flight{ref: ref, coordinator: coordinator}
  end

  @doc """
  Waits for an async flight result.
  """
  @spec await(OneFlight.Flight.t(), timeout()) :: {:ok, term()} | {:error, OneFlight.Error.t()}
  def await(%OneFlight.Flight{ref: ref, coordinator: coordinator}, timeout \\ 5_000) do
    receive do
      {^ref, result} ->
        result
    after
      timeout ->
        OneFlight.Coordinator.waiter_timed_out(coordinator, ref)
        {:error, OneFlight.Error.timeout()}
    end
  end

  @doc """
  Detaches `key` so the next caller starts a fresh flight.
  """
  @spec forget(key(), [option()]) :: :ok
  def forget(key, opts \\ []) do
    name = Keyword.get(opts, :name, __MODULE__)
    OneFlight.Coordinator.forget(OneFlight.Supervisor.coordinator_name(name), key)
  end

  @doc """
  Returns local in-flight stats for `key`.
  """
  @spec stats(key(), [option()]) :: %{in_flight?: boolean(), waiters_count: non_neg_integer()}
  def stats(key, opts \\ []) do
    name = Keyword.get(opts, :name, __MODULE__)
    OneFlight.Coordinator.stats(OneFlight.Supervisor.coordinator_name(name), key)
  end

  defp validate_scope(:node), do: :ok

  defp validate_scope(:cluster) do
    {:error,
     OneFlight.Error.invalid_option("scope: :cluster is not supported in one_flight v0.1")}
  end
end