# `AshHooks.Dispatcher`
[🔗](https://github.com/baselabs/ash_hooks/blob/v2.0.3/lib/ash_hooks/dispatcher.ex#L1)

The outbound fanout driver: one event → every matching subscription's
endpoint → a durable per-endpoint delivery row (+ an enqueue handoff),
with per-endpoint isolation — the outbound twin of `AshHooks.Ingress`.

    {:ok, event} =
      AshHooks.Event.new(
        type: :order_paid,
        payload: body,
        # Deterministic: derived from the artifact's stable id, so a
        # re-fired producer matches the existing delivery row (the
        # {endpoint_id, event_uuid} dedup) instead of fanning out a
        # duplicate POST per sweep
        id: "msg_order-" <> order.id
      )

    AshHooks.dispatch(Order, :order_paid, event, enqueue: {MyRuntime, :enqueue})

Each row stores two ownership identifiers. `dispatch_source` binds the row
to this outbound declaration and endpoint resource. `dispatch_route` binds
it to the enqueue callback. A duplicate with a different source or route
fails explicitly; it cannot silently adopt another declaration or queue.
Named callbacks and external captures normalize to one route. Anonymous
callbacks need a stable `:enqueue_key` for periodic recovery. Without one,
dispatch can enqueue immediately, but recovery reports `:unresolved_route`.

The runtime contract:

  * every matching enabled endpoint gets a delivery row unique on
    `{endpoint_id, event_uuid}`. The worker's queue identity adds the
    complete delivery key, resource modules, source, route, and tenant;
  * the row persists before enqueue — payload bytes included —
    so a later runtime can sign and send from the ledger of record
    alone;
  * one endpoint's enqueue failure or raise records `:enqueue_failed`
    on its row and never stops the others (the fanout-isolation
    guarantee);
  * re-dispatching an `:enqueue_failed` event conditionally moves that row
    back to `:pending`; only the dispatcher that wins the update calls the
    enqueuer. Scheduled reconciliation uses a separate enqueue token and
    30-second lease, so a crashed claimant becomes recoverable without
    borrowing the send lease or retry schedule;
  * with no `:enqueue` configured the rows persist `:pending` and the
    results say `:deferred` — the durable ledger IS the source of
    truth; the delivery runtime (`AshHooks.Delivery`) drives pending
    rows from there.

Global failures (unknown outbound declaration, an invalid event or enqueue
route, missing DSL module options, an unreadable subscription set, or
divergent effective signing modes for one endpoint) return
`{:error, reason}` before any row is written. Subscriptions are read through the
consumer's primary read action and endpoints through their resource's
`get` — both unauthorized, the signature of the inbound machine: the
dispatch call is the trust boundary for writes, read surfaces stay
governed by consumer policies (ADR-0005).

# `dispatch_result`

```elixir
@type dispatch_result() :: %{
  endpoint_id: term(),
  subscription_id: term(),
  status:
    :created
    | :duplicate
    | :deferred
    | :enqueue_failed
    | :mark_failed
    | :endpoint_error
    | :reconciled,
  error: term()
}
```

The dispatch result contract (public since 1.2.0): one map per endpoint.

# `dispatch`

```elixir
@spec dispatch(module(), atom(), AshHooks.Event.t() | term(), keyword()) ::
  {:ok, [dispatch_result()]} | {:error, term()}
```

Fans one event out to every matching subscription's enabled endpoint.
Returns `{:ok, results}` — one map per endpoint
(`%{endpoint_id:, subscription_id:, status:, error:}` with `status` in
`:created | :duplicate | :deferred | :enqueue_failed | :mark_failed |
:endpoint_error`) — or `{:error, reason}` for global failures.

Options:

  * `:enqueue` — the enqueue seam: a 2-arity function
    (`fn delivery, event -> :ok | {:error, term}`) or a `{module, function}`
    pair applied as `apply(module, function, [delivery, event])`. The
    delivery runtime `AshHooks.Delivery.run/2` via the worker macro is its
    canonical implementation; `nil` (the default) persists `:pending` rows
    and returns `:deferred` results.
  * `:enqueue_key` — stable, nonempty binary (up to 512 bytes) or atom
    identity for an anonymous enqueue callback. Supply it when periodic
    recovery must reconstruct that route.
  * `:tenant` — the dispatch tenant. On multitenant resources (the
    tenancy contract: every touched resource declares attribute
    multitenancy over the same attribute) this scopes the fanout, the
    endpoint resolution, and every ledger write to the tenant; omitting
    it there is `{:error, :tenant_required}` before any data access.
    On single-tenant resources the option is inert.

# `reconcile_pending`

```elixir
@spec reconcile_pending(module(), atom(), keyword()) ::
  {:ok, [dispatch_result()]} | {:error, term()}
```

Recovers source-owned delivery rows: stale `:pending`, `:enqueue_failed`,
due `:failed_retryable`, expired `:sending`, and `:disable_pending`.
Future retries and live send leases are ineligible.

Each row is claimed with a separate enqueue token and a 30-second lease in the
same atomic update that checks its state and due time. Successful durable
admission releases the token and PARKS the enqueue lease for one 30-second
horizon (the row stays unclaimable while concurrent sweeps from the same
recovery window are in flight — a sweep inside that horizon sees the row as a
contended `:duplicate`); the row's delivery state, attempts, retry time, send
lease, and `last_error` diagnostic are preserved. A failed enqueue releases
the claim fully and records its bounded error. A crash leaves a reclaimable lease. The enqueue
seam must remain idempotent because an external effect can occur before a
crashed claimant clears its lease.

Reconciliation validates the stored declaration source and route. A
deferred unbound route may bind once to a named callback or to an anonymous
callback carrying the same stable `:enqueue_key`. An unkeyed anonymous route
returns `:unresolved_route` and is never rebound implicitly.

Options: `:older_than` (staleness cutoff; default now minus 5 minutes),
`:tenant` (the pre-flight applies), `:enqueue`, and `:enqueue_key`. The seam
receives the event reconstructed from the row, never nil. The resource
must carry `inserted_at` (the retention hooks' requirement).

Keep the cutoff beyond normal enqueue latency. The default cutoff moves
with the clock; periodic recovery is a host scheduling obligation.

---

*Consult [api-reference.md](api-reference.md) for complete listing*