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

The wire format every queue-style sink publishes (`Sink.RabbitMQ`,
`Sink.Kafka`, `Sink.NATS`, `Sink.Redis`, `Sink.SQS`, `Sink.GooglePubSub`), so
a consumer parses one format regardless of transport.

A body of at most `inline_max_bytes` (default 64 KiB) rides inline,
base64-encoded. Anything larger lives in the claim check and the message
carries its `Ankusa.ClaimCheck.Ref` as one string instead, plus the
lowercase hex sha256 the reader checks the redeemed bytes against (see
`docs/claim-check.md`). Every message carries `sha256`, `dedupe_key`,
`replay_id`, `idempotency_key` (the tenant-scoped key a consumer dedupes on:
`Ankusa.Envelope.idempotency_key/1`), `verification` (how the hook was
verified at ingest: `status` is `"ok"`, `"failed"` or `"skipped"`, and
`flagged` is `true` when a failed check was let through by the source's
`on_verify_failure: accept_flag`, so a consumer treats that hook as
unverified), the forwarded provider `headers` and the W3C trace context
`traceparent` / `tracestate` (`Ankusa.Trace`: strings, or `null` for a hook
that carries none; a consumer continues the trace from them) and the origin,
`instance` and `region` (strings, or `null`):

    {"v": 1, "id": "01a0...", "source_id": "stripe", "tenant_id": "acme",
     "received_at": 1737500000000, "content_type": "application/json", "size": 245,
     "dedupe_key": "evt_1", "replay_id": null, "idempotency_key": "acme:stripe:evt_1",
     "verification": {"status": "ok", "flagged": false},
     "headers": {"x-github-event": "push"},
     "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
     "tracestate": null,
     "instance": "default", "region": "eu-west-1",
     "sha256": "2cf24dba...", "body_base64": "eyJpZCI6..."}

    {"v": 1, "id": "01a0...", "source_id": "stripe", "tenant_id": "acme",
     "received_at": 1737500000000, "content_type": "application/json", "size": 3145728,
     "dedupe_key": "evt_1", "replay_id": null, "idempotency_key": "acme:stripe:evt_1",
     "verification": {"status": "ok", "flagged": false},
     "headers": {},
     "instance": "default", "region": "eu-west-1",
     "claim": "urn:ankusa:claim:v1:acme:01M39VMD8RA3C5HR4RBV67Y002",
     "sha256": "3bea8a9a07c1e8dc..."}

## Versioning

`v` is the wire format's major version. The producer writes `v: 1`. Every
SDK accepts `min_supported <= v <= max_supported` (1 and 2, exposed by each
SDK) and maps each to its current struct: a `v: 2` message is defined as a
`v: 1` message plus additive fields, so an SDK decodes it exactly like a
`v: 1` one and ignores the keys it does not know. Adding a field keeps `v`,
and consumers must ignore keys they don't know. A field that changes meaning
or disappears needs a new `v` (3 or later), which no SDK accepts until its
`max_supported` is raised. `instance` (the name of the instance that
published) and `region` (`Ankusa.Config` `region`, `null` for none) are such
additive fields. A hook of an ordered sink (`Ankusa.Ordering`) also carries
`ordering_key`, the key of its lane (`"tenant/source"` for the sink's keyless
lane): the only place a sink with no headers, `Sink.Redis`, can show a
consumer which lane a message is in.

What this function writes is pinned by the producer vectors in
`conformance/cases/producer.json`, generated from it by
`mix ankusa.conformance.gen`; `mise run check:conformance` fails when a field
is added, removed or changed here without regenerating them.

## Size budget

Base64 inflates the inline body by 4/3, so an inline message is about
`inline_max_bytes * 4/3` plus the metadata: ~1 KiB of envelope fields plus
the forwarded `headers` (at most `forward_headers_max_bytes`, 16 KiB, in at
most `max_headers`, 64, per source). The 64 KiB default is about 106 KiB
encoded, under every transport's limit (Kafka `max.message.bytes`, 1 MiB by
default; NATS `max_payload`, 1 MiB; SQS, 1 MiB).

`encode/4` enforces that budget before anything is published, against the
sink's `max_bytes/1` (`c:Ankusa.Sink.Messaging.max_message_bytes/1`):

  * an inline body whose message is over the limit goes to the claim check
    instead, so a raised `inline_max_bytes` degrades to a claim, not to a
    message the broker refuses. That claim is checked in by the attempt, not
    by dispatch, so it is not persisted: a retry checks in again, and the
    orphaned object is swept by claim retention;
  * a body that is over the limit by itself (`size > max`, with
    `inline_max_bytes` at least `size`) cannot be shrunk by any retry:
    `{:error, {:permanent, {:message_too_large, size, max}}}`, dead-lettered
    at once, with no broker call and no claim written;
  * a claim message over the limit (`max` below the ~1 KiB the envelope
    needs) is permanent too.

Forwarded headers are capped in `forwarded_headers/2`: a header that would
take a message past either cap is dropped, and each drop is counted
(`ankusa_sink_headers_dropped_total{reason="count" | "bytes"}`).

# `attributes`

```elixir
@spec attributes(Ankusa.Envelope.t(), Ankusa.Sink.ctx()) :: [{String.t(), String.t()}]
```

The metadata a queue sink carries outside the body, as `{name, value}` pairs in
this order: `ankusa_id`, `ankusa_source_id`, `ankusa_tenant_id`,
`ankusa_message_version` ("1"), `ankusa_idempotency_key`, `content_type`
("application/json"), `ankusa_dedupe_key`, `ankusa_replay_id`,
`ankusa_ordering_key` (the lane's key, of a hook of an ordered sink). A pair
whose value is `nil` or `""` is left out (no tenant, no dedupe key, not a
replay, not ordered). A hook that carries a trace context (`Ankusa.Trace`)
adds `traceparent` and, when it has one, `tracestate`, last.

# `check_in`

```elixir
@spec check_in(atom(), Ankusa.Envelope.t()) ::
  {:ok, Ankusa.ClaimCheck.claim()} | {:error, Ankusa.ClaimCheck.reason()}
```

Check `env`'s body in on its own, as a one-claim pack under a fresh pack id
(`Ref.new_pack_id/0`: the current time and 64 random bits, the entropy every
batch pack has). Nothing about the id is derivable from the hook, and its
date partition is the check-in day, so retention counts from when the object
was written.

Dispatch persists the ref with the attempt's outcome (`fresh_claim`, see
`Ankusa.Dispatch.Pipeline`), so a retry reuses it. A crash between the
check-in and that write orphans one object, which retention sweeps.

# `claim_item`

```elixir
@spec claim_item(Ankusa.Envelope.t()) :: Ankusa.ClaimCheck.item()
```

The claim-check item for an envelope's body.

# `default_forward_headers_max_bytes`

```elixir
@spec default_forward_headers_max_bytes() :: pos_integer()
```

The default `forward_headers_max_bytes` of a source: 16 KiB.

# `default_inline_max_bytes`

```elixir
@spec default_inline_max_bytes() :: pos_integer()
```

The default `inline_max_bytes` for every queue-style sink: 64 KiB.

# `default_max_headers`

```elixir
@spec default_max_headers() :: pos_integer()
```

The default `max_headers` of a source: 64.

# `default_max_message_bytes`

```elixir
@spec default_max_message_bytes() :: pos_integer()
```

The default largest message a queue-style sink publishes: 1 MiB.

# `delivery_id`

```elixir
@spec delivery_id(Ankusa.Envelope.t(), Ankusa.Sink.ctx()) :: String.t()
```

The id a broker dedupes one delivery on: the hook's `id`, or
`id:replay:<replay_id>` for a replay, so a retry after a lost ack collapses
while a deliberate replay of the same hook is stored again (SQS's
`MessageDeduplicationId`, NATS's `Nats-Msg-Id`).

# `encode`

```elixir
@spec encode(Ankusa.Envelope.t(), Ankusa.Sink.ctx(), pos_integer(), pos_integer()) ::
  {:ok, binary()}
  | {:error, {:claim_check, Ankusa.ClaimCheck.reason()}}
  | {:error,
     {:permanent, {:message_too_large, non_neg_integer(), pos_integer()}}}
```

Encode `env` for a sink whose threshold is `inline_max_bytes` and whose
largest message is `max_bytes` (default 1 MiB; see "Size budget").

A body over the threshold uses the ref dispatch already checked in
(`ctx.claim`). A sink called outside dispatch has none, so the body is
checked in here, as a one-claim pack.

# `fits`

```elixir
@spec fits(non_neg_integer(), pos_integer()) ::
  :ok
  | {:error,
     {:permanent, {:message_too_large, non_neg_integer(), pos_integer()}}}
```

Refuse a message the transport would never take, before sending it:
`size` bytes (the sink's own accounting — value plus key, body plus
attributes) against `max`. Over it is `{:error, {:permanent,
{:message_too_large, size, max}}}`, dead-lettered at once.

# `forwarded_headers`

```elixir
@spec forwarded_headers(
  Ankusa.Envelope.t(),
  :default | [String.t()] | Ankusa.Sink.ctx()
) :: %{
  required(String.t()) =&gt; String.t()
}
```

The provider request headers a sink may see, per the source's
`forward_headers` option — and capped, so a provider with large headers
cannot push a small body over a transport limit.

The second argument is either the forwarding mode on its own or a sink
`t:Ankusa.Sink.ctx/0`, whose `:forward_headers`, `:max_headers` (default 64)
and `:forward_headers_max_bytes` (default 16 KiB) apply:

* `:default` forwards every header except the never-forwarded set.
* A list of names forwards only those, still minus the never-forwarded set.
* `[]` forwards nothing.

Names are lowercased; repeated names are joined with `", "` in arrival
order. Never forwarded: `authorization` `proxy-authorization` `cookie` `x-api-key` `host` `content-length` `content-type` `connection` `keep-alive` `transfer-encoding` `te` `trailer` `upgrade` `expect`, plus any
`x-ankusa-*` header. A header the stored-header policy dropped
(`Ankusa.Edge.StoredHeaders`) is not in the hook at all, so no
`forward_headers` setting brings it back; a hashed one is forwarded as its
`sha256:<hex>` digest.

Caps apply to the headers that would be forwarded, in order of first
arrival. A name is kept while fewer than `max_headers` are kept and its
name plus joined value fits in what is left of `forward_headers_max_bytes`;
otherwise it is dropped whole (a header is never truncated), and later,
smaller ones are still considered. With a ctx, each reason that dropped
anything emits `[:ankusa, :sink, :headers_dropped]` once, with `count` the
number dropped and `reason` `:count` or `:bytes`.

# `headers`

```elixir
@spec headers(Ankusa.Envelope.t(), Ankusa.Sink.ctx()) :: [{String.t(), String.t()}]
```

The metadata a sink with transport headers carries beside the body
(`Ankusa.Sink.Kafka`, `Ankusa.Sink.NATS`), as `{name, value}` pairs in this
order: `ankusa_id`, `ankusa_source_id`, `ankusa_tenant_id`,
`ankusa_message_version` ("1"), `ankusa_idempotency_key`, `content_type`
("application/json"), then `ankusa_dedupe_key` and `ankusa_replay_id` when
the hook carries them, then `traceparent` and `tracestate` when it carries a
trace context (`Ankusa.Trace`). Unlike `attributes/2`, `ankusa_tenant_id` is
always present — `""` without a tenant — because a header can carry an empty
value where an SQS or Pub/Sub attribute cannot. The AsyncAPI header schema
(`SinkMessageHeadersV1`) therefore does not require it: it describes both sets.

# `inline_max_bytes`

```elixir
@spec inline_max_bytes(keyword()) :: pos_integer()
```

A sink's configured `:inline_max_bytes`, or the default.

# `max_bytes`

```elixir
@spec max_bytes(keyword()) :: pos_integer()
```

A sink's configured `:max_message_bytes`, or the default (1 MiB). A sink
whose transport has another limit, or another option name, overrides
`c:Ankusa.Sink.Messaging.max_message_bytes/1` instead of calling this.

# `verification`

```elixir
@spec verification(Ankusa.Verification.t() | nil) :: %{
  status: Ankusa.Verification.status(),
  flagged: boolean()
}
```

How the hook was verified, as every sink carries it: `status` is `:ok`,
`:failed` or `:skipped`, and `flagged` is `true` when a failed check was let
through by the source's `on_verify_failure: :accept_flag`.

`:skipped` is a hook nothing checked: a source whose verifier is
`Ankusa.Verifier.None` (which accepts everything, so its `:ok` says nothing
about authenticity), or an envelope that never went through ingest (a
lifecycle event, `verification: nil`). The reason and scheme stay off the
wire: a failed-but-accepted hook is the consumer's to distrust, not to
diagnose.

---

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