# `Ankusa.Sink.Http`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/sink/http.ex#L1)

Forward the raw hook body to an HTTP endpoint, via
[`Req`](https://hex.pm/packages/req).

`opts`:

  * `:url`        — required target URL, or one computed per hook (see
                     "Computed destinations")
  * `:method`     — HTTP method (default `:post`)
  * `:headers`    — extra request headers as `[{key, value}]` strings, or
                     computed per hook
  * `:timeout_ms` — request/connect timeout (default `5000`)
  * `:secret`     — **required unless `allow_unsigned: true`.** Signs every
                     delivery ([Standard
                     Webhooks](https://www.standardwebhooks.com/)): a
                     `whsec_…` secret, or a list of them while rotating (one
                     `v1,` signature each); computed per hook like `:url`.
                     See `Ankusa.Sink.Http.Signer`
  * `:allow_unsigned` — `true` sends deliveries with no signature (default
                     `false`). Without it a sink with no `:secret` is refused
                     by `validate_opts!/1` at load, and a delivery attempt
                     for one that skipped it is dead-lettered: any host that
                     can reach the receiver's URL could otherwise inject
                     hooks. A `:secret` still signs when this is also set
  * `:max_response_bytes` — read at most this much of the response body
                     (default `65_536`); a longer one is cut off and the
                     connection closed. Only the status is used, so a large
                     or endless response cannot pin the attempt or its memory
  * `:req_options` — transport options for the HTTP client (custom Finch pool,
                     proxy, or `plug:` for `Req.Test` in tests). See
                     `Ankusa.HttpClient` for the accepted keys
  * `:pool`       — this sink's connection pool, over the instance's
                     `egress.pools`: `:pool_size`, `:pool_count`,
                     `:pool_timeout` (ms to wait for a free connection) and
                     `:protocols`. See "Connection pools"
  * `:cacertfile` — a PEM CA bundle to verify the receiver's certificate
                     against (a private CA); default: the system's trust store
  * `:allow_insecure_tls` — `true` skips certificate verification for this
                     sink (default `false`; a `verify_none` in `:req_options`
                     is refused without it). Such a sink's connection is
                     also pinned to the address the egress check resolved

The original `env.body` is sent verbatim with the envelope's content-type
(falling back to `application/octet-stream`). Identity headers `x-ankusa-id`,
`x-ankusa-source`, `x-ankusa-idempotency-key` (the key a consumer dedupes on,
see `Ankusa.Envelope.idempotency_key/1`), `x-ankusa-verification` (`ok`,
`failed` or `skipped`: how the hook was verified at ingest), `x-ankusa-received-at`
(the unix time in milliseconds the edge accepted the hook, the message's
`received_at`), `x-ankusa-body-sha256` (the lowercase hex SHA-256 of the body
sent, the message's `sha256` for a claim) and (when set)
`x-ankusa-tenant` are always added, plus `x-ankusa-verification-flagged: true`
when a failed check was let through by `on_verify_failure: accept_flag` (treat
that hook as unverified), `x-ankusa-dedupe-key` and `x-ankusa-replay-id` when
the hook carries them, `traceparent` (and `tracestate`) when it carries a
trace context (`Ankusa.Trace`: the provider's own, or the ingest span's with an
OpenTelemetry SDK; a provider's header is replaced, not forwarded beside it),
and `webhook-id`/`webhook-timestamp`/`webhook-signature`
with a `:secret` (always, unless `allow_unsigned: true`).
Provider request headers are forwarded per the source's `forward_headers`
option (see `Ankusa.Sink.Message.forwarded_headers/2`); a forwarded name that
collides with one of these or with `opts[:headers]` is dropped.

## Responses

| Status | Result | What dispatch does |
| --- | --- | --- |
| `2xx` | `:ok` | delivered |
| `400`, `401`, `403`, `404`, `410`, `413`, `422`, `451` | `{:error, {:permanent, {:status, s}}}` | dead-letter now |
| `408`, `429`, `5xx` with a `Retry-After` | `{:error, {:retry_after, ms, {:status, s}}}` | retry no sooner than that |
| `425` | `{:error, {:retry_after, ms, {:status, 425}}}`, `ms` the `Retry-After` or else a full-jitter pause | retry no sooner than that |
| `409` | `{:error, {:status, 409}}` | the retry policy: the receiver's state usually moves; bounded by `retry.max_elapsed_ms` |
| anything else | `{:error, {:status, s}}` | the retry policy |

`451 Unavailable For Legal Reasons` is permanent with the other refusals in
the first row: the receiver will not take the hook from this sender, and no
amount of retrying changes that.

`401` and `403` are permanent because a credential (or `:secret`) fix is
followed by a DLQ replay, not by hours of retries against a receiver that
will keep refusing. `Retry-After` is read as seconds or as an HTTP-date.
See "Error classes" in `Ankusa.Sink`.

Redirects are never followed: this body is the hook, and a followed redirect
would re-send it as a `GET`. A `3xx` is reported as its status, for the
source's retry policy to act on.

## Egress policy

Every attempt is checked against `Ankusa.Config`'s `egress` policy before a
byte is sent (`Ankusa.HttpClient.Egress`): `https` only unless the policy
allows `http`, and no address on the private network, loopback or link-local
range (the cloud metadata service included) unless the policy allows it. A
denied target is `{:error, {:permanent, {:egress_denied, reason}}}`: dead-letter
now, like a `401`; fix the policy or the URL, then replay. A name that does not
resolve is `{:error, {:egress_denied, {:resolve, host, reason}}}` and is retried.
The policy comes from the instance in `ctx.instance`; a ctx with none gets the
default, which denies the private ranges.

## Computed destinations

`:url`, `:headers` and `:secret` may each be a `{mod, fun, args}` (called as
`mod.fun(env, args...)`) or a 1-arity function of the `Ankusa.Envelope`,
resolved on **every attempt**, so one shared source delivers each tenant's
hooks to that tenant's receiver:

    {Ankusa.Sink.Http,
     url: fn env -> Tenants.endpoint(env.tenant_id) end,
     secret: {Tenants, :signing_secret, []},
     headers: fn env -> [{"x-api-key", Tenants.api_key(env.tenant_id)}] end}

A function returns what the static option holds (`:url` a string, `:headers`
a list of `{name, value}`, `:secret` what `:secret` accepts, `nil` for none),
or `{:error, reason}` to fail the attempt (`{:error, {:permanent, reason}}` to
dead-letter it, for a tenant with no endpoint); a raise is a transient
failure. A value of the wrong shape is `{:permanent, {:computed, key,
:invalid}}`. The function runs on the delivery task: it must not block on
slow I/O. These are Elixir options; the YAML loader and the admin API only
build static sinks.

The resolved URL is a target like any other: the egress policy above is
applied to it on every attempt, so a tenant-supplied URL cannot reach the
private network, and the signing rule is applied to the resolved secret (an
empty one is `{:permanent, :unsigned_sink}` unless `allow_unsigned: true`).

## Connection pools

Every delivery goes through the node's owned pools
(`Ankusa.HttpClient.Pools`), never Req's default of one pool per origin kept
for the life of the VM: one pool per receiver address and TLS profile, sized
by the instance's `egress.pools` (`pool_size` connections, 32 by default, so
several sinks posting to one receiver are not capped below
`dispatch.concurrency`) or this sink's `:pool`, closed after `pool_max_idle_ms`
idle and when the source is deleted, at most `max_pools` live per instance and
`max_pools_per_tenant` per tenant. A request that cannot check a connection out
within `pool_timeout` fails that attempt as a transport error; one that would
open a pool past a ceiling with nothing idle to close is `{:error, {:unavailable,
:pool_limit}}`; both are retried.

The pool is keyed by the address the egress check validated: its connections go
to that address and verify the peer against the original name, so a name that
later resolves somewhere else is checked again on its next attempt and never
reaches a pool opened for the old answer. `:cacertfile` and
`:allow_insecure_tls` are part of the pool's connection options, so a computed
`:url` may use them too.

# `validate_opts!`

```elixir
@spec validate_opts!(keyword()) :: :ok
```

Raise `ArgumentError` unless `opts` is a usable HTTP sink config: it has a
`:secret` or `allow_unsigned: true`, the `:req_options` are
transport-only and keep TLS verification on, `:allow_insecure_tls` is a
boolean and `:cacertfile` names a file. Boot
(`Ankusa.HttpClient.Egress.validate_config!/1`) and the image's config loader
call it, so a bad sink fails at load instead of at its first delivery; the
admin API turns the same failure into `400 invalid_source`.

---

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