AshHooks guided tour

You will learn

• Verify webhooks from any sender by configuration: an HMAC header, a timestamped structured header with a replay window, an Ed25519 signature, a signed JWS body, and a sender that canonicalizes form fields
• Ingest through the ledger: deduplicate a replay, reject a tamper, refuse a stale timestamp, and rotate a secret without dropping deliveries
• Classify events with your own table and run an idempotent handler
• Verify directly, without a ledger, and see what an invalid configuration reports
• Send an outbound webhook to an existing WebHook Tester inbox and verify its captured bytes
• Scope a multi-tenant deployment: one tenant's dispatch never reaches another's rows
• Fit the extensions to a domain that reserves payload, owns an active switch, and types its registers with atoms
• Observe everything through telemetry
• Prune old rows with the retention hooks

This notebook uses a real SQLite database and an existing WebHook Tester receiver. The receiver must already answer at http://127.0.0.1:52871; set LOCAL_WEBHOOK_TESTER_PORT to use another existing receiver. See the local receiver guide. The notebook starts no server. Uses ash_hooks 3.0.0. The tour plays both sides of each webhook: it signs requests the way a sender would and verifies them the way your application does. Real senders, a production queue, and your HTTP endpoint come from your own application.

Setup

package =
  case System.get_env("ASH_HOOKS_NOTEBOOK_SOURCE") do
    nil -> {:ash_hooks, "== 3.0.0"}
    source -> {:ash_hooks, path: Path.expand(source)}
  end

# Release verification supplies the unpacked candidate archive explicitly.
# Normal notebook execution resolves the exact published version above.
Mix.install(
  [
    {:ash, ">= 3.34.3 and < 4.0.0-0"},
    package,
    {:ash_sqlite, "~> 0.2"},
    {:jason, "~> 1.2"}
  ],
  consolidate_protocols: false
)

# Ash 3.33+ requires every application to choose how string length is
# counted before any resource compiles — this notebook is its own app.
# :codepoints is Ash's recommended value: max_length bounds value size
# and matches how SQL data layers count.
Application.put_env(:ash, :default_string_length_count, :codepoints)
"3.0.0" = Application.spec(:ash_hooks, :vsn) |> to_string()

# Keys for the tour's senders. In your application these come from your
# secret store through the inbound `secret` sources, never from literals.
#
# Shared secrets for the HMAC senders:
Application.put_env(:ash_hooks, :tour_orders_secret, :crypto.strong_rand_bytes(32))
Application.put_env(:ash_hooks, :tour_orders_previous_secret, :crypto.strong_rand_bytes(32))
Application.put_env(:ash_hooks, :tour_payments_secret, :crypto.strong_rand_bytes(32))
Application.put_env(:ash_hooks, :tour_sms_secret, :crypto.strong_rand_bytes(20))

# An Ed25519 key pair: the sender keeps the private key, you configure the
# public key (here as base64 of its 32 raw bytes).
{partner_public, partner_private} = :crypto.generate_key(:eddsa, :ed25519)
Application.put_env(:ash_hooks, :tour_partner_public_key, Base.encode64(partner_public))

# An RSA key pair for a sender that signs a JWS body. You configure its
# public key as PEM.
envelope_private = :public_key.generate_key({:rsa, 2048, 65_537})
{:RSAPrivateKey, _, modulus, exponent, _, _, _, _, _, _, _} = envelope_private
envelope_public_pem =
  :public_key.pem_encode([
    :public_key.pem_entry_encode(:SubjectPublicKeyInfo, {:RSAPublicKey, modulus, exponent})
  ])

Application.put_env(:ash_hooks, :tour_envelope_public_key, envelope_public_pem)
IO.puts(envelope_public_pem)

defmodule Tour.Repo do
  use AshSqlite.Repo, otp_app: :ash_hooks
end

Describe your senders

Each sender gets one provider module. use AshHooks.Provider, verify: [...] describes how the sender signs its requests; ash_hooks checks the options when the module compiles and generates the verification callbacks. The options come from the sender's documentation: the header, the algorithm, how the signature and secret are encoded, what is signed, and where the timestamp is. The Verify any sender guide has recipes for common shapes, and AshHooks.Provider.Scheme documents every option.

defmodule Tour.VerifiedEvent do
  defstruct [:type, :payload]
end

# 1. An HMAC-SHA256 of the raw body, lowercase hex, in one header.
#    This provider also classifies events and runs a handler.
defmodule Tour.Orders do
  use AshHooks.Provider,
    verify: [
      algorithm: :hmac_sha256,
      signature: {:header, "x-tour-signature"},
      encoding: :hex
    ]

  # Map the sender's type strings to your event atoms with an explicit
  # table, never String.to_atom on request data (that would let a hostile
  # sender grow your atom table). Unknown types fail closed.
  @types %{"order.paid" => :order_paid}

  @impl true
  def parse_event_type(%{"type" => type}) do
    case Map.fetch(@types, type) do
      {:ok, atom} -> {:ok, atom}
      :error -> {:error, :unknown_event_type}
    end
  end

  def parse_event_type(_payload), do: {:error, :malformed_payload}

  # Your application's action. It must be safe to run twice for the same
  # event: the ledger deduplicates deliveries, but a crash between this
  # call and the ledger update re-runs it on redelivery.
  @impl true
  def handle_event(type, payload) do
    Agent.update(Tour.Handled, &[payload["id"] | &1])
    {:ok, %Tour.VerifiedEvent{type: type, payload: payload}}
  end
end

# 2. A structured header carrying a timestamp and one or more signatures:
#    x-tour-payments: t=1731312000,v1=<hex>,v1=<hex>
#    The signed input is "timestamp.body"; any listed v1 that verifies is
#    accepted, so the sender can rotate keys.
defmodule Tour.Payments do
  use AshHooks.Provider,
    verify: [
      algorithm: :hmac_sha256,
      signature: {:structured_header, "x-tour-payments", signature_key: "v1"},
      timestamp: {:signature_field, "t"},
      signed_input: [:timestamp, ".", :body],
      tolerance_seconds: 300
    ]
end

# 3. A public-key signature: Ed25519 over the raw body, base64 in a header.
#    The secret source returns the sender's public key.
defmodule Tour.Partner do
  use AshHooks.Provider,
    verify: [
      algorithm: :ed25519,
      signature: {:header, "x-partner-signature"},
      encoding: :base64,
      key_encoding: :base64
    ]
end

# 4. The whole body is a compact JWS (an RS256 JWT). The algorithm is pinned
#    here and never read from the token; the verified payload is what the
#    ledger stores.
defmodule Tour.Envelope do
  use AshHooks.Provider,
    verify: [
      algorithm: :rsa_pkcs1_sha256,
      signature: :jws_body
    ]
end

# 5. A sender that canonicalizes before signing: HMAC-SHA1 over the public
#    URL followed by every form field's name and value, sorted by name. A host
#    function builds that string; returning anything but a binary fails.
defmodule Tour.Sms do
  use AshHooks.Provider,
    verify: [
      algorithm: :hmac_sha1,
      signature: {:header, "x-sms-signature"},
      encoding: :base64,
      signed_input: {__MODULE__, :signed_input}
    ]

  def signed_input(raw_body, %{request_uri: url}) when is_binary(url) do
    fields =
      raw_body
      |> URI.decode_query()
      |> Enum.sort()
      |> Enum.map(fn {name, value} -> name <> value end)

    url <> Enum.join(fields)
  end

  def signed_input(_raw_body, _ctx), do: nil
end

defmodule Tour.Ledger do
  use Ash.Resource,
    domain: Tour.Domain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks, AshHooks.InboundDelivery]

  sqlite do
    table("tour_ledgers")
    repo(Tour.Repo)
  end

  inbound_delivery do
    scope_identity([:account_id])
  end

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

  actions do
    defaults([:read])
  end

  # One declaration per source. `provider` is required; the name identifies
  # the source in the ledger. Secrets are sources, never literals.
  webhooks do
    inbound :orders do
      provider(Tour.Orders)
      secret fn -> {:ok, Application.fetch_env!(:ash_hooks, :tour_orders_secret)} end
      # the outgoing key during a rotation: both keys verify until you remove it
      previous_secret fn ->
        {:ok, Application.fetch_env!(:ash_hooks, :tour_orders_previous_secret)}
      end

      # the sender's event id is the deduplication identity
      event_id(&__MODULE__.event_id/1)
    end

    inbound :payments do
      provider(Tour.Payments)
      secret fn -> {:ok, Application.fetch_env!(:ash_hooks, :tour_payments_secret)} end
      # overrides the provider's tolerance_seconds for this source
      replay_window_seconds(120)
    end

    inbound :partner do
      provider(Tour.Partner)
      secret fn -> {:ok, Application.fetch_env!(:ash_hooks, :tour_partner_public_key)} end
    end

    inbound :envelope do
      provider(Tour.Envelope)
      secret fn -> {:ok, Application.fetch_env!(:ash_hooks, :tour_envelope_public_key)} end
    end

    inbound :sms do
      provider(Tour.Sms)
      body_format(:form)
      secret fn -> {:ok, Application.fetch_env!(:ash_hooks, :tour_sms_secret)} end
    end
  end

  def event_id(%{"id" => id}) when is_binary(id), do: {:ok, id}
  def event_id(_payload), do: :error
end

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

  sqlite do
    table("tour_endpoints")
    repo(Tour.Repo)
  end

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

  endpoint do
    # per-endpoint response capture: the four columns exist only when
    # declared (a migration is required to adopt them)
    response_capture_attributes(:on)
  end
end

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

  sqlite do
    table("tour_subscriptions")
    repo(Tour.Repo)
  end

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

  subscription do
    endpoint_resource(Tour.Endpoint)
  end
end

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

  sqlite do
    table("tour_deliveries")
    repo(Tour.Repo)
  end

  attributes do
    timestamps()
  end

  actions do
    defaults([:read])
  end

  outbound_delivery do
    # resend/replay bookkeeping (migrate the four columns when you adopt)
    resend_tracking(:on)
  end
end

defmodule Tour.Attempt do
  # The attempt history: one durable row per delivery attempt (time,
  # outcome, classified reason, HTTP status, latency, the signing key
  # REFERENCE). Written by the delivery runtime when the worker config
  # carries `attempts:` — the audit trail behind delivery logs and health.
  use Ash.Resource,
    domain: Tour.Domain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks.OutboundAttempt]

  sqlite do
    table("tour_attempts")
    repo(Tour.Repo)
  end

  actions do
    defaults([:read])
  end
end

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

  sqlite do
    table("tour_orders")
    repo(Tour.Repo)
  end

  attributes do
    uuid_primary_key(:id)
  end

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

  webhooks do
    outbound :order_paid do
      subscriptions(Tour.Subscription)
      deliveries(Tour.Delivery)
    end
  end
end

defmodule Tour.Domain do
  use Ash.Domain, otp_app: nil, validate_config_inclusion?: false

  resources do
    resource(Tour.Ledger)
    resource(Tour.Endpoint)
    resource(Tour.Subscription)
    resource(Tour.Delivery)
    resource(Tour.Attempt)
    resource(Tour.Order)
    resource(Tour.TenantEndpoint)
    resource(Tour.TenantSubscription)
    resource(Tour.TenantDelivery)
    resource(Tour.TenantOrder)
  end
end

Application.put_env(:ash_hooks, Tour.Repo,
  database: Path.join(System.tmp_dir!(), "ash_hooks_tour_#{Ash.UUID.generate()}.sqlite3"),
  pool_size: 1
)

{:ok, _} = Tour.Repo.start_link()

The tables

Tour.Repo.query!("""
CREATE TABLE tour_ledgers (
  id TEXT PRIMARY KEY, provider TEXT NOT NULL, external_event_id TEXT NOT NULL,
  external_event_type TEXT, payload TEXT NOT NULL, payload_digest TEXT NOT NULL,
  status TEXT NOT NULL, fencing_token INTEGER NOT NULL DEFAULT 0,
  lease_expires_at TEXT, error_class TEXT, attempts INTEGER NOT NULL DEFAULT 0,
  account_id TEXT NOT NULL, inserted_at TEXT, updated_at TEXT)
""")

Tour.Repo.query!(
  "CREATE UNIQUE INDEX tour_ledgers_unique ON tour_ledgers (provider, external_event_id, account_id)"
)

Tour.Repo.query!("""
CREATE TABLE tour_endpoints (
  id TEXT PRIMARY KEY, url TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'enabled',
  secret_ref TEXT NOT NULL, previous_secret_ref TEXT, legacy_secret_ref TEXT,
  legacy_previous_secret_ref TEXT,
  response_capture INTEGER NOT NULL DEFAULT 0,
  response_capture_max_bytes INTEGER NOT NULL DEFAULT 2048,
  response_capture_retention_seconds INTEGER NOT NULL DEFAULT 2592000,
  response_header_allowlist TEXT)
""")

Tour.Repo.query!("""
CREATE TABLE tour_subscriptions (
  id TEXT PRIMARY KEY, event_types TEXT NOT NULL, endpoint_id TEXT NOT NULL, signing_mode TEXT)
""")

Tour.Repo.query!("""
CREATE TABLE tour_deliveries (
  id TEXT PRIMARY KEY, event_uuid TEXT NOT NULL, event_type TEXT NOT NULL,
  payload BLOB NOT NULL, endpoint_id TEXT NOT NULL, subscription_id TEXT,
  signing_mode TEXT, status TEXT NOT NULL DEFAULT 'pending',
  attempts INTEGER NOT NULL DEFAULT 0, response_status INTEGER,
  response_snippet TEXT, last_error TEXT, next_attempt_at TEXT,
  dispatch_source TEXT NOT NULL DEFAULT 'v1:direct:unbound',
  dispatch_route TEXT NOT NULL DEFAULT 'v1:route:unbound',
  attempt_token TEXT, send_lease_expires_at TEXT,
  enqueue_token TEXT, enqueue_lease_expires_at TEXT,
  endpoint_snapshot JSONB,
  resend_baseline INTEGER NOT NULL DEFAULT 0,
  resend_count INTEGER NOT NULL DEFAULT 0,
  last_resend_by TEXT, last_resend_at TEXT,
  inserted_at TEXT, updated_at TEXT)
""")

Tour.Repo.query!(
  "CREATE UNIQUE INDEX tour_deliveries_unique ON tour_deliveries (endpoint_id, event_uuid)"
)

Tour.Repo.query!("""
CREATE TABLE tour_attempts (
  id TEXT PRIMARY KEY, delivery_pk TEXT NOT NULL, endpoint_id TEXT NOT NULL,
  event_uuid TEXT NOT NULL, event_type TEXT NOT NULL, attempt INTEGER NOT NULL,
  outcome TEXT NOT NULL, reason TEXT, response_status INTEGER,
  response_snippet TEXT, response_headers TEXT, latency_ms INTEGER,
  signing_key_ref TEXT, test INTEGER NOT NULL DEFAULT 0,
  attempted_at TEXT NOT NULL)
""")

Tour.Repo.query!(
  "CREATE UNIQUE INDEX tour_attempts_unique ON tour_attempts (delivery_pk, attempt)"
)

Tour.Repo.query!("CREATE TABLE IF NOT EXISTS tour_orders (id TEXT PRIMARY KEY)")

:ok

Observe everything: telemetry

# exact event names — :telemetry.execute/3 matches exact names only
:telemetry.attach_many(
  "tour",
  [
    [: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, _ ->
    IO.puts("EVENT #{inspect(event)} #{inspect(metadata)}")
  end,
  nil
)

:ok

Inbound: verify, deduplicate, reject a tamper

Your controller passes AshHooks.Ingress.ingest/4 the exact raw body and the request headers (in Phoenix, conn.private[:ash_hooks_raw_body] and conn.req_headers, which keeps every header line). One call verifies the signature, stores the payload, deduplicates, claims the row under a lease, runs your handler, and records the outcome.

# A record of what the orders handler ran, to show it runs once per event.
{:ok, _} = Agent.start_link(fn -> [] end, name: Tour.Handled)

# The sender's side: an HMAC-SHA256 of the exact bytes, lowercase hex.
sign_orders = fn body, key ->
  :crypto.mac(:hmac, :sha256, key, body) |> Base.encode16(case: :lower)
end

orders_secret = Application.fetch_env!(:ash_hooks, :tour_orders_secret)
body = Jason.encode!(%{"id" => "evt-tour-1", "type" => "order.paid"})
scope = %{"account_id" => "acct-42"}
ctx = %{headers: %{"x-tour-signature" => sign_orders.(body, orders_secret)}, scope: scope}

# First delivery: verified, persisted, claimed, handled, marked processed.
{:ok, :created, first} = AshHooks.Ingress.ingest(Tour.Ledger, :orders, body, ctx)
IO.inspect({first.status, first.external_event_type}, label: "first delivery")
:processed = first.status

# The same delivery again: the unique index answers. It is a duplicate, and
# the handler does not run a second time.
{:ok, :duplicate, _} = AshHooks.Ingress.ingest(Tour.Ledger, :orders, body, ctx)
["evt-tour-1"] = Agent.get(Tour.Handled, & &1)
IO.puts("duplicate recognized; the handler ran once")

# One changed byte in the body: rejected before anything is stored.
{:error, %{__struct__: AshHooks.Errors.Invalid.InvalidSignature}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :orders, body <> " ", ctx)

# No signature header at all: rejected the same way, never a crash.
{:error, %{__struct__: AshHooks.Errors.Invalid.InvalidSignature}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :orders, body, %{headers: %{}, scope: scope})

IO.puts("tampered and unsigned requests rejected")

Event types your handler does not know

unknown = Jason.encode!(%{"id" => "evt-tour-2", "type" => "order.cancelled"})
unknown_ctx = %{headers: %{"x-tour-signature" => sign_orders.(unknown, orders_secret)}, scope: scope}

# A verified request with an unknown type is recorded as a permanent
# failure, so the sender is acknowledged and does not retry forever. Put
# these rows on your operator surface.
{:ok, :created, unknown_row} = AshHooks.Ingress.ingest(Tour.Ledger, :orders, unknown, unknown_ctx)
IO.inspect({unknown_row.status, unknown_row.error_class}, label: "unknown type")
{:failed_permanent, "unknown_event_type"} = {unknown_row.status, unknown_row.error_class}

Rotate a secret without dropping deliveries

# The sender still signs with the outgoing key. Because the declaration
# carries `previous_secret`, the request verifies with either key until you
# remove that option.
previous = Application.fetch_env!(:ash_hooks, :tour_orders_previous_secret)
rotated = Jason.encode!(%{"id" => "evt-tour-3", "type" => "order.paid"})
rotated_ctx = %{headers: %{"x-tour-signature" => sign_orders.(rotated, previous)}, scope: scope}

{:ok, :created, %{status: :processed}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :orders, rotated, rotated_ctx)

IO.puts("a request signed with the previous key verified during the overlap")

A timestamp and a replay window

payments_secret = Application.fetch_env!(:ash_hooks, :tour_payments_secret)

payments_header = fn body, timestamp, key ->
  mac = :crypto.mac(:hmac, :sha256, key, "#{timestamp}.#{body}") |> Base.encode16(case: :lower)
  # a second v1 from a key the receiver does not have: any match verifies
  stale_key_mac = String.duplicate("0", 64)
  "t=#{timestamp},v1=#{stale_key_mac},v1=#{mac}"
end

charge = Jason.encode!(%{"type" => "charge.succeeded", "amount" => 1200})
now = System.system_time(:second)

# Headers can also arrive as an ordered list, the shape of conn.req_headers.
# Providers then see every line in ctx.header_list, and repeated lines are
# joined with ", " in ctx.headers.
fresh_ctx = %{
  headers: [
    {"content-type", "application/json"},
    {"x-tour-payments", payments_header.(charge, now, payments_secret)}
  ],
  scope: scope
}
{:ok, :created, payment} = AshHooks.Ingress.ingest(Tour.Ledger, :payments, charge, fresh_ctx)

# Configured providers classify every verified request as :webhook and keep
# the sender's event name in the payload.
IO.inspect({payment.status, payment.external_event_type, payment.payload["type"]}, label: "payment")

# A correctly signed request from an hour ago is outside the 120-second
# window declared on this source: a replay, refused.
old = now - 3600
old_ctx = %{headers: [{"x-tour-payments", payments_header.(charge, old, payments_secret)}], scope: scope}

{:error, %{__struct__: AshHooks.Errors.Invalid.StaleTimestamp}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :payments, charge, old_ctx)

# A fresh timestamp never rescues a bad signature: that is still an invalid signature.
forged_ctx = %{headers: %{"x-tour-payments" => "t=#{now},v1=#{String.duplicate("a", 64)}"}, scope: scope}

{:error, %{__struct__: AshHooks.Errors.Invalid.InvalidSignature}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :payments, charge, forged_ctx)

IO.puts("stale timestamp refused; forged signature refused")

A public-key signature

shipment = Jason.encode!(%{"shipment" => "shp_1", "status" => "delivered"})
partner_signature = :crypto.sign(:eddsa, :none, shipment, [partner_private, :ed25519])

partner_ctx = %{headers: %{"x-partner-signature" => Base.encode64(partner_signature)}, scope: scope}
{:ok, :created, %{status: :processed}} = AshHooks.Ingress.ingest(Tour.Ledger, :partner, shipment, partner_ctx)

# A signature from any other key fails: only the configured public key verifies.
{_other_public, other_private} = :crypto.generate_key(:eddsa, :ed25519)
imposter = :crypto.sign(:eddsa, :none, shipment, [other_private, :ed25519])
imposter_ctx = %{headers: %{"x-partner-signature" => Base.encode64(imposter)}, scope: scope}

{:error, %{__struct__: AshHooks.Errors.Invalid.InvalidSignature}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :partner, shipment, imposter_ctx)

IO.puts("Ed25519: the sender's key verifies, another key does not")

A signed JWS body

b64url = &Base.url_encode64(&1, padding: false)

jws = fn header, claims ->
  input = b64url.(Jason.encode!(header)) <> "." <> b64url.(Jason.encode!(claims))
  input <> "." <> b64url.(:public_key.sign(input, :sha256, envelope_private))
end

claims = %{"event" => "member.joined", "member" => "m_7", "exp" => System.system_time(:second) + 300}
token = jws.(%{"alg" => "RS256", "typ" => "JWT"}, claims)

# No signature header: the body is the signed envelope. The ledger stores the
# verified claims.
{:ok, :created, envelope} = AshHooks.Ingress.ingest(Tour.Ledger, :envelope, token, %{headers: %{}, scope: scope})
IO.inspect(envelope.payload, label: "stored JWS payload")
"member.joined" = envelope.payload["event"]

# A token that names another algorithm is rejected, whatever its signature:
# the algorithm comes from your configuration, never from the token.
unsigned = b64url.(Jason.encode!(%{"alg" => "none"})) <> "." <> b64url.(Jason.encode!(claims)) <> "."

{:error, %{__struct__: AshHooks.Errors.Invalid.InvalidSignature}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :envelope, unsigned, %{headers: %{}, scope: scope})

# A validly signed token past its exp claim is refused as stale.
expired = jws.(%{"alg" => "RS256"}, %{claims | "exp" => System.system_time(:second) - 60})

{:error, %{__struct__: AshHooks.Errors.Invalid.StaleTimestamp}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :envelope, expired, %{headers: %{}, scope: scope})

IO.puts("JWS: pinned algorithm, alg none refused, exp enforced")

A sender that canonicalizes form fields

# This sender posts a form and signs the public URL it called plus the
# sorted fields. Pass that URL as request_uri: behind a proxy, build it from
# a base URL you configure, never from forwarded headers.
url = "https://hooks.example.test/sms?account=acct-42"
form = "To=%2B15550100&From=%2B15550199&Body=hello"
sms_secret = Application.fetch_env!(:ash_hooks, :tour_sms_secret)
signed = url <> "Bodyhello" <> "From+15550199" <> "To+15550100"
sms_signature = :crypto.mac(:hmac, :sha, sms_secret, signed) |> Base.encode64()

sms_ctx = %{headers: %{"x-sms-signature" => sms_signature}, request_uri: url, scope: scope}
{:ok, :created, sms} = AshHooks.Ingress.ingest(Tour.Ledger, :sms, form, sms_ctx)

# body_format :form decoded the verified bytes into the stored payload.
IO.inspect(sms.payload, label: "stored form payload")
"hello" = sms.payload["Body"]

# The same bytes posted to a different URL do not verify.
{:error, %{__struct__: AshHooks.Errors.Invalid.InvalidSignature}} =
  AshHooks.Ingress.ingest(Tour.Ledger, :sms, form, %{sms_ctx | request_uri: "https://evil.example.test/sms"})

What to answer the sender

Verify without a ledger

# Every provider's verify_signature/3 can be called directly, for example in
# your own tests. Direct calls may pin the clock with :now_ms.
:ok = Tour.Orders.verify_signature(body, %{headers: ctx.headers}, orders_secret)

# The configuration also works without a module...
:ok =
  AshHooks.Provider.Scheme.verify(body, %{headers: ctx.headers}, orders_secret,
    algorithm: :hmac_sha256,
    signature: {:header, "x-tour-signature"}
  )

# ...and the primitives underneath are public, for schemes you implement yourself.
:ok = AshHooks.Provider.Signature.hmac(body, ctx.headers["x-tour-signature"], orders_secret, digest: :sha256)
:ok = AshHooks.Provider.Signature.ed25519(shipment, Base.encode64(partner_signature), [partner_public])
{:ok, _claims_json} = AshHooks.Provider.Signature.jws(token, [envelope_public_pem], alg: "RS256")

# A list of keys verifies with any of them: rotation for direct calls.
:ok = AshHooks.Provider.Signature.hmac(rotated, rotated_ctx.headers["x-tour-signature"], [orders_secret, previous])
IO.puts("direct verification agrees with ingress")

# Options are checked when the provider compiles. A typo is a compile error
# that names the problem, not a provider that silently rejects everything.
try do
  Code.compile_string("""
  defmodule Tour.Misconfigured do
    use AshHooks.Provider,
      verify: [algorithm: :hmac_sha256, signature: {:header, "x-sig"}, encoding: :hexadecimal]
  end
  """)
rescue
  error -> IO.puts("compile error: " <> Exception.message(error))
else
  _ -> raise "an invalid encoding must not compile"
end

Outbound: deliver to the real receiver

port = String.to_integer(System.get_env("LOCAL_WEBHOOK_TESTER_PORT", "52871"))
receiver_base = "http://127.0.0.1:#{port}"
receiver_session = Ash.UUID.generate()
{:ok, %{status: 200}} = AshHooks.Http.Bounded.request(
  :get, receiver_base <> "/api/version", %{}, "", validate_destination: false
)
IO.puts("using the existing receiver on #{port}")

# Registration rejects private/loopback URLs by design — ash_hooks
# treats webhook destinations as untrusted input. This demo needs a
# trusted local receiver, so we register the endpoint with a public URL and
# repoint the stored row for this local inspection; production
# endpoints keep their public URLs.
endpoint =
  Ash.create!(
    Tour.Endpoint,
    %{
      url: "https://partner.example.test/hook",
      secret_ref: "tour-endpoint",
      # this endpoint opts IN to response capture per row (bounded bytes,
      # redaction floor, allowlisted headers only)
      response_capture: true,
      response_header_allowlist: ["content-type"]
    },
    authorize?: false
  )

Tour.Repo.query!("UPDATE tour_endpoints SET url = ? WHERE id = ?", [
  receiver_base <> "/" <> receiver_session,
  endpoint.id
])

Ash.create!(Tour.Subscription, %{endpoint_id: endpoint.id, event_types: ["order_paid"]},
  authorize?: false
)

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

# No enqueue configured: dispatch persists the durable rows only
# (the deferred pattern — an Oban-backed app wires enqueue: here)
{:ok, results} = AshHooks.dispatch(Tour.Order, :order_paid, event)
IO.inspect(Enum.map(results, & &1.status), label: "dispatch results")

# Drive the delivery machine directly (the shape an Oban worker uses).
# Two options permit this trusted loopback receiver: `ssrf_check` replaces the
# send-time destination re-check, and http_opts' `validate_destination:
# false` relaxes the HTTP adapter's own pin — both exist for tests and
# diagnostics; production configs keep both guards on.
outbound_secret = AshHooks.Signing.generate_secret()
config = [
  deliveries: Tour.Delivery,
  endpoints: Tour.Endpoint,
  attempts: Tour.Attempt,
  secret_resolver: fn "tour-endpoint" ->
    {:ok, outbound_secret}
  end,
  max_attempts: 3,
  base_backoff_seconds: 1,
  max_backoff_seconds: 60,
  retry_after_cap_seconds: 100,
  ssrf_check: fn url -> url == receiver_base <> "/" <> receiver_session end,
  http_opts: [validate_destination: false],
  now: fn -> DateTime.utc_now() |> DateTime.truncate(:second) end
]

[row] = Ash.read!(Tour.Delivery, authorize?: false)

:ok =
  AshHooks.Delivery.run(%{"endpoint_id" => row.endpoint_id, "event_uuid" => row.event_uuid}, config)

final = Ash.reload!(row, authorize?: false)
{:succeeded, 200} = {final.status, final.response_status}
IO.inspect({final.status, final.response_status, final.response_snippet}, label: "delivered")

{:ok, %{status: 200, body: captures}} = AshHooks.Http.Bounded.request(
  :get, receiver_base <> "/api/session/" <> receiver_session <> "/requests", %{}, "",
  validate_destination: false, max_body_bytes: 1_048_576
)
matching_captures = Enum.filter(Jason.decode!(captures), fn request ->
  Enum.any?(request["headers"], &(&1["name"] |> String.downcase() == "webhook-id" and &1["value"] == event.id))
end)
[capture] = matching_captures
captured_bytes = Base.decode64!(capture["request_payload_base64"])
true = captured_bytes == event.payload
captured_headers = Map.new(capture["headers"], &{String.downcase(&1["name"]), &1["value"]})
{:ok, %{id: verified_id}} = AshHooks.Signing.verify(captured_bytes, captured_headers, outbound_secret)
true = verified_id == event.id
IO.puts("receiver retained exact bytes and a valid signature")

# A completed row is harmless to drive again: no second network request.
:ok = AshHooks.Delivery.run(%{"endpoint_id" => row.endpoint_id, "event_uuid" => row.event_uuid}, config)
{:ok, %{status: 200, body: captures_after}} = AshHooks.Http.Bounded.request(
  :get, receiver_base <> "/api/session/" <> receiver_session <> "/requests", %{}, "",
  validate_destination: false, max_body_bytes: 1_048_576
)
1 = length(Jason.decode!(captures_after))
IO.puts("terminal redrive retained exactly one receiver request")

# One attempt row was recorded beside the delivery: outcome, HTTP status,
# latency, the signing key REFERENCE — and, because this endpoint opted in
# per row, the redacted body and only allowlisted response headers.
[attempt] = Ash.read!(Tour.Attempt, authorize?: false)
:succeeded = attempt.outcome
200 = attempt.response_status
true = is_integer(attempt.latency_ms) and attempt.latency_ms >= 0
"tour-endpoint" = attempt.signing_key_ref
true = String.starts_with?(attempt.response_snippet, "[captured]")
# only allowlisted names survive when the receiver sends any (this one
# answers empty-bodied: no headers to keep)
true = attempt.response_headers == nil or is_map(attempt.response_headers)
IO.puts("attempt history recorded the send")

# The delivery log answers from the rows, keyset-paginated.
{:ok, %{records: [logged], next_cursor: nil}} =
  AshHooks.Delivery.list(Tour.Delivery, endpoint_id: endpoint.id)

true = logged.id == row.id
{:ok, %{records: [history], next_cursor: nil}} =
  AshHooks.OutboundAttempt.list(Tour.Attempt, endpoint_id: endpoint.id)

:succeeded = history.outcome
IO.puts("delivery log and attempt history answer from the rows")

IO.puts("""
Response bodies stay off by default everywhere: without the endpoint's
per-row opt-in, attempt rows carry the status + content-type summary only
(the delivery row's response_snippet stays that summary in all cases).
""")

Resend and replay: the same event id, again

# A resend re-arms the TERMINAL row: same webhook-id (the receiver
# deduplicates), a fresh retry ladder, and who asked recorded on the row.
{:ok, %{status: :resent, delivery: re_armed}} =
  AshHooks.Delivery.resend(final,
    requested_by: "tour-operator",
    endpoints: Tour.Endpoint
  )

:pending = re_armed.status
0 = re_armed.attempts - re_armed.resend_baseline
"tour-operator" = re_armed.last_resend_by
1 = re_armed.resend_count

# Bulk replay counts before it touches anything (the re-armed row is
# pending — that is the replays-eligible set here)
{:ok, %{count: 1}} =
  AshHooks.Delivery.replay(Tour.Delivery,
    endpoints: Tour.Endpoint,
    status: [:pending],
    requested_by: "tour-operator",
    dry_run: true
  )

# ...and the second sweep re-arms nothing more (the only row is pending,
# not terminal — live rows are left to their schedule)
{:ok, %{resent: 0, skipped: 0}} =
  AshHooks.Delivery.replay(Tour.Delivery,
    endpoints: Tour.Endpoint,
    status: [:dead_letter],
    requested_by: "tour-operator",
    per_second: 10
  )

IO.puts("resend recorded who asked; replay counted and rate-limited")

Multi-tenancy: one tenant's event, one tenant's rows

# The same machine, scoped by an explicit tenant: declare attribute
# multitenancy on the touched resources (in your app: the subscription,
# the endpoint, and both ledgers) and pass :tenant to every call. Ash
# scopes every query and write to the tenant; the package additionally
# fails closed with a named error when a multitenant resource is touched
# without one.
defmodule Tour.TenantEndpoint do
  use Ash.Resource,
    domain: Tour.Domain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks.Endpoint]

  sqlite do
    table("tour_tenant_endpoints")
    repo(Tour.Repo)
  end

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

  multitenancy do
    strategy(:attribute)
    attribute(:org_id)
  end

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

defmodule Tour.TenantSubscription do
  use Ash.Resource,
    domain: Tour.Domain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks.Subscription]

  sqlite do
    table("tour_tenant_subscriptions")
    repo(Tour.Repo)
  end

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

  multitenancy do
    strategy(:attribute)
    attribute(:org_id)
  end

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

  subscription do
    endpoint_resource(Tour.TenantEndpoint)
  end
end

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

  sqlite do
    table("tour_tenant_deliveries")
    repo(Tour.Repo)
  end

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

  multitenancy do
    strategy(:attribute)
    attribute(:org_id)
  end

  actions do
    defaults([:read])
  end
end

defmodule Tour.TenantOrder do
  use Ash.Resource,
    domain: Tour.Domain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks]

  sqlite do
    table("tour_tenant_orders")
    repo(Tour.Repo)
  end

  attributes do
    uuid_primary_key(:id)
  end

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

  webhooks do
    outbound :order_paid do
      subscriptions(Tour.TenantSubscription)
      deliveries(Tour.TenantDelivery)
    end
  end
end

# Tenant-bearing tables: the multitenancy attribute prefixes the unique
# identity — "have I seen this event for this endpoint" is per-tenant
Tour.Repo.query!("""
CREATE TABLE tour_tenant_endpoints (
  id TEXT PRIMARY KEY, url TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'enabled',
  secret_ref TEXT NOT NULL, previous_secret_ref TEXT, legacy_secret_ref TEXT,
  legacy_previous_secret_ref TEXT, org_id TEXT NOT NULL)
""")

Tour.Repo.query!("""
CREATE TABLE tour_tenant_subscriptions (
  id TEXT PRIMARY KEY, event_types TEXT NOT NULL, endpoint_id TEXT NOT NULL,
  signing_mode TEXT, org_id TEXT NOT NULL)
""")

Tour.Repo.query!("""
CREATE TABLE tour_tenant_deliveries (
  id TEXT PRIMARY KEY, event_uuid TEXT NOT NULL, event_type TEXT NOT NULL,
  payload BLOB NOT NULL, endpoint_id TEXT NOT NULL, subscription_id TEXT,
  signing_mode TEXT, status TEXT NOT NULL DEFAULT 'pending',
  attempts INTEGER NOT NULL DEFAULT 0, response_status INTEGER,
  response_snippet TEXT, last_error TEXT, next_attempt_at TEXT,
  dispatch_source TEXT NOT NULL DEFAULT 'v1:direct:unbound',
  dispatch_route TEXT NOT NULL DEFAULT 'v1:route:unbound',
  attempt_token TEXT, send_lease_expires_at TEXT,
  enqueue_token TEXT, enqueue_lease_expires_at TEXT,
  endpoint_snapshot JSONB,
  org_id TEXT NOT NULL, inserted_at TEXT, updated_at TEXT)
""")

Tour.Repo.query!(
  "CREATE UNIQUE INDEX tour_tenant_deliveries_unique ON tour_tenant_deliveries (org_id, endpoint_id, event_uuid)"
)

Tour.Repo.query!("CREATE TABLE IF NOT EXISTS tour_tenant_orders (id TEXT PRIMARY KEY)")
:ok

# Two tenants, each with an endpoint subscribed to order_paid
for org <- ["org_b", "org_a"] do
  endpoint =
    Ash.create!(Tour.TenantEndpoint, %{url: "https://#{org}.example.test/hook", secret_ref: "ref-#{org}"},
      tenant: org,
      authorize?: false
    )

  Ash.create!(Tour.TenantSubscription, %{endpoint_id: endpoint.id, event_types: ["order_paid"]},
    tenant: org,
    authorize?: false
  )
end

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

# org_a dispatches: the fanout read itself is tenant-scoped — org_b's
# subscription is invisible, so no row, no signature, no bytes for org_b
{:ok, [result]} = AshHooks.dispatch(Tour.TenantOrder, :order_paid, event, tenant: "org_a")
IO.inspect(result.status, label: "org_a dispatch result")

[a_row] = Ash.read!(Tour.TenantDelivery, authorize?: false, tenant: "org_a")
IO.inspect({a_row.org_id, a_row.status}, label: "org_a's row")

[] = Ash.read!(Tour.TenantDelivery, authorize?: false, tenant: "org_b")
IO.puts("org_b's partition: empty — the event never existed for org_b")

# A tenant-less call against a multitenant set fails closed before any
# data access — the named error, not a crash:
{:error, :tenant_required} = AshHooks.dispatch(Tour.TenantOrder, :order_paid, event)
IO.puts("tenant-less dispatch: {:error, :tenant_required}")

Retention: prune the terminal rows

# Nothing is older than 30 days — prune finds nothing yet
{:ok, 0} = AshHooks.Ingress.prune(Tour.Ledger, older_than: DateTime.add(DateTime.utc_now(), -30, :day))

# Backdate the processed rows past a 30-day TTL and prune again
%{rows: [[processed]]} =
  Tour.Repo.query!("SELECT COUNT(*) FROM tour_ledgers WHERE status = 'processed'")

Tour.Repo.query!("UPDATE tour_ledgers SET inserted_at = ? WHERE status = 'processed'", [
  DateTime.add(DateTime.utc_now(), -31, :day)
  |> DateTime.truncate(:microsecond)
  |> DateTime.to_iso8601()
])

{:ok, ^processed} =
  AshHooks.Ingress.prune(Tour.Ledger, older_than: DateTime.add(DateTime.utc_now(), -30, :day))

IO.puts("Pruned #{processed} terminal rows past the TTL; recent and non-terminal rows are never touched.")

Fit the extensions to your domain

# A consumer whose domain reserves `payload` for its own sole store,
# owns an `active` switch on the endpoint register, types its
# subscription register as a closed atom enum, keys rows with
# uuid_v7, and keeps the ledger append-only.
defmodule Tour.FitEndpoint do
  use Ash.Resource,
    domain: Tour.FitDomain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks.Endpoint]

  sqlite do
    table("tour_fit_endpoints")
    repo(Tour.Repo)
  end

  attributes do
    uuid_v7_primary_key(:id)
    # the consumer's own switch — the package injects no `status`
    attribute(:active, :boolean, allow_nil?: false, default: true, public?: true)
  end

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

  endpoint do
    status_attribute(:active)
    enabled_values([true])
    disabled_value(false)
  end
end

defmodule Tour.FitSubscription do
  use Ash.Resource,
    domain: Tour.FitDomain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks.Subscription]

  sqlite do
    table("tour_fit_subscriptions")
    repo(Tour.Repo)
  end

  attributes do
    uuid_v7_primary_key(:id)

    # the consumer's closed atom register replaces the injected
    # {:array, :string} default ["*"] — matching handles atoms
    attribute(:event_types, {:array, :atom},
      allow_nil?: false,
      public?: true,
      constraints: [min_length: 1, items: [one_of: [:order_paid, :order_shipped]]]
    )
  end

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

  subscription do
    endpoint_resource(Tour.FitEndpoint)
  end
end

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

  sqlite do
    table("tour_fit_deliveries")
    repo(Tour.Repo)
  end

  attributes do
    uuid_v7_primary_key(:id)
  end

  actions do
    defaults([:read])
  end

  outbound_delivery do
    # the domain reserves `payload` — the exact-bytes column is renamed
    payload_attribute(:event_bytes)
    # append-only ledger: no destroy action is injected at all
    prune_action(:none)
  end
end

defmodule Tour.FitOrder do
  use Ash.Resource,
    domain: Tour.FitDomain,
    data_layer: AshSqlite.DataLayer,
    extensions: [AshHooks]

  sqlite do
    table("tour_fit_orders")
    repo(Tour.Repo)
  end

  attributes do
    uuid_primary_key(:id)
  end

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

  webhooks do
    outbound :order_paid do
      subscriptions(Tour.FitSubscription)
      deliveries(Tour.FitDelivery)
    end
  end
end

defmodule Tour.FitDomain do
  use Ash.Domain, otp_app: nil, validate_config_inclusion?: false

  resources do
    resource(Tour.FitEndpoint)
    resource(Tour.FitSubscription)
    resource(Tour.FitDelivery)
    resource(Tour.FitOrder)
  end
end

Tour.Repo.query!("""
CREATE TABLE tour_fit_endpoints (
  id TEXT PRIMARY KEY, url TEXT NOT NULL, active BOOLEAN NOT NULL DEFAULT 1,
  secret_ref TEXT NOT NULL, previous_secret_ref TEXT, legacy_secret_ref TEXT,
  legacy_previous_secret_ref TEXT)
""")

Tour.Repo.query!("""
CREATE TABLE tour_fit_subscriptions (
  id TEXT PRIMARY KEY, event_types TEXT NOT NULL, endpoint_id TEXT NOT NULL, signing_mode TEXT)
""")

# NO payload column and no destroy action: the sole-store reservation
# and the append-only posture, in the schema itself
Tour.Repo.query!("""
CREATE TABLE tour_fit_deliveries (
  id TEXT PRIMARY KEY, event_uuid TEXT NOT NULL, event_type TEXT NOT NULL,
  event_bytes BLOB NOT NULL, endpoint_id TEXT NOT NULL, subscription_id TEXT,
  signing_mode TEXT, status TEXT NOT NULL DEFAULT 'pending',
  attempts INTEGER NOT NULL DEFAULT 0, response_status INTEGER,
  response_snippet TEXT, last_error TEXT, next_attempt_at TEXT,
  dispatch_source TEXT NOT NULL DEFAULT 'v1:direct:unbound',
  dispatch_route TEXT NOT NULL DEFAULT 'v1:route:unbound',
  attempt_token TEXT, send_lease_expires_at TEXT,
  enqueue_token TEXT, enqueue_lease_expires_at TEXT,
  endpoint_snapshot JSONB)
""")

Tour.Repo.query!(
  "CREATE UNIQUE INDEX tour_fit_deliveries_unique ON tour_fit_deliveries (endpoint_id, event_uuid)"
)

Tour.Repo.query!("CREATE TABLE IF NOT EXISTS tour_fit_orders (id TEXT PRIMARY KEY)")
:ok

fit_endpoint =
  Ash.create!(
    Tour.FitEndpoint,
    %{url: "https://partner.example.test/hook", secret_ref: "fit-ref", active: true},
    authorize?: false
  )

# the atom-typed register matches the event's canonical string type
Ash.create!(Tour.FitSubscription, %{endpoint_id: fit_endpoint.id, event_types: [:order_paid]},
  authorize?: false
)

# a deterministic ID: the re-fire below matches the existing row
{:ok, fit_event} =
  AshHooks.Event.new(
    type: :order_paid,
    payload: Jason.encode!(%{"order" => 9}),
    id: "msg_fit-order-9"
  )

# Persist a deferred row; repeating the stable event ID returns its duplicate.
{:ok, [first]} = AshHooks.dispatch(Tour.FitOrder, :order_paid, fit_event)
{:ok, [second]} = AshHooks.dispatch(Tour.FitOrder, :order_paid, fit_event)

IO.inspect({first.status, second.status}, label: "deferred, then duplicate")
%{status: :deferred} = first
%{status: :duplicate} = second

[fit_row] = Ash.read!(Tour.FitDelivery, authorize?: false)

# uuid_v7 ids (version nibble 7), bytes under the renamed column
IO.inspect(binary_part(fit_row.id, 14, 1), label: "uuid version nibble")
IO.inspect(fit_row.event_bytes, label: "exact bytes under event_bytes")
"7" = binary_part(fit_row.id, 14, 1)
fit_bytes = Jason.encode!(%{"order" => 9})
^fit_bytes = fit_row.event_bytes

# the consumer's own off switch suppresses the fanout entirely
Ash.update!(fit_endpoint, %{active: false}, authorize?: false)

{:ok, results} = AshHooks.dispatch(Tour.FitOrder, :order_paid, fit_event)
IO.inspect(results, label: "dispatch against a deactivated endpoint")
[] = results

# and the retention hook returns an error on the append-only ledger
{:error, prune_error} =
  AshHooks.Delivery.prune(Tour.FitDelivery, older_than: DateTime.utc_now())

IO.inspect(Exception.message(prune_error), label: "prune on append-only")
IO.puts("""
The resource keeps its own column names, generated keys, typed subscriptions,
enable switch, and append-only retention policy.
""")

Clean up

The tour created one disposable SQLite file. Close its repository and remove only that file and its SQLite sidecars when you finish:

database_path = Application.fetch_env!(:ash_hooks, Tour.Repo)[:database]
:ok = Supervisor.stop(Process.whereis(Tour.Repo))
for suffix <- ["", "-shm", "-wal"], do: File.rm(database_path <> suffix)
for key <- [
      :tour_orders_secret,
      :tour_orders_previous_secret,
      :tour_payments_secret,
      :tour_sms_secret,
      :tour_partner_public_key,
      :tour_envelope_public_key,
      Tour.Repo
    ],
    do: Application.delete_env(:ash_hooks, key)

:ok = Agent.stop(Tour.Handled)
:ok = :telemetry.detach("tour")
IO.puts("Tour database removed; captured receiver requests follow the receiver's retention policy.")

Where to go next

• Get-started tutorial: the full walkthrough with Oban
• Verify any sender: signature recipes and the configuration reference
• Multi-tenant adoption checklist
• DSL reference
• AshHooks.Provider.Scheme, AshHooks.Provider.Signature, AshHooks.Telemetry, AshHooks.Delivery, AshHooks.Ingress module docs