# `Ankusa.Edge.Ingest`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/edge/ingest.ex#L1)

Orchestrates one webhook on the hot path: build the envelope, verify inline,
apply the per-source failure policy, then ack once the hook is durable.

Which system accepts it is one config key. With `wal.type: disk` (the
default) the group-commit batcher commits the hook to this node's WAL and
dispatch reads it back later; with `wal.type: none` the hook is published to
the source's sinks inside the request and the ack waits for their confirms
(`Ankusa.Edge.Publish`).

The only place a `2xx`-worthy result is produced is *after* one of those
durable accepts returns, or after a durable quarantine write. Everything
else is a non-2xx.

## Stored headers

Verification and the dedupe key read the request's complete header list.
Only then does the stored-header policy (`Ankusa.Edge.StoredHeaders`: the
instance's `store_headers`, overridden by the source's) remove or hash
credential-bearing headers, so the hook that is committed, archived,
claim-checked and published to sinks never carries them. The number removed
is `:dropped_headers` on the `[:ankusa, :ingest]` span's stop metadata.

## Lookup precedes the body

The source is looked up (`lookup/2`) before the request body is read, and
`ingest/4` only ever runs on a source that exists. A request for a source that
does not exist is a refusal (`refused/2`), counted on its own bounded counter
and never on the `[:ankusa, :ingest]` span, so an unauthenticated caller
cannot mint a metric series per URL it invents. `ingest/2` is the
transport-less entry point (tests, benches, embedding): lookup, then ingest.

## Admission

What may be refused without the body — the tenant's `max_body_bytes`, and the
per-address, per-node and per-source limits — is the HTTP edge's, after the
lookup and before the read (`Ankusa.Edge.Router`,
`Ankusa.Edge.RateLimiter.pre_read/4`). What needs the verified hook is here:
the tenant's hooks and bytes per second (`{:error, {:rate_limited, ms}}`), and
its share of the batchers, charged by `Ankusa.Edge.Batcher.commit/4`
(`{:error, {:tenant_overload, ms}}`). Both are `429`s, and neither is the
node's `503 overload`. With `rate_limits.scope: :fleet` / `dedupe.scope: :fleet`
the quota and the provider event key are counted fleet-wide
(`Ankusa.Edge.Fleet`): an event key another node has claimed but not yet made
durable is `{:error, {:dedupe_pending, ms}}` (`409`, retried by the sender), and
one it has committed is a `{:duplicate, env}` carrying that node's hook id.

## Tracing

The request's `traceparent`/`tracestate` become `Ankusa.Envelope.trace`
(`Ankusa.Trace`). With an OpenTelemetry SDK running, the whole ingest is an
`ankusa.ingest` span (a child of the provider's), with `ankusa.verify` and
`ankusa.commit` inside it, and the hook stores the ingest span's context
instead. Without one, the request's context is stored verbatim and no span
exists.

The request process's `Logger` metadata carries `source_id`, `tenant_id`,
`hook_id` and `trace_id` while a hook is ingested (and `Ankusa.Edge.Router`
clears them per request), so any line logged for it correlates; the server
image's JSON log formatter prints them.

# `raw_request`

```elixir
@type raw_request() :: %{
  :method => String.t(),
  :path => String.t(),
  :headers => [{String.t(), String.t()}],
  :body => binary(),
  optional(atom()) => term()
}
```

# `refusal`

```elixir
@type refusal() ::
  :unknown_source
  | :payload_too_large
  | :body_read_failed
  | :invalid_header
  | :store_unavailable
  | :ip_rate_limited
  | :node_rate_limited
  | :source_rate_limited
```

Why a request was answered before it reached `ingest/4`. The whole set: every
refusal goes through `refused/2`, so a metric label built from it is bounded.
The `*_rate_limited` ones are the pre-read admission limits
(`Ankusa.Edge.RateLimiter.pre_read/4`): per client address, per node, per
source.

# `request`

```elixir
@type request() :: %{
  :source_id => String.t(),
  :method => String.t(),
  :path => String.t(),
  :headers => [{String.t(), String.t()}],
  :body => binary(),
  optional(:tenant_id) => String.t()
}
```

# `result`

```elixir
@type result() ::
  {:ok, Ankusa.Envelope.t()}
  | {:duplicate, Ankusa.Envelope.t()}
  | {:quarantined, term()}
  | {:rejected, term()}
  | {:error,
     :unknown_source
     | :overload
     | :tenant_limit
     | :tenant_erasing
     | :dedupe_quota
     | :dlq_full
     | :store_unavailable
     | :intake_unavailable
     | :intake_pending
     | :quarantine_full
     | {:invalid_header, String.t() | nil}
     | {:rate_limited, pos_integer()}
     | {:tenant_overload, pos_integer()}
     | {:quarantine_rate_limited, pos_integer()}
     | {:dedupe_pending, pos_integer()}}
```

# `check_headers`

```elixir
@spec check_headers([{String.t(), String.t()}]) ::
  :ok | {:invalid_header, String.t() | nil}
```

Refuse header bytes no sink can carry: a name must be visible ASCII
(`0x21..0x7E`), a value visible ASCII, space or tab (RFC 9110 field-value
without `obs-text`). Returns the offending name when it is itself printable.

# `ingest`

```elixir
@spec ingest(atom(), request()) :: result()
```

Look the source up, refuse undeliverable header bytes, then `ingest/4`.

# `ingest`

```elixir
@spec ingest(atom(), Ankusa.Source.t(), String.t(), raw_request()) :: result()
```

Ingest one hook for a source `lookup/2` returned. The caller has already run
`check_headers/1`.

# `lookup`

```elixir
@spec lookup(atom(), Ankusa.Route.t(), Ankusa.Routes.Route.t() | nil) ::
  {:ok, Ankusa.Source.t(), tenant_id :: String.t()}
  | {:error, :unknown_source | :store_unavailable}
```

Resolve a route to its source and the tenant the hook is stored under.

The tenant names a storage partition and a gateway path segment, so it is
checked here, before anything is written: a bad one from any RouteResolver is
a 404, like an unknown source, not a dispatch failure after the provider was
already acked. A miss is counted as a `:unknown_source` refusal.

A source owned by a tenant (any `tenant_id` but `"default"`, the value a
source gets when none is given) answers only for that tenant: a route that
names a different one is the same `404` as a missing source, so the URL's
tenant learns nothing about another tenant's sources. A `"default"` source
is shared: it serves whatever tenant the route names (one `stripe` source
behind `/webhooks/:tenant/stripe` for every customer).

A source store that cannot answer (`{:error, :unavailable}`) is
`{:error, :store_unavailable}`, a `503` the provider retries, never a `404`.

`binding` is the route the guard matched (`Ankusa.Routes.Route`; `nil` with
routes off, and for `ingest/2`), held against what the resolver said. A
disagreement is the same `404` as a missing source:

  * a route bound to a `source_id` admits only that source;
  * a route owned by a tenant (any `tenant_id` but `"default"`) admits only
    that tenant: a resolver that names another tenant, or a source owned by
    another, is refused. Where the resolver names no tenant and the source is
    shared, the hook is stored under the route's tenant: the tenant a route
    resolves for a shared source;
  * a route owned by `"default"` binds no tenant, as before.

# `refused`

```elixir
@spec refused(atom(), refusal()) :: :ok
```

Count a request answered without being ingested. Emits
`[:ankusa, :ingest, :refused]` with the `:reason`; the one place a refusal is
counted.

---

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