ferry

defmodule Ferry do
  @moduledoc """
  In-memory operation queue with batched processing, dead-letter queues, and telemetry.

  Ferry buffers external operations in-memory, processes them in configurable batches
  via a user-defined resolver function, and provides full operation lifecycle tracking.

  ## Quick Start

      # 1. Define your resolver
      defmodule MyResolver do
        def resolve(%Ferry.Batch{operations: ops}) do
          Enum.map(ops, fn op ->
            case process(op.payload) do
              {:ok, result} -> {op.id, :ok, result}
              {:error, reason} -> {op.id, :error, reason}
            end
          end)
        end
      end

      # 2. Define your Ferry
      defmodule MyFerry do
        use Ferry,
          resolver: &MyResolver.resolve/1,
          batch_size: 25,
          flush_interval: 5_000
      end

      # 3. Add to supervision tree
      children = [MyFerry]
      Supervisor.start_link(children, strategy: :one_for_one)

      # 4. Push operations
      {:ok, id} = Ferry.push(MyFerry, %{data: "hello"})
      {:ok, %Ferry.Operation{status: :completed}} = Ferry.status(MyFerry, id)

  ## Configuration

  | Option | Default | Description |
  |---|---|---|
  | `name` | *required* | Atom identifier for the Ferry instance |
  | `resolver` | *required* | Function that processes a `%Ferry.Batch{}` |
  | `batch_size` | `10` | Max operations per batch flush |
  | `flush_interval` | `5_000` | Milliseconds between auto-flushes |
  | `max_queue_size` | `10_000` | Max pending operations (back-pressure) |
  | `auto_flush` | `true` | Whether timer-based flushing is active |
  | `operation_timeout` | `300_000` | Max ms for resolver execution |
  | `max_completed` | `1_000` | Max completed operations in history |
  | `completed_ttl` | `1_800_000` | TTL for completed operations (ms) |
  | `persistence` | `:memory` | `:memory` or `:ets` |
  | `id_generator` | `&Ferry.IdGenerator.generate/1` | Custom ID generator |
  | `batch_tracking` | `false` | Enable batch history tracking |
  | `max_batch_history` | `1_000` | Max batch records to retain |
  | `batch_history_ttl` | `3_600_000` | TTL for batch records (ms) |
  """

  @type ferry_name :: atom()

  # ── use Ferry macro ──

  @doc """
  Generates `child_spec/1` and `start_link/1` for clean supervision tree integration.

  The module name becomes the default ferry name.

  ## Example

      defmodule MyApp.ApiFerry do
        use Ferry,
          resolver: &MyApp.ApiResolver.resolve/1,
          batch_size: 25,
          flush_interval: :timer.seconds(5)
      end

      # In your supervision tree:
      children = [MyApp.ApiFerry]
  """
  defmacro __using__(opts) do
    quote do
      def child_spec(override_opts \\ []) do
        merged = Keyword.merge(unquote(opts), override_opts)
        merged = Keyword.put_new(merged, :name, __MODULE__)

        %{
          id: __MODULE__,
          start: {Ferry, :start_link, [merged]},
          type: :supervisor
        }
      end

      def start_link(override_opts \\ []) do
        merged = Keyword.merge(unquote(opts), override_opts)
        merged = Keyword.put_new(merged, :name, __MODULE__)
        Ferry.start_link(merged)
      end

      defoverridable child_spec: 1, start_link: 1
    end
  end

  # ── Lifecycle ──

  @doc """
  Starts a Ferry instance under its own supervision tree.

  ## Required options

    * `:name` — atom identifier for the instance
    * `:resolver` — function that processes a `%Ferry.Batch{}`

  See module documentation for all available options.
  """
  @spec start_link(keyword()) :: Supervisor.on_start()
  def start_link(opts) do
    Ferry.InstanceSupervisor.start_link(opts)
  end

  @doc """
  Returns a child specification for starting under a supervisor.
  """
  def child_spec(opts) do
    name = Keyword.fetch!(opts, :name)

    %{
      id: name,
      start: {__MODULE__, :start_link, [opts]},
      type: :supervisor
    }
  end

  # ── Push operations ──

  @doc """
  Push a single operation. Returns `{:ok, id}` or `{:error, :queue_full}`.
  """
  @spec push(ferry_name(), term()) :: {:ok, String.t()} | {:error, :queue_full}
  def push(ferry_name, payload) do
    Ferry.Server.push(ferry_name, payload)
  end

  @doc """
  Push multiple operations atomically.

  All-or-nothing: if the queue can't fit all operations, none are pushed.
  """
  @spec push_many(ferry_name(), [term()]) :: {:ok, [String.t()]} | {:error, :queue_full}
  def push_many(ferry_name, payloads) do
    Ferry.Server.push_many(ferry_name, payloads)
  end

  # ── Query ──

  @doc """
  Get operation by ID. Returns `{:error, :not_found}` if purged or never existed.
  """
  @spec status(ferry_name(), String.t()) :: {:ok, Ferry.Operation.t()} | {:error, :not_found}
  def status(ferry_name, operation_id) do
    Ferry.Server.status(ferry_name, operation_id)
  end

  @doc """
  Get current stats snapshot.
  """
  @spec stats(ferry_name()) :: Ferry.Stats.t()
  def stats(ferry_name) do
    Ferry.Server.stats(ferry_name)
  end

  @doc """
  Get current queue size.
  """
  @spec queue_size(ferry_name()) :: non_neg_integer()
  def queue_size(ferry_name) do
    Ferry.Server.queue_size(ferry_name)
  end

  @doc """
  Get the runtime configuration for this Ferry instance.
  """
  @spec config(ferry_name()) :: map()
  def config(ferry_name) do
    Ferry.Server.config(ferry_name)
  end

  @doc """
  Get multiple operations by ID. Returns a map of `%{id => {:ok, Operation} | {:error, :not_found}}`.
  """
  @spec status_many(ferry_name(), [String.t()]) :: %{
          String.t() => {:ok, Ferry.Operation.t()} | {:error, :not_found}
        }
  def status_many(ferry_name, operation_ids) do
    Ferry.Server.status_many(ferry_name, operation_ids)
  end

  # ── Flow control ──

  @doc """
  Trigger immediate flush. Synchronous — blocks the caller until the resolver
  finishes processing the batch (subject to `operation_timeout`).
  """
  @spec flush(ferry_name()) :: :ok | {:error, :empty_queue}
  def flush(ferry_name) do
    Ferry.Server.flush(ferry_name)
  end

  @doc """
  Pause auto-flush timer. Manual flush still works.
  """
  @spec pause(ferry_name()) :: :ok
  def pause(ferry_name) do
    Ferry.Server.pause(ferry_name)
  end

  @doc """
  Resume auto-flush timer.
  """
  @spec resume(ferry_name()) :: :ok
  def resume(ferry_name) do
    Ferry.Server.resume(ferry_name)
  end

  # ── Dead Letter Queue ──

  @doc """
  List all dead-lettered operations.
  """
  @spec dead_letters(ferry_name()) :: [Ferry.Operation.t()]
  def dead_letters(ferry_name) do
    Ferry.Server.dead_letters(ferry_name)
  end

  @doc """
  Count dead-lettered operations.
  """
  @spec dead_letter_count(ferry_name()) :: non_neg_integer()
  def dead_letter_count(ferry_name) do
    Ferry.Server.dead_letter_count(ferry_name)
  end

  @doc """
  Move all dead letters back to the pending queue for re-processing.
  """
  @spec retry_dead_letters(ferry_name()) :: {:ok, non_neg_integer()}
  def retry_dead_letters(ferry_name) do
    Ferry.Server.retry_dead_letters(ferry_name)
  end

  @doc """
  Permanently discard all dead letters.
  """
  @spec drain_dead_letters(ferry_name()) :: :ok
  def drain_dead_letters(ferry_name) do
    Ferry.Server.drain_dead_letters(ferry_name)
  end

  @doc """
  Clear all pending operations from the queue.

  Pending operations are moved to the Dead Letter Queue with error `:canceled`.
  Returns `{:ok, count}` with the number of cleared operations.
  """
  @spec clear(ferry_name()) :: {:ok, non_neg_integer()}
  def clear(ferry_name) do
    Ferry.Server.clear(ferry_name)
  end

  # ── Introspection ──

  # ── Batch History ──

  @doc """
  List recent batch history records.

  Requires `batch_tracking: true` in instance config.

  ## Options

    * `:limit` — max records to return (default: 50)
    * `:status` — filter by batch status (`:completed`, `:timeout`, etc.)
  """
  @spec batch_history(ferry_name(), keyword()) :: [Ferry.BatchRecord.t()] | {:error, :not_enabled}
  def batch_history(ferry_name, opts \\ []) do
    Ferry.BatchTracker.batch_history(ferry_name, opts)
  end

  @doc """
  Get details for a specific batch by ID.

  Requires `batch_tracking: true` in instance config.
  """
  @spec batch_info(ferry_name(), String.t()) ::
          {:ok, Ferry.BatchRecord.t()} | {:error, :not_found} | {:error, :not_enabled}
  def batch_info(ferry_name, batch_id) do
    Ferry.BatchTracker.batch_info(ferry_name, batch_id)
  end

  @doc """
  Manually purge all batch history.

  Requires `batch_tracking: true` in instance config.
  """
  @spec purge_batch_history(ferry_name()) :: :ok | {:error, :not_enabled}
  def purge_batch_history(ferry_name) do
    Ferry.BatchTracker.purge_history(ferry_name)
  end

  # ── Introspection ──

  @doc """
  List pending operations.

  ## Options

    * `:limit` — max operations to return (default: 50)
  """
  @spec pending(ferry_name(), keyword()) :: [Ferry.Operation.t()]
  def pending(ferry_name, opts \\ [limit: 50]) do
    Ferry.Server.pending(ferry_name, opts)
  end

  @doc """
  List completed operations.

  ## Options

    * `:limit` — max operations to return (default: 50)
  """
  @spec completed(ferry_name(), keyword()) :: [Ferry.Operation.t()]
  def completed(ferry_name, opts \\ [limit: 50]) do
    Ferry.Server.completed(ferry_name, opts)
  end
end