AshHooks.Ingress (AshHooks v2.0.3)

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). Request context carries the signature, lowercased headers, public request method/URI, scope, connection, and tenant. It does not accept or forward a request-controlled clock; replay-window verification uses the runtime system clock. The machine's crash-safety contract:

the decoded payload persists before handling; payload_digest binds it to the signed raw bytes;
the unique index preserves one row per provider/event/scope identity. With the injected writable UUID key, the surviving key distinguishes :created from :duplicate. Custom generated keys use a pre-read; concurrent first arrivals may both be reported as :created, while the stored row and claim remain deduplicated;
a claim is a WHERE-gated update (status == :received, a retryable failure, or an expired lease) that bumps the monotonic fencing token, allowing redelivery to recover a stranded row;
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, or retained :superseded); 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.

Internal ledger operations use authorize?: false; callers must restrict access to these machine functions. Signature verification gates the sync ingest path. Your domain policies govern application read access to decoded provider payloads and signed-body digests. See the README's Security section.

Summary

Types

ctx()

The request context. :signature is the provider's signature header value; :headers contains lowercased request headers; :method and :request_uri must reproduce the public request values covered by the provider signature; :scope carries values for the ledger's declared scope_identity slots; :connection is the per-connection provider's secret source argument; and :tenant is required on multitenant ledgers. Ingress constructs the provider verification context from these known keys and never forwards an input :now_ms clock override.

Functions

adopt_legacy_identity(resource, name, opts \\ [])

Applies one reviewed, quiesced provider-identity adoption partition.

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 decoded payload and signed-body digest 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).

plan_legacy_identity_adoption(resource, name, opts \\ [])

Builds the legacy provider-identity adoption audit without writing rows.

prune(resource, opts)

Retention hook: deletes one bounded batch of TERMINAL ledger rows (:processed, :failed_permanent, :superseded) 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

adopt_legacy_identity(resource, name, opts \\ [])

Applies one reviewed, quiesced provider-identity adoption partition.

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

plan_legacy_identity_adoption(resource, name, opts \\ [])

Builds the legacy provider-identity adoption audit without writing rows.

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. :batch_size defaults to 100 and is capped at 1,000. Only primary-key columns are selected before deletion. Returns {:ok, deleted_count} for this batch; call again until it returns a count below the requested size.

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.