scriba

defmodule Scriba do
  @moduledoc """
  Public API surface for Scriba projections.

  ## The five-line API

      defmodule MyApp.Projections.Orders do
        use Scriba.Projection,
          name: "orders",
          source: {Scriba.Source.Commanded, application: MyApp.CommandedApp},
          target: {Scriba.Target.Ecto, repo: MyApp.Repo},
          parallelism: 16

        def handle(_event, _meta), do: :skip
      end

      {:ok, _pid} = Scriba.start_projection(MyApp.Projections.Orders)
      Scriba.info(MyApp.Projections.Orders)
      Scriba.pause(MyApp.Projections.Orders)
      Scriba.resume(MyApp.Projections.Orders)
      Scriba.stop(MyApp.Projections.Orders)
      Scriba.list()

  Every public function accepts either:

    * **Module form** — `Scriba.pause(MyApp.Projections.Orders)`. The
      module's `__scriba_config__/0` (generated by
      `use Scriba.Projection`) provides name and version. Refactor-
      friendly: rename the module, the call site renames with it.
    * **String form** — `Scriba.pause("orders", 1)`. Operator-friendly
      when you only know the projection's name. Defaults to version 1
      for the arity-1 calls.

  `Scriba.list/0` enumerates running projections via `Scriba.Registry`.
  """

  alias Scriba.Info
  alias Scriba.Projection.Coordinator

  @stream_position_truncation_limit 1000

  ## Lifecycle — start / list

  @doc """
  Starts a projection from its module's compile-time config.

  Reads the projection's `__scriba_config__/0` map (generated by
  `use Scriba.Projection`) and starts a `Scriba.Projection.Supervisor`
  under `Scriba.Projections.Supervisor`. Returns the per-projection
  supervisor pid.

  Returns immediately (does not wait for the Coordinator to transition
  from `:initializing` to `:running`). Wait for the
  `[:scriba, :projection, :started]` telemetry event or poll
  `Scriba.info/1` if you need a confirmation point.

  Returns `{:error, :already_started}` if a projection with the same
  `(name, version)` is already running.
  """
  @spec start_projection(module()) ::
          {:ok, pid()} | {:error, :already_started | term()}
  def start_projection(module) when is_atom(module) do
    start_projection(module, [])
  end

  @doc """
  Same as `start_projection/1`, but merges `overrides` into the module's
  compile-time config before starting.

  Useful for runtime tuning of `:parallelism`, `:batch_size`, `:retry`,
  etc. per-environment without recompiling the module.

  **Rejects `:name` and `:version` overrides** with `ArgumentError`.
  Identity is compile-time: passing a different name/version at runtime
  would silently create a *different* projection rather than override
  the existing one's config, which is almost certainly a bug. If you
  want a second projection with a different version, declare a second
  module with `version: 2` in its `use`.
  """
  @spec start_projection(module(), keyword()) ::
          {:ok, pid()} | {:error, :already_started | term()}
  def start_projection(module, overrides) when is_atom(module) and is_list(overrides) do
    config = load_config!(module)

    case Enum.find(overrides, fn {k, _} -> k in [:name, :version] end) do
      nil ->
        :ok

      {bad_key, _} ->
        raise ArgumentError, """
        Scriba.start_projection/2 cannot override :#{bad_key} at runtime — it is part
        of the projection's identity, set at compile time via `use Scriba.Projection`.
        Got override: #{inspect(bad_key)}=#{inspect(Keyword.get(overrides, bad_key))}
        """
    end

    opts =
      config
      |> Map.to_list()
      |> Keyword.merge(overrides)

    case DynamicSupervisor.start_child(
           Scriba.Projections.Supervisor,
           {Scriba.Projection.Supervisor, opts}
         ) do
      {:ok, pid} -> {:ok, pid}
      {:error, {:already_started, pid}} -> {:error, :already_started, pid}
      {:error, :already_started} -> {:error, :already_started}
      other -> other
    end
    |> normalize_start_result()
  end

  defp normalize_start_result({:error, :already_started, _pid}), do: {:error, :already_started}
  defp normalize_start_result(other), do: other

  @doc """
  Returns running projections as `[%{name, version, state}]`.

  Enumerates Coordinators registered in `Scriba.Registry`; includes
  projections in any state, including `:stopped` (terminal Coordinators
  that haven't been terminated from the supervision tree). Empty list
  when no projections are running.
  """
  @spec list() :: [%{name: String.t(), version: pos_integer(), state: atom()}]
  def list do
    # Match spec: select Registry entries whose KEY shape is
    # {:coordinator, name, version} and extract the {name, version} pair.
    # See Registry.select/2 docs for the match-spec grammar.
    Scriba.Registry
    |> Registry.select([
      {{{:coordinator, :"$1", :"$2"}, :_, :_}, [], [{{:"$1", :"$2"}}]}
    ])
    |> Enum.map(fn {name, version} ->
      case Coordinator.get_status(name, version) do
        {:ok, status} -> %{name: name, version: version, state: status.state}
        # Coordinator was registered but disappeared between select and
        # get_status — race during shutdown. Skip.
        {:error, :not_found} -> nil
      end
    end)
    |> Enum.reject(&is_nil/1)
  end

  ## Lifecycle — info / pause / resume / stop (module + string forms)

  @doc """
  Returns a snapshot of the projection's runtime state. See `Scriba.Info` for
  the shape.

  Accepts either a projection module or a `(name, version)` pair. The
  arity-1 string form defaults to version 1.

  Returns `{:error, :not_found}` if no Coordinator is registered for the
  given identity.

  When the projection has more than #{@stream_position_truncation_limit}
  distinct streams, `:stream_positions` is `:truncated` rather than a
  giant map; `:safe_position` is always populated regardless.

  `:safe_position` is the minimum across *cached* streams — an introspection
  figure, not a replay point. See `Scriba.Info` for why
  it reads high and what it is not safe to build on.

  The `:status` field is one of `:initializing | :running | :paused |
  :draining | :stopped`. `:initializing` is the brief window between
  Coordinator start and Broadway producer registration; transitions to
  `:running` automatically.
  """
  @spec info(module() | String.t()) :: {:ok, Info.t()} | {:error, :not_found}
  def info(module_or_name)

  def info(module) when is_atom(module) and module not in [nil, true, false] do
    config = load_config!(module)
    info(config.name, config.version)
  end

  def info(name) when is_binary(name), do: info(name, 1)

  @doc "See `info/1`. Takes explicit `name` (string) and `version` (integer)."
  @spec info(String.t(), pos_integer()) :: {:ok, Info.t()} | {:error, :not_found}
  def info(name, version) when is_binary(name) and is_integer(version) do
    case Coordinator.get_status(name, version) do
      {:ok, status} -> {:ok, build_info(name, version, status)}
      {:error, :not_found} -> {:error, :not_found}
    end
  end

  @doc """
  Pauses a running projection — the source stops yielding new events.
  In-flight events already in Pipeline processors or batchers continue
  through their commit lifecycle.

  Accepts either a projection module or a name string. Arity-1 string
  form defaults to version 1; see `pause/2` for explicit version.

  ## Return values

    * `:ok` — pause signal sent to the source. The source's `handle_info/2`
      will run on its own schedule; by the time this function returns the
      signal is in the source's mailbox but the source may not yet have
      flipped its internal flag. Operators should NOT assume "no commits
      possible" the instant `pause/1` returns.

    * `{:error, {:invalid_state, state}}` — projection is not in a state
      where pause makes sense. The inner atom is the projection's current
      state:

        - `:initializing` — engine starting up; retry once `info/1`
          reports `:running`.
        - `:paused` — already paused. **No idempotency** — callers wanting
          "make sure this is paused" semantics check `info/1` first or
          pattern-match this error case as success.
        - `:stopped` — terminal state.
        - `:draining` — stop in progress.
  """
  @spec pause(module() | String.t()) :: :ok | {:error, {:invalid_state, atom()}}
  def pause(module_or_name)

  def pause(module) when is_atom(module) and module not in [nil, true, false] do
    config = load_config!(module)
    Coordinator.pause(config.name, config.version)
  end

  def pause(name) when is_binary(name), do: pause(name, 1)

  @doc "See `pause/1`. Takes explicit `name` (string) and `version` (integer)."
  @spec pause(String.t(), pos_integer()) ::
          :ok | {:error, {:invalid_state, atom()}}
  def pause(name, version) when is_binary(name) and is_integer(version),
    do: Coordinator.pause(name, version)

  @doc """
  Resumes a paused projection.

  Accepts a module or name string; same dispatch semantics as `pause/1`.

  ## Return values

    * `:ok` — resume signal sent. Same asynchrony caveat as `pause/1`:
      first commit follows whenever the source's `handle_info/2` runs and
      Broadway's processor stage drains accumulated demand.

    * `{:error, {:invalid_state, state}}` — projection is not paused.
      `:running` is **not** idempotent here either.
  """
  @spec resume(module() | String.t()) :: :ok | {:error, {:invalid_state, atom()}}
  def resume(module_or_name)

  def resume(module) when is_atom(module) and module not in [nil, true, false] do
    config = load_config!(module)
    Coordinator.resume(config.name, config.version)
  end

  def resume(name) when is_binary(name), do: resume(name, 1)

  @doc "See `resume/1`."
  @spec resume(String.t(), pos_integer()) ::
          :ok | {:error, {:invalid_state, atom()}}
  def resume(name, version) when is_binary(name) and is_integer(version),
    do: Coordinator.resume(name, version)

  @doc """
  Stops a running or paused projection. Waits for in-flight Broadway
  shutdown to complete before returning.

  Returns `:ok` on success, `{:error, {:invalid_state, state}}` if the
  projection is `:initializing`, `:stopped`, or `:draining`.
  """
  @spec stop(module() | String.t()) :: :ok | {:error, {:invalid_state, atom()}}
  def stop(module_or_name)

  def stop(module) when is_atom(module) and module not in [nil, true, false] do
    config = load_config!(module)
    Coordinator.stop(config.name, config.version)
  end

  def stop(name) when is_binary(name), do: stop(name, 1)

  @doc "See `stop/1`."
  @spec stop(String.t(), pos_integer()) ::
          :ok | {:error, {:invalid_state, atom()}}
  def stop(name, version) when is_binary(name) and is_integer(version),
    do: Coordinator.stop(name, version)

  ## Internals

  defp load_config!(module) do
    unless Code.ensure_loaded?(module) do
      raise ArgumentError, """
      Module #{inspect(module)} is not loaded. Cannot read Scriba projection config.
      Make sure the module is compiled and the application is started before calling.
      """
    end

    unless function_exported?(module, :__scriba_config__, 0) do
      raise ArgumentError, """
      Module #{inspect(module)} does not appear to be a Scriba projection.
      Expected #{inspect(module)} to be defined with `use Scriba.Projection, ...`.
      """
    end

    module.__scriba_config__()
  end

  defp build_info(name, version, status) do
    streams = Scriba.Position.stream_positions(name, version)
    safe = Scriba.Position.safe_position(name, version)

    stream_positions =
      if map_size(streams) > @stream_position_truncation_limit do
        :truncated
      else
        streams
      end

    %Info{
      name: name,
      version: version,
      status: Map.get(status, :state),
      source: Map.get(status, :source),
      target: Map.get(status, :target),
      safe_position: safe,
      stream_positions: stream_positions
    }
  end
end