AshHooks.Dispatcher (AshHooks v1.0.2)

Copy Markdown View Source

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 slice drives pending rows.

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

Functions

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.

Functions

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

@spec dispatch(module(), atom(), AshHooks.Event.t() | term(), keyword()) ::
  {:ok, [map()]} | {: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.