AshHooks.Dispatcher (AshHooks v2.0.3)

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).

Summary

Types

dispatch_result()

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

Functions

dispatch(resource, name, event, opts \\ [])

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.

reconcile_pending(resource, name, opts \\ [])

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.

Types

dispatch_result()

@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.

Functions

dispatch(resource, name, event, opts \\ [])

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

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(resource, name, opts \\ [])

@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.