AshHooks

Webhooks for Ash Framework — inbound (receive, verify per-provider signatures, deduplicate, invoke the provider handler and record its outcome) and outbound (sign, deliver, retry, track).

Status: v0.1.0 — inbound + outbound complete. Not yet shipped: retention/TTL cleanup hooks (ledger and delivery rows accumulate until you clean them — ADR-0005 names the hooks as a floor; tracked for a following release). Inbound: per-provider signature verification (AshHooks.Provider behaviour; ComplyCube + HubSpot v3 reference providers), a fenced unique-ingest ledger with claim/lease fencing and a reaper, and fail-closed DSL verifiers. Outbound: AshHooks.dispatch/4 fanout with per-endpoint isolation, the delivery runtime on Oban (use AshHooks.Worker) with row-owned retry policy, Standard Webhooks signing (v1/v1a, legacy :dual migration mode), and a memory-bounded native HTTP adapter. Security floors ship in the package (ADR-0005): secrets as sources only, SSRF guards at registration + send, response snippets store no body bytes by default. Telemetry events for the whole lifecycle (see AshHooks.Telemetry). Design records: #1, ADR-0001–0008.

Inbound and outbound are independently consumable: inbound-only applications pull no queue infrastructure.

Installation

def deps do
  [
    {:ash_hooks, "~> 0.1.0"},
    # outbound delivery only (inbound-only apps need none of these):
    {:oban, "~> 2.20"}
  ]
end

Elixir ~> 1.15, Ash _{> 3.0. Optional components: Oban (}> 2.20) for outbound delivery — added, migrated, configured, and supervised by your app; Plug/Phoenix for inbound receipt (the raw-body reader below).

Or with igniter:

mix igniter.install ash_hooks

The installer ATTEMPTS to patch your endpoint's Plug.Parsers with a body_reader — signature schemes sign the exact wire bytes, and a router plug cannot recover pre-parser bytes. Review the generated diff; if it could not locate the call, add it by hand:

plug Plug.Parsers,
  parsers: [:json],
  pass: ["*/*"],
  body_reader: {AshHooks.BodyReader, :read_body, []},
  json_decoder: Phoenix.json_library()

By default every parsed request carries a cached copy of its raw body; pass [only: ["/webhooks"]] as the reader's third element to scope that memory cost to the webhook routes.

Your migrations create the tables AND the two UNIQUE INDEXES the dedup guarantees rest on ([provider, external_event_id | scope] for inbound, [endpoint_id, event_uuid] for outbound) — the get-started tutorial carries complete runnable shapes.

Usage

Attach to a resource and declare sources and events:

use Ash.Resource,
  data_layer: AshSqlite.DataLayer,
  extensions: [AshHooks, AshHooks.InboundDelivery]

inbound_delivery do
  # provider event ids are not globally unique across accounts — the
  # unique-ingest identity extends by your scope slots; each slot must
  # be a non-nullable attribute (the DSL verifier enforces both)
  scope_identity([:account_id])
end

attributes do
  attribute(:account_id, :string, allow_nil?: false)
end

webhooks do
  # convention-resolves to AshHooks.Provider.ComplyCube
  inbound :comply_cube do
    secret {:app_env, [:my_app, :complycube_secret]}
  end

  # convention-resolves to AshHooks.Provider.HubSpotV3 — the vendor's
  # five-minute replay window applies by default; replay_window_seconds
  # overrides it in either direction
  inbound :hub_spot_v3 do
    secret {:app_env, [:my_app, :hubspot_client_secret]}
  end

  outbound :order_paid do
    signing_mode :standard
  end
end

Inbound (sync mode) — a controller reads the cached raw body and drives the fenced machine:

AshHooks.Ingress.ingest(Ledger, :comply_cube, conn.private[:ash_hooks_raw_body], %{
  signature: get_req_header(conn, "complycube-signature") |> List.first(),
  headers: Map.new(conn.req_headers),
  scope: %{account_id: connection.account_id}
})

HubSpot's v3 scheme signs the method and the full request URI alongside the body, so its controller passes both. Build the PUBLIC URI from values you configure (a base URL you control), not blindly from conn — behind a TLS-terminating proxy, conn.host/port/scheme are the INTERNAL ones and the signature will not match the URI HubSpot signed. Where you do reconstruct from conn, remember conn.query_string excludes the ? and conn.host excludes a non-default port:

query = if conn.query_string == "", do: "", else: "?" <> conn.query_string

AshHooks.Ingress.ingest(Ledger, :hub_spot_v3, conn.private[:ash_hooks_raw_body], %{
  signature: get_req_header(conn, "x-hubspot-signature-v3") |> List.first(),
  headers: Map.new(conn.req_headers),
  method: conn.method,
  request_uri: "https://" <> conn.host <> conn.request_path <> query
})

HubSpot delivers batches — a top-level JSON array of event objects. The ledger stores the array verbatim; a homogeneous batch parses to its subscription type (contact.creation → :contact_creation), a mixed batch to :mixed (fan out per event in your handler), and an undocumented subscription type fails closed into the ledger as failed_permanent (unknown_event_type) — recorded and auditable.

The machine persists the decoded payload (with a digest binding it to the signed raw bytes) before handling, deduplicates on storage-level uniqueness (exactly one :created per delivery, concurrent or sequential), and fences claims with a monotonic token and an expiring lease. Once the durable row exists, a crash between any two steps re-drives on redelivery instead of silently dropping, and a stale owner (superseded or expired lease) can never mark. Terminal rows (:processed / :failed_permanent) are never processed again, but a crash after handler side effects and before the ledger mark re-invokes the handler on redelivery — durable deduplication with AT-LEAST-ONCE handler invocation; write handlers idempotent, keyed on the external event identity. Handler outcomes land in the ledger (:processed, :failed_retryable, :failed_permanent); expired leases are re-driven by AshHooks.Ingress.reap/1.

Outbound deliveries are signed per the Standard Webhooks specification (webhook-id / webhook-timestamp / webhook-signature, v1 HMAC-SHA256 and v1a ed25519 — old+new key rotation on both schemes), so receivers verify with any conformant library; a :dual mode additionally emits a legacy envelope during receiver migration — it REQUIRES the endpoint to carry a legacy_secret_ref (the resolver's base value always signs the Standard Webhooks envelope; legacy slots come only from the endpoint's legacy references).

Outbound fanout (landed): declare the Subscription / Endpoint / OutboundDelivery resources on your data layer, point the outbound declaration at them, and dispatch:

defmodule MyApp.WebhookEndpoint do
  use Ash.Resource,
    data_layer: AshSqlite.DataLayer,
    domain: MyApp,
    extensions: [AshHooks.Endpoint]

  sqlite do
    table("webhook_endpoints")
    repo(MyApp.Repo)
  end

  actions do
    defaults([:read, :create, :update])
  end
end

defmodule MyApp.WebhookSubscription do
  use Ash.Resource,
    data_layer: AshSqlite.DataLayer,
    domain: MyApp,
    extensions: [AshHooks.Subscription]

  sqlite do
    table("webhook_subscriptions")
    repo(MyApp.Repo)
  end

  actions do
    defaults([:read, :create])
  end

  subscription do
    endpoint_resource(MyApp.WebhookEndpoint)
  end
end

defmodule MyApp.OutboundDelivery do
  use Ash.Resource,
    data_layer: AshSqlite.DataLayer,
    domain: MyApp,
    extensions: [AshHooks.OutboundDelivery]

  sqlite do
    table("outbound_deliveries")
    repo(MyApp.Repo)
  end

  actions do
    defaults([:read])
  end
end

webhooks do
  outbound :order_paid do
    subscriptions(MyApp.WebhookSubscription)
    deliveries(MyApp.OutboundDelivery)
  end
end

With no enqueue: configured this persists :pending rows — the durable ledger only; nothing sends until a runtime drives them:

{:ok, event} =
  AshHooks.Event.new(type: :order_paid, payload: Jason.encode!(order))

AshHooks.dispatch(Order, :order_paid, event)

Each matching enabled endpoint gets a durable delivery row unique on {endpoint_id, event_uuid} — the same pair the Oban job uniqueness keys use — carrying the exact payload bytes to sign and the frozen effective signing mode. Endpoints store secret REFERENCES only (whsec_-shaped literals are rejected at cast, on every write path); endpoints carry a durable :enabled | :disabled state the dispatcher respects. One endpoint's enqueue failure records :enqueue_failed on its row and never stops its siblings; a re-dispatch claims the failed row via a CAS and retries the enqueue exactly once per won claim. With no :enqueue configured, rows persist :pending (:deferred results).

Delivery runtime (landed): define ONE worker module in your app and wire it as the dispatch enqueuer —

defmodule MyApp.WebhookDeliveryWorker do
  use AshHooks.Worker,
    deliveries: MyApp.OutboundDelivery,
    endpoints: MyApp.WebhookEndpoint,
    secret_resolver: {MyApp.Secrets, :webhook_secret},
    queue: :webhooks
end

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

The worker drives AshHooks.Delivery (ADR-0008: the ROW owns the retry policy — attempts, next_attempt_at, the dead-letter ceiling — and Oban is the durable trigger). Sends are Standard-Webhooks signed per the row's mode with the same webhook-id on every retry; only 2xx succeeds; redirects are never followed; 410 disables the endpoint durably; 408/429 honor Retry-After (bounded); 5xx/transport failures back off exponentially with jitter; other 4xx and refused redirects dead-letter immediately. Response snippets store NO body bytes by default — a fixed summary of the status and an allowlisted content-type token (ADR-0005's snippet amendment); a per-call snippet_capture: true in the AshHooks.Delivery.run/2 config opts one diagnostic run into body capture, which persists the [captured]-marked body under the package floor (NFKC homoglyph folding, a bounded-fixpoint decode chain, separator-tolerant marker patterns, and a ≥16-char union-alphabet entropy rule) with an optional fail-closed snippet_redactor callback ahead of it. Machine-written fields accept no action input. SSRF is guarded at registration (the endpoint's url type rejects private/loopback/link-local/metadata literals and non-http schemes on every write path) and re-checked at send with DNS re-resolution. HTTP goes through the AshHooks.Http adapter behaviour — the default is AshHooks.Http.Bounded, a minimal HTTP/1.1 client whose EVERY read is capped (headers, and bodies under Content-Length, chunked, and read-to-close framings alike — no response can balloon a worker's memory); AshHooks.Http.Httpc (OTP :httpc) is available as an alternative, and you can inject your own for tests or proxies. The package still compiles and runs Oban-free (CI no-optional leg + the inbound-only proof); use AshHooks.Worker without Oban on the host fails deterministically at compile.

Dispatch-time capture (consumer-owned, no package change)

snippet_capture is deliberately a per-call runtime config key, not a worker-macro knob — but dispatch-time opt-in needs no contract change: bring your own enqueue seam and your own Oban worker driving the public runtime with the flag merged in.

defmodule MyApp.CaptureWorker do
  use Oban.Worker, queue: :webhook_diagnostics

  def perform(%Oban.Job{args: args}) do
    AshHooks.Delivery.run(args,
      deliveries: MyApp.OutboundDelivery,
      endpoints: MyApp.WebhookEndpoint,
      secret_resolver: {MyApp.Secrets, :webhook_secret},
      snippet_capture: true
    )
  end
end

# the enqueue seam contract is (delivery, event) -> :ok | {:error, term}
AshHooks.dispatch(Order, :order_paid, event,
  enqueue: fn delivery, _event ->
    %{ "endpoint_id" => to_string(delivery.endpoint_id),
       "event_uuid" => delivery.event_uuid }
    |> MyApp.CaptureWorker.new()
    |> Oban.insert()
    |> case do
      {:ok, _job} -> :ok
      {:error, reason} -> {:error, reason}
    end
  end
)

For a one-row diagnostic re-drive of an already-dispatched event, call AshHooks.Delivery.run/2 directly with snippet_capture: true — the row's {endpoint_id, event_uuid} args and your config are all it takes.

Observability

Nine lifecycle events cover the send/receive hot paths — ingress verify/dedup/claim, dispatch enqueue-failure, delivery attempt/result/backoff/dead-letter/endpoint-disable. (Successful dispatches and post-claim inbound outcomes are observed on the ledger rows themselves, not as events.) Events carry ids, integers, fixed-vocabulary atoms, and classified reason strings only — never secrets, bodies, or payloads (ADR-0005), so they are safe to ship to any metrics/APM backend. :telemetry.execute/3 matches exact event names, so consume the surface with one attach_many — the full list and a copy-paste handler live in AshHooks.Telemetry's docs and the get-started tutorial.

Design records

Architectural decisions live in docs/adr/.

License

MIT.