Get Started

Copy Markdown View Source

This walkthrough takes a new application from install to a verified inbound webhook and a delivered outbound webhook. Both halves are independently consumable: inbound-only applications need no Oban.

Requirements: Elixir ~> 1.20 (OTP 28+), Ash >= 3.34.3, < 4.0. Optional components: Oban (~> 2.20) for outbound delivery, Plug/Phoenix for inbound receipt.

This walkthrough builds on a working Ash application — one with a domain, a repo (Postgres or sqlite), and a migration workflow. If you are starting from an empty directory, follow Ash's get-started guide first; the examples below refer to your existing MyApp.Domain and MyApp.Repo (your names will differ).

Installation

def deps do
  [
    {:ash_hooks, "~> 2.0"},
    # for outbound delivery only:
    {:oban, "~> 2.20"}
  ]
end

Or mix igniter.install ash_hooks, which also attempts to patch your endpoint's Plug.Parsers with a body_reader (see below) — review the generated diff; if the patch could not be applied, add it by hand.

One Ash-side requirement to know about: Ash 3.33 and later require every application to choose how string length is counted, and compiling any resource fails until it is set:

# config/config.exs
config :ash, default_string_length_count: :codepoints

:codepoints is Ash's recommendation — max_length bounds value size and matches how SQL data layers count. :mixed keeps the pre-3.33 grapheme counting if you need it. This is an Ash requirement for every app compiling resources, not an ash_hooks one; an application already on Ash 3.33+ necessarily has it set.

The database migrations

ash_hooks injects fields and identities onto your resources; your migrations create the tables and — critically — the two unique indexes that make the dedup guarantees real. Minimal shapes:

# inbound ledger — payload is the decoded body (a :map / JSONB column);
# payload_digest binds it to the signed raw bytes
create table(:webhook_ledgers, primary_key: false) do
  add :id, :uuid, primary_key: true
  add :provider, :text, null: false
  add :external_event_id, :text, null: false
  add :external_event_type, :text
  add :payload, :map, null: false
  add :payload_digest, :text, null: false
  add :status, :text, null: false, default: "received"
  add :fencing_token, :integer, null: false, default: 0
  add :lease_expires_at, :utc_datetime_usec
  add :error_class, :text
  add :attempts, :integer, null: false, default: 0
  # one column per scope_identity slot:
  add :account_id, :text, null: false
  timestamps(type: :utc_datetime_usec)
end

create unique_index(:webhook_ledgers, [:provider, :external_event_id, :account_id])

# outbound ledger — payload here is the exact bytes to sign
create table(:outbound_deliveries, primary_key: false) do
  add :id, :uuid, primary_key: true
  add :event_uuid, :text, null: false
  add :event_type, :text, null: false
  add :payload, :binary, null: false
  add :endpoint_id, :uuid, null: false
  add :subscription_id, :uuid
  add :signing_mode, :text
  add :status, :text, null: false, default: "pending"
  add :attempts, :integer, null: false, default: 0
  add :response_status, :integer
  add :response_snippet, :text
  add :last_error, :text
  add :next_attempt_at, :utc_datetime_usec
  add :dispatch_source, :text, null: false, default: "v1:direct:unbound"
  add :dispatch_route, :text, null: false, default: "v1:route:unbound"
  add :attempt_token, :uuid
  add :send_lease_expires_at, :utc_datetime_usec
  add :enqueue_token, :uuid
  add :enqueue_lease_expires_at, :utc_datetime_usec
  add :endpoint_snapshot, :map
  timestamps(type: :utc_datetime_usec)
end

create unique_index(:outbound_deliveries, [:endpoint_id, :event_uuid])

create table(:webhook_endpoints, primary_key: false) do
  add :id, :uuid, primary_key: true
  add :url, :text, null: false
  add :status, :text, null: false, default: "enabled"
  add :secret_ref, :text, null: false
  add :previous_secret_ref, :text
  add :legacy_secret_ref, :text
  add :legacy_previous_secret_ref, :text
end

create table(:webhook_subscriptions, primary_key: false) do
  add :id, :uuid, primary_key: true
  add :event_types, {:array, :text}, null: false
  add :endpoint_id, :uuid, null: false
  add :signing_mode, :text
end

These are PostgreSQL migrations. For SQLite examples, the guided tour creates its tables and JSON columns against a real SQLite database.

Inbound: receive, verify, dedup

Configure the raw-body reader first — signature schemes sign the exact wire bytes, and a router plug cannot recover pre-parser bytes:

# in your Phoenix Endpoint:
plug Plug.Parsers,
  parsers: [:json],
  pass: ["*/*"],
  body_reader: {AshHooks.BodyReader, :read_body, []},
  # optional: scope the raw-body memory cost to webhook routes only
  # body_reader: {AshHooks.BodyReader, :read_body, [only: ["/webhooks"]]},
  json_decoder: Phoenix.json_library()

The ledger resource (one resource per inbound surface):

defmodule MyApp.WebhookLedger do
  use Ash.Resource,
    domain: MyApp.Domain,
    data_layer: AshPostgres.DataLayer,   # or AshSqlite.DataLayer
    extensions: [AshHooks, AshHooks.InboundDelivery]

  postgres do
    table("webhook_ledgers")
    repo(MyApp.Repo)
  end

  inbound_delivery do
    # provider event ids are not globally unique across accounts — your
    # scope slots extend the unique-ingest identity (each slot must be a
    # non-nullable attribute, supplied on every ingest)
    scope_identity([:account_id])
  end

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

  actions do
    defaults([:read])
  end

  webhooks do
    inbound :comply_cube do
      # convention-resolves the provider to AshHooks.Provider.ComplyCube;
      # a secret source, never a literal
      secret {:app_env, [:my_app, :complycube_secret]}
    end
  end
end

The bundled ComplyCube handler returns a typed verified event; it does not apply your application's business changes. For those changes, set the declaration's provider to your own module, delegate signature verification and event-type parsing to the built-in provider, and implement handle_event/2 with your idempotent domain action. The provider API describes the callbacks. A HubSpot wrapper must also delegate event_identity/1 so retries retain the built-in stable batch identity.

Register the ledger in your existing domain before calling ingress. Keep any resources your application already declares:

defmodule MyApp.Domain do
  use Ash.Domain, otp_app: :my_app

  resources do
    resource(MyApp.WebhookLedger)
    # ... your existing resources ...
  end
end

AshHooks.Ingress.ingest/4 runs the whole sync pipeline — verify the signature over the raw bytes, persist the decoded payload plus a digest binding it to those bytes, dedup on the unique index, claim under a fenced lease, invoke the provider handler, and mark the outcome. From your controller:

raw = conn.private[:ash_hooks_raw_body]

case AshHooks.Ingress.ingest(MyApp.WebhookLedger, :comply_cube, raw, %{
       signature: List.first(get_req_header(conn, "complycube-signature")),
       headers: Map.new(conn.req_headers),
       scope: %{"account_id" => conn.params["account_id"]}
     }) do
  {:ok, :created, %{status: status}} ->
    # Acknowledge terminal outcomes; retry only recoverable failures.
    code = if status in [:processed, :failed_permanent, :superseded], do: 200, else: 500
    send_resp(conn, code, "")

  {:ok, :duplicate, delivery} ->
    # Already seen, but a redelivery can retry the row, so its handler
    # may have failed again: judge the status, not just the tag
    code = if delivery.status in [:processed, :failed_permanent, :superseded], do: 200, else: 500
    send_resp(conn, code, "")

  {:error, _invalid_signature_or_payload} ->
    send_resp(conn, 400, "")
end

The ledger deduplicates deliveries and invokes handlers at least once. A delivery whose row is terminal (:processed / :failed_permanent / :superseded) is never processed again, but a crash after your handler's side effects and before the ledger mark will re-invoke it on redelivery. Write handlers idempotent, keyed on the external event identity. claim_delivery/3, mark_processed/4, mark_failed/6, renew/4 and reap/2 are public if you need to drive the lease machine yourself (e.g. from your own async pipeline) — the sync ingest/4 above is the default.

Permanent failures need an operator-visible error record, rather than an endless provider retry loop. :superseded identifies a retained legacy row replaced by the canonical identity during the 2.0 upgrade; see UPGRADING.

Crash safety: once the durable row exists, a crash between any two steps re-drives on redelivery instead of silently dropping.

Outbound: sign, deliver, retry

The three resources:

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

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

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

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

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

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

  subscription do
    endpoint_resource(MyApp.WebhookEndpoint)
  end
end

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

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

  actions do
    defaults([:read])
  end

  attributes do
    timestamps()
  end
end

The emitting resource declares the event:

defmodule MyApp.Order do
  use Ash.Resource,
    domain: MyApp.Domain,
    data_layer: AshPostgres.DataLayer,
    extensions: [AshHooks]

  # ... your attributes/actions ...

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

Add the outbound resources and emitter to that same domain:

resources do
  resource(MyApp.WebhookLedger)
  resource(MyApp.WebhookEndpoint)
  resource(MyApp.WebhookSubscription)
  resource(MyApp.OutboundDelivery)
  resource(MyApp.Order)
  # ... your existing resources ...
end

Define a worker in your app (Oban must be in your dependencies):

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

The secret resolver maps an endpoint's secret reference to its value — endpoints store references only, never secrets:

defmodule MyApp.Secrets do
  # ref is whatever string you stored on the endpoint's secret_ref.
  # The value is a complete generated secret — create it once with
  # AshHooks.Signing.generate_secret/0 (returns a "whsec_"-prefixed,
  # correctly-encoded binary), store it whole in your secret store,
  # and return it unchanged:
  def webhook_secret("acme-main") do
    case System.fetch_env("ACME_WEBHOOK_SECRET") do
      {:ok, secret} -> {:ok, secret}
      :error -> {:error, :missing_secret}
    end
  end

  def webhook_secret(_unknown), do: {:error, :unknown_ref}
end

Oban itself (dependency, migration, and a supervised instance with a :webhooks queue) is the consumer's to set up — see Oban's installation guide; the worker above plugs into it.

Set PUBLIC_WEBHOOK_TEST_URL to your existing public receiver endpoint and EXAMPLE_ORDER_ID to the stable ID of the order-paid event you are sending. Register an endpoint and a subscription, then dispatch:

# a public, DNS-resolvable HTTPS URL — the send-time SSRF check
# resolves DNS and refuses private, loopback, and special-purpose targets
{:ok, endpoint} =
  Ash.create(MyApp.WebhookEndpoint, %{
    url: System.fetch_env!("PUBLIC_WEBHOOK_TEST_URL"),
    secret_ref: "acme-main"
  }, authorize?: false)

{:ok, _sub} =
  Ash.create(MyApp.WebhookSubscription, %{
    endpoint_id: endpoint.id,
    event_types: ["order_paid"]
  }, authorize?: false)

order_id = System.fetch_env!("EXAMPLE_ORDER_ID")

{:ok, event} =
  AshHooks.Event.new(
    type: :order_paid,
    payload: Jason.encode!(%{"order_id" => order_id}),
    # Reuse this ID when retrying the same logical order-paid event.
    id: "msg_order-" <> order_id
  )

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

With no enqueue: the rows persist :pending (durable ledger only, nothing sends). With the worker seam wired, each row is delivered by AshHooks.Delivery: Standard-Webhooks signed, retried with bounded Retry-After and jittered backoff, dead-lettered at the ceiling, the endpoint durably disabled on 410.

Signing modes: :standard (default) needs only secret_ref. Both :legacy and :dual require the endpoint to carry a legacy_secret_ref — signing fails (and the row retries as signing_failed) without one; :dual additionally emits the legacy envelope alongside the Standard Webhooks one during receiver migration.

Fit the extensions to your domain with per-resource options: rename the exact-bytes column when your domain reserves payload (payload_attribute :event_bytes in the delivery's outbound_delivery block — the dispatch/signing/send paths follow the configured name), omit the retention destroy action on an append-only ledger (prune_action :none — no destroy action exists, and AshHooks.Delivery.prune/2 fails loud), and map the durable enable/disable onto your own switch (the endpoint's endpoint do status_attribute :active; enabled_values [true]; disabled_value false end — the dispatcher, the send path, and the 410 breaker all consult YOUR attribute; the package injects no status). Generated primary keys (uuid_v7_primary_key) compile as-is, and a subscription register redeclared as a closed {:array, :atom} enum matches normally with your own constraints. One obligation comes with the injected surface: the generated write actions are yours to cover with action-specific policies, or keep these resources off actor-facing surfaces. The resource module docs list the full action set, including ownership binding, recovery claims, and pending endpoint disable. Broad application update actions must not expose machine fields.

Response snippets store NO body bytes by default (a status + content-type summary). For a one-row diagnostic capture, re-drive the row with AshHooks.Delivery.run/2 — it takes the same configuration the worker macro bakes (deliveries, endpoints, secret_resolver, retry policy) plus the per-call snippet_capture: true; the guided-tour Livebook shows a worked config. The captured body persists only after passing the package's redaction, marked [captured].

The direct driver uses the same retry defaults as the worker. When copying a custom retry policy, map the worker's delivery_max_attempts to the driver's max_attempts; the other retry option names stay the same. Copy the complete policy for a diagnostic run. A lower delivery ceiling can dead-letter a row that the worker would still retry.

Keep delivery moving after a crash

Schedule recovery alongside the worker. It reconnects a durable row to its queue after a crash between persistence and enqueue, a discarded job, or an expired send lease. Run it for each outbound declaration and tenant:

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

The declaration and queue route must match the stored row. Recovery preserves attempt counts and future retry times; it also completes a pending 410 disable without sending again. Named worker callbacks provide a stable route. If you use an anonymous enqueue callback, supply a stable enqueue_key at dispatch and recovery. Synchronize application-node clocks because leases use application UTC time.

The ledger fences each attempt's result, while network delivery remains at-least-once. A receiver can accept a request before the sender dies, so deduplicate by webhook-id at the receiver. For an existing 1.x installation, follow the ordered upgrade before enabling 2.0 workers.

Observing: telemetry

:telemetry.attach_many("my-ash-hooks", [
  [:ash_hooks, :ingress, :verify],
  [:ash_hooks, :ingress, :dedup],
  [:ash_hooks, :ingress, :claim],
  [:ash_hooks, :dispatch, :enqueue_failed],
  [:ash_hooks, :delivery, :attempt],
  [:ash_hooks, :delivery, :result],
  [:ash_hooks, :delivery, :backoff],
  [:ash_hooks, :delivery, :dead_letter],
  [:ash_hooks, :delivery, :disable]
], fn event, _measurements, metadata, _config ->
  require Logger
  Logger.debug("ash_hooks #{inspect(event)} #{inspect(metadata)}")
end, nil)

attach_many with the exact names is required — :telemetry.execute/3 matches exact names only. Events carry ids, integers, fixed atoms, and classified reasons — never secrets, bodies, or payloads (ADR-0005).

Where next