AshHooks.Dispatcher (AshHooks v1.2.0)

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)

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

The machine's contract:

every matching ENABLED endpoint gets a delivery row unique on {endpoint_id, event_uuid} — the same pair the Oban job uniqueness keys use (ADR-0004, verified against deps/oban 2.23.1: fields: [:args], keys: [...] matches on a jsonb CONTAINMENT of exactly these top-level arg keys, so the enqueue seam's args must always carry both);
the row persists BEFORE anything else — 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 CLAIMS the row via a WHERE-gated :enqueue_failed → :pending CAS first — only the dispatcher that wins the flip calls the enqueuer, so concurrent repairs cannot double-enqueue;
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, missing DSL module opts, an unreadable subscription set, 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 \\ [])

Reconciles delivery rows stranded at :pending by a crash between the row write and the enqueue: a WHERE-gated bulk CAS flips status == :pending and inserted_at < cutoff to :enqueue_failed on the already-injected :mark_enqueue_failed action — a REAL state flip, so of N concurrent reconcilers exactly one wins each row (the matched-records count is the win signal, the dispatcher's claim-then-enqueue discipline). Winners are then enqueued through :enqueue (the same seam dispatch takes); nil leaves them CAS-flipped for the existing re-dispatch repair path — the state that path already owns, no new stranded state.

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.
: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()}

A reconciled winner's row stays at :enqueue_failed (marked reconcile_pending in last_error): concurrent reconcilers are one-shot — the CAS only matches :pending, so of N racing reconcilers exactly one wins each row. The row is also exactly the state the re-dispatch repair path owns (a re-dispatch of the same event claims it via the repair CAS, requeues it to :pending, and enqueues again). The package's exactly-once bound is therefore PER CAS TRANSITION, one winner each — cross-MECHANISM exactly-once enqueue is the SEAM's contract: the canonical Oban seam's job uniqueness makes it effect-once (proven on the real engine); a custom seam must be idempotent itself. Note that a row a re-dispatch repair requeued (back to :pending, inserted_at unchanged) is eligible for a LATER reconciliation too — a pending row awaiting its job is indistinguishable from a stranded one until the job drives it. That is one more reason the enqueue seam itself must tolerate a duplicate. A crash mid-reconcile leaves the row at :enqueue_failed: repairable, never lost.

Options: :older_than (staleness cutoff; default now − 5 minutes), :tenant (the pre-flight applies), :enqueue (the dispatch seam — it receives the event RECONSTRUCTED from the row, never nil). The resource must carry inserted_at (the retention hooks' requirement).

Do not run with a cutoff overlapping still-live enqueue latency — a wrongly claimed row's worst case is the :enqueue_failed repair path, but the noise is avoidable by leaving the default. The default cutoff moves with the clock for the same reason: re-running with a FIXED, already-consumed cutoff re-claims previously reconciled rows (they are :pending again with old timestamps) and re-enqueues them.