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

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

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

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

@spec default_forward_headers_max_bytes() :: pos_integer()

The default forward_headers_max_bytes of a source: 16 KiB.

@spec default_inline_max_bytes() :: pos_integer()

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

@spec default_max_headers() :: pos_integer()

The default max_headers of a source: 64.

@spec default_max_message_bytes() :: pos_integer()

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

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

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

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

@spec forwarded_headers(
  Ankusa.Envelope.t(),
  :default | [String.t()] | Ankusa.Sink.ctx()
) :: %{
  required(String.t()) => 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 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.

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

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

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

@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 Ankusa.Sink.Messaging.max_message_bytes/1 instead of calling this.

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