# `AshHooks.Ingress`
[🔗](https://github.com/baselabs/ash_hooks/blob/v2.0.3/lib/ash_hooks/ingress.ex#L1)

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.

# `ctx`

```elixir
@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()
}
```

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.

# `adopt_legacy_identity`

Applies one reviewed, quiesced provider-identity adoption partition.

# `claim_delivery`

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

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`

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

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`

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

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`

```elixir
@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`

```elixir
@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`

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

# `prune`

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

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.

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`

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

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

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`

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

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

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`

```elixir
@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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*