AshHooks.Ingress (AshHooks v1.2.0)

The inbound pipeline driver: verify → ingest → claim → handle → mark, over a ledger resource carrying the AshHooks.InboundDelivery extension.

ingest/4 is the sync-mode entry point a consumer's controller calls with the raw request body (captured pre-parser via the endpoint body_reader — a router plug cannot recover pre-parse bytes). The machine's crash-safety contract:

the raw payload persists BEFORE any handling (audit + verification integrity via payload_digest);
the unique identity + storage-level uniqueness make exactly one :created per delivery — concurrent or sequential duplicates get :duplicate;
a claim is a WHERE-gated update (status == :received, a re-driveable failure, or an EXPIRED lease) that bumps the monotonic fencing token — so a redelivery of a stranded row re-drives it, never no-ops (the incumbent's silent-loss window);
marks are gated on the caller's token AND an unexpired lease — a stale owner (superseded or expired) is rejected.

Returns {:ok, :created | :duplicate, delivery} — the delivery's status carries the outcome (:processed, :failed_retryable, :failed_permanent); handler failures are recorded in the ledger, not raised. Verification/config failures return {:error, error} BEFORE any ledger write (fail closed: missing secret verifies nothing; a missing raw body never runs).

The individual machine steps (ingest_delivery/2, claim_delivery/3, mark_processed/4, mark_failed/6, renew/4, reap/2) are public for composition and monitoring — a consumer-driven async pipeline (your own 202-then-claim flow) and the reaper drive the same steps. Each accepts the tenancy options; the pre-1.2 arities (one fewer argument) keep working for single-tenant ledgers.

All ledger operations run unauthorized: the signature verification IS the trust boundary for writes. The package injects NO read policies — read access to the ledger resource is governed ENTIRELY by the consumer's own domain policies. The ledger stores raw provider payloads: mount it behind policies that deny reads by default (see README → Security).

Summary

Types

ctx()

The request context. :signature is the provider's signature header value; :headers the lowercased request headers; :scope carries values for the ledger's declared scope_identity slots; :connection is the per-connection provider's secret source argument; :tenant is the ingest tenant (required on multitenant ledgers — the named {:error, :tenant_required} comes back before any data access when absent).

Functions

claim_delivery(resource, delivery_id, opts \\ [])

Claims a delivery: WHERE-gated on status == :received, a re-driveable failure, or an EXPIRED lease — bumps the fencing token atomically and takes a fresh lease. A live foreign lease (or a terminal row) returns {:error, :lease_held}. :tenant scopes the claim (required on multitenant ledgers).

ingest(resource, name, raw_body, ctx)

Drives one inbound delivery through the full sync pipeline. The tenant rides the ctx (ctx[:tenant]) and threads every ledger call — the dedup identity, the claim fence, and every mark are per-tenant.

ingest_delivery(resource, env)

Persists the ledger row (raw payload BEFORE handling) via the no-touch unique upsert. Classifies :created by comparing the surviving row's client-generated id against ours. The env carries the resolved tenant (env[:tenant]).

mark_failed(resource, delivery_id, token, error_class, permanent?, opts \\ [])

Marks a claimed delivery failed — same token/lease gate as mark_processed/4. permanent? selects :failed_permanent over :failed_retryable.

mark_processed(resource, delivery_id, token, opts \\ [])

Marks a claimed delivery processed — gated on the caller's token under an unexpired lease. :tenant scopes the mark (required on multitenant ledgers).

prune(resource, opts)

Retention hook: deletes TERMINAL ledger rows (:processed, :failed_permanent) older than older_than, by the resource's inserted_at (add Ash timestamps() to the resource and its migration). Non-terminal rows are never deleted — retryable and lease-held deliveries keep their dedup identity and re-drive path.

reap(resource, opts \\ [])

Re-drives deliveries whose claims died with an expired lease (crash between claim and mark). Returns the number re-driven; unexpired leases and terminal rows are left alone. :tenant scopes the sweep — a tenant-less global reap over a multitenant ledger is the named {:error, :tenant_required} (this head would otherwise crash on the bang read); per-tenant sweeps compose with AshHooks.reap_all/2.

redact_payload(resource, delivery_id, token, redactor, opts \\ [])

Retention field-redaction hook: replaces the stored payload of a CLAIMED delivery with redactor.(payload) — under the caller's token and an unexpired lease (the mark_processed/3 fence). The row keeps its dedup identity; only the payload changes. payload_digest is deliberately NOT updated (it binds the ORIGINAL signed bytes — the audit trail).

renew(resource, delivery_id, token, opts \\ [])

Extends the caller's lease — gated on the caller's token under an unexpired lease.

Types

ctx()

@type ctx() :: %{
  optional(:signature) => String.t(),
  optional(:headers) => %{optional(String.t()) => String.t()},
  optional(:method) => String.t() | nil,
  optional(:request_uri) => String.t() | nil,
  optional(:connection) => term(),
  optional(:scope) => %{optional(atom() | String.t()) => term()},
  optional(:tenant) => term()
}

Functions

claim_delivery(resource, delivery_id, opts \\ [])

@spec claim_delivery(module(), term(), keyword()) ::
  {:ok, non_neg_integer(), struct()} | {:error, :lease_held | term()}

ingest(resource, name, raw_body, ctx)

@spec ingest(module(), atom(), binary() | nil, ctx()) ::
  {:ok, :created | :duplicate, struct()} | {:error, term()}

ingest_delivery(resource, env)

@spec ingest_delivery(module(), map()) ::
  {:ok, boolean(), struct()} | {:error, term()}

mark_failed(resource, delivery_id, token, error_class, permanent?, opts \\ [])

@spec mark_failed(
  module(),
  term(),
  non_neg_integer(),
  String.t(),
  boolean(),
  keyword()
) ::
  :ok | {:error, :stale_token | term()}

Marks a claimed delivery failed — same token/lease gate as mark_processed/4. permanent? selects :failed_permanent over :failed_retryable.

mark_processed(resource, delivery_id, token, opts \\ [])

@spec mark_processed(module(), term(), non_neg_integer(), keyword()) ::
  :ok | {:error, :stale_token | term()}

Marks a claimed delivery processed — gated on the caller's token under an unexpired lease. :tenant scopes the mark (required on multitenant ledgers).

prune(resource, opts)

@spec prune(module(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()}

Deleting a terminal row re-opens its dedup identity: a replayed delivery of the same webhook processes again (inbound), and — on the outbound side — a re-emission of the same event re-dispatches and double-sends. Set the TTL beyond any replay/re-emission horizon. Returns {:ok, deleted_count}.

reap(resource, opts \\ [])

@spec reap(module(), keyword()) ::
  {:ok, non_neg_integer()} | {:error, :tenant_required | :tenancy_mismatch}

Rows that cannot be re-driven (inbound declaration removed, provider unresolved, a raising handler) are skipped without stopping the sweep — a poison row must never starve the rows behind it.

redact_payload(resource, delivery_id, token, redactor, opts \\ [])

@spec redact_payload(module(), term(), non_neg_integer(), term(), keyword()) ::
  :ok
  | {:error, :stale_token | :redactor_crash | :invalid_redactor_result | term()}

The redactor ({m, f} | 1-arity fun) receives the stored payload (map or list — HubSpot batches are lists) and returns the replacement (same type) or nil to leave it unchanged. A crashing or invalid redactor returns an error and leaves the payload UNCHANGED (fail-safe for the audit record — the caller may retry). Redact, then mark promptly: a lease expiry between redact and mark re-drives the row with the redacted payload.

renew(resource, delivery_id, token, opts \\ [])

@spec renew(module(), term(), non_neg_integer(), keyword()) ::
  :ok | {:error, :stale_token | term()}

Extends the caller's lease — gated on the caller's token under an unexpired lease.