AshHooks — a guided tour

You will learn

• Receive an inbound webhook: verify its signature, deduplicate a replay, reject a tamper
• 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 2.0.4. Its examples exercise the library locally; vendor-origin delivery and a production queue require your own application integration.

Setup

package =
  case System.get_env("ASH_HOOKS_NOTEBOOK_SOURCE") do
    nil -> {:ash_hooks, "== 2.0.4"}
    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)
Application.put_env(:ash_hooks, :tour_inbound_secret, :crypto.strong_rand_bytes(32))
"2.0.4" = Application.spec(:ash_hooks, :vsn) |> to_string()

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

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

# A minimal inbound provider — your integration point for any vendor.
# The scheme here is the common one: lowercase-hex HMAC-SHA256 over the
# raw body, so verify delegates to the package's default.
defmodule Tour.Provider do
  @behaviour AshHooks.Provider

  @impl true
  def verify_signature(raw_body, %{signature: signature}, secret),
    do: AshHooks.Provider.default_verify_signature(raw_body, signature, secret, :hmac_sha256)

  # Map the vendor's type strings to your event atoms — an explicit
  # table, never String.to_atom on provider input (that would let a
  # hostile sender grow your atom table). Unknown types fail closed.
  @impl true
  def parse_event_type(%{"type" => "order.paid"}), do: {:ok, :order_paid}

  def parse_event_type(_), do: {:error, :unknown_event_type}

  @impl true
  def handle_event(type, payload), do: {:ok, %Tour.VerifiedEvent{type: type, payload: payload}}
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

  webhooks do
    inbound :tour do
      provider(Tour.Provider)
      # a secret source — never a literal
      secret fn -> {:ok, Application.fetch_env!(:ash_hooks, :tour_inbound_secret)} end
      event_id(&__MODULE__.event_id/1)
    end
  end

  def event_id(%{"id" => id}), do: {:ok, id}
  def event_id(_), 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
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
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.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)
""")

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,
  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 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

body = Jason.encode!(%{"id" => "evt-tour-1", "type" => "order.paid"})

# our provider signs the raw bytes: lowercase-hex HMAC-SHA256
signature =
  :crypto.mac(:hmac, :sha256, Application.fetch_env!(:ash_hooks, :tour_inbound_secret), body)
  |> Base.encode16(case: :lower)

ctx = %{signature: signature, headers: %{}, scope: %{"account_id" => "acct-42"}}

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

# The exact same delivery again: the unique index answers — duplicate,
# never re-processed
{:ok, :duplicate, _} = AshHooks.Ingress.ingest(Tour.Ledger, :tour, body, ctx)

# A tampered signature: rejected before anything is trusted
<<first, rest::binary>> = signature
tampered = %{ctx | signature: <<Bitwise.bxor(first, 1), rest::binary>>}
{:error, error} = AshHooks.Ingress.ingest(Tour.Ledger, :tour, body, tampered)
IO.inspect(error.__struct__, label: "tamper rejected with")

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"},
    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,
  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")

IO.puts("""
Note the response_snippet: a status + content-type summary — response
bodies are never stored by default (opt-in diagnostic capture exists
under the package's redaction pipeline).
""")

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 row past a 30-day TTL and prune again
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, 1} = AshHooks.Ingress.prune(Tour.Ledger, older_than: DateTime.add(DateTime.utc_now(), -30, :day))
IO.puts("Pruned the terminal row — 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)
Application.delete_env(:ash_hooks, :tour_inbound_secret)
Application.delete_env(:ash_hooks, Tour.Repo)
: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
• Multi-tenant adoption checklist
• DSL reference
• AshHooks.Telemetry, AshHooks.Delivery, AshHooks.Ingress module docs