ash_hooks

# Get Started

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](https://hexdocs.pm/ash/get-started.html) first; the
examples below refer to your existing `MyApp.Domain` and `MyApp.Repo`
(your names will differ).

## Installation

```elixir
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:

```elixir
# 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:

```elixir
# 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](../livebooks/get-started.livemd) 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:

```elixir
# 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):

```elixir
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](https://hexdocs.pm/ash_hooks/AshHooks.Provider.html)
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:

```elixir
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:

```elixir
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](../../UPGRADING.md).

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:

```elixir
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:

```elixir
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:

```elixir
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):

```elixir
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:

```elixir
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](https://hexdocs.pm/oban/installation.html); 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:

```elixir
# 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](../livebooks/get-started.livemd) 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:

```elixir
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](../../UPGRADING.md) before enabling 2.0 workers.

## Observing: telemetry

```elixir
: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

- [Multi-tenant adoption checklist](tenancy-adoption-checklist.md) —
  when one app serves many organizations
- [Guided tour (Livebook)](../livebooks/get-started.livemd) — run the
  whole library inside one notebook
- `AshHooks.Dispatcher`, `AshHooks.Delivery`, and `AshHooks.Ingress`
  module docs — the machines this walkthrough drove