AshHooks.Ingress (AshHooks v0.2.2)

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/2, mark_processed/3, mark_failed/5, renew/3, reap/1) are public for composition and monitoring — the async runtime (#delivery-runtime slice) and the reaper drive the same steps.

All ledger operations run unauthorized: the signature verification IS the trust boundary for writes, and external read surfaces remain governed by the consumer's own policies (ADR-0005).

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.

Functions

claim_delivery(resource, delivery_id)

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}.

ingest(resource, name, raw_body, ctx)

Drives one inbound delivery through the full sync pipeline.

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.

mark_failed(resource, delivery_id, token, error_class, permanent?)

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

mark_processed(resource, delivery_id, token)

Marks a claimed delivery processed — gated on the caller's token under an unexpired lease.

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)

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.

redact_payload(resource, delivery_id, token, redactor)

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)

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()}
}

Functions

claim_delivery(resource, delivery_id)

@spec claim_delivery(module(), term()) ::
  {: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()}

Drives one inbound delivery through the full sync pipeline.

ingest_delivery(resource, env)

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

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.

mark_failed(resource, delivery_id, token, error_class, permanent?)

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

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

mark_processed(resource, delivery_id, token)

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

Marks a claimed delivery processed — gated on the caller's token under an unexpired lease.

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)

@spec reap(module()) :: {:ok, non_neg_integer()}

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.

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 (cross-vendor finding).

redact_payload(resource, delivery_id, token, redactor)

@spec redact_payload(module(), term(), non_neg_integer(), term()) ::
  :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)

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

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