Sends delivered hooks to an Amazon SQS queue, standard or FIFO (or to anything that speaks SQS's JSON protocol, such as the floci emulator in docker-compose.integration.yml). The message body is Ankusa.Sink.Message, byte-identical to what Ankusa.Sink.Kafka and Ankusa.Sink.NATS publish: inline up to :inline_max_bytes, a claim ticket above it.

Dispatch's deliveries are batched (see "Batching"): up to 10 of them to one queue go out as one SigV4-signed SendMessageBatch POST. Anything else (direct publishing, batch: false) is one SendMessage per delivery. Both go through Ankusa.HttpClient (Req's connection pool, no hidden retries or redirects). The sink needs nothing beyond core's req and aws_signature; its batchers are started by Ankusa.Instance. Credentials come from Ankusa.AWS.Credentials: static opts, then the AWS_* environment, web identity (IRSA), then the EC2 instance role (IMDSv2).

This sink never creates a queue. A queue's type, retention, visibility timeout, redrive policy and encryption are an operator's contract, not something the first hook to arrive should decide. A queue that does not exist is {:error, {:permanent, {:config, {:sqs, 400, "QueueDoesNotExist", message}}}}: dead-lettered at once, since only an operator creating the queue fixes it, and a DLQ replay then re-sends the hooks. (Compare Ankusa.Sink.Kafka and Ankusa.Sink.NATS, which likewise never create their topic or stream.)

Publishing is confirmed

deliver/3 returns :ok only on a 200 from SendMessage (or, batched, an entry of a SendMessageBatch reply under Successful) whose MD5OfMessageBody matches the MD5 of the body sent. SQS answers 200 once it has stored the message redundantly, so the sink keeps the default Ankusa.Sink.durable?/1 of true. Everything else is an error (see "Error classes" in Ankusa.Sink):

• {:permanent, {:message_too_large, size, max}} — body plus message attributes over :max_message_bytes, refused before any request.
• {:permanent, {:sqs, status, code, message}} — SQS refused the message itself: InvalidMessageContents (characters outside SQS's allowed set), InvalidParameterValue, MissingParameter, InvalidAttributeName or InvalidAttributeValue. Dead-lettered at once.
• {:permanent, {:config, {:sqs, status, code, message}}} — SQS refused this sink's configuration or credentials: AccessDenied, QueueDoesNotExist, InvalidClientTokenId, UnrecognizedClientException, SignatureDoesNotMatch, InvalidAccessKeyId, MissingAuthenticationToken. Dead-lettered at once and counted in ankusa_dispatch_config_errors_total{sink}.
• {:retry_after, ms, {:sqs, status, code, message}} — SQS throttled the send (Throttling, ThrottlingException, RequestThrottled, TooManyRequestsException, ServiceUnavailable, KmsThrottled, or any HTTP 429 or 503). ms is a full-jitter draw (Ankusa.Sink.throttle_after_ms/1).
• {:sqs, status, code, message} — any other SQS error (a 5xx, an expired token): transient, the retry policy.
• {:status, status, body} — a non-2xx response that is not an SQS error document (a proxy, a load balancer); a 429 or 503 is a {:retry_after, ms, {:status, status, body}}.
• {:md5_mismatch, expected, got}, {:unexpected_reply, body}, :no_credentials and transport errors — retried.

FIFO queues

A :queue_url ending in .fifo is a FIFO queue (AWS requires the suffix on FIFO queue names). Each message then carries:

• MessageGroupId — :message_group_id, default "tenant_id/source_id" ("/source_id" without a tenant), the scope Ankusa.Sink.Kafka uses for its default key: one group's messages are received in the order they were sent.
• MessageDeduplicationId — the hook's id, or id:replay:<replay_id> for a replay (the rule Ankusa.Sink.NATS uses for Nats-Msg-Id), so SQS collapses a retry after a lost response, while a deliberate replay of the same hook is always stored.

SQS's deduplication window is 5 minutes. A retry later than that (a long outage, a slow backoff) can store a second copy, so delivery is at-least-once; consumers dedupe on the message's idempotency_key.

On a standard queue MessageGroupId is sent only when :message_group_id is set (fair queues), and MessageDeduplicationId never is.

Batching

A delivery from dispatch waits in its queue's Ankusa.Sink.Batcher (one per queue and set of credentials) and is sent in a SendMessageBatch with up to 9 others, 20 ms after the first arrived or as soon as the batch is full (10 entries, 256 KiB). deliver/3 returns when the batch's response is known: the hook's row stays claimed until then, so a batcher that dies loses nothing and the attempt is retried. The response is read entry by entry:

• Successful with the MD5 of the body sent: :ok. A different MD5 is {:md5_mismatch, expected, got}.
• Failed: classified on its own, as above (classify_error/4), so a batch with one refused message dead-letters exactly that message. A throttled entry is {:retry_after, ms, reason}, ms full jitter.
• a Failed or a missing entry never fails its neighbours; a request that fails as a whole (a 5xx, a timeout, a 429, QueueDoesNotExist) fails every entry the way a single SendMessage would, and each retries on its own schedule.

Entry Ids are positions in the request ("0" to "9"), not hook ids; the MessageDeduplicationId is unchanged. A batch may hold hooks of several tenants and sources; the batcher takes entries round-robin across tenants, and refuses, with {:retry_after, ms, {:batch_queue_full, reason}}, a delivery that would pass its queue's ceilings (see Ankusa.Sink.Batcher).

On a FIFO queue a MessageGroupId has one request in flight at a time, and its entries leave in the order they were admitted; SQS enqueues a batch's messages in the order they are sent. With the default group (tenant/source) that means a source's messages to one FIFO queue go out as fast as SQS answers, up to 10 to a request. A message that fails and is retried is a later send, as it always was.

What each message carries

• body — the Ankusa.Sink.Message JSON.
• message attributes (all String) — ankusa_id, ankusa_source_id, ankusa_tenant_id, ankusa_message_version, ankusa_idempotency_key, content_type (application/json) — the set Ankusa.Sink.Kafka sets as headers — plus ankusa_dedupe_key and ankusa_replay_id when the hook carries them. ankusa_tenant_id is left out when the hook has no tenant, because SQS refuses an empty attribute value. At most 8 of SQS's 10.

opts

• :queue_url — required, e.g.

                     `"https://sqs.us-east-1.amazonaws.com/123456789012/hooks.fifo"`

• :region — required, e.g. "us-east-1"
• :endpoint — default: the queue URL's origin

                     (`scheme://host[:port]`). Set it for floci or a
                     VPC endpoint.

• :message_group_id — a string, or (Envelope.t() -> String.t());

                     see "FIFO queues"

• :inline_max_bytes — default 64 KiB (65,536), configurable
• :max_message_bytes — the largest body plus attributes sent; default

                     `1_048_576`, SQS's limit (1 MiB). Lower it to a
                     queue's own `MaximumMessageSize`.

• :timeout_ms — default 5_000, for both connect and response
• :access_key_id, :secret_access_key, :session_token — static

                     credentials. Without them the chain in
                     `Ankusa.AWS.Credentials` runs.

• :req_options — transport options for the HTTP client (a Finch

                     pool, a proxy via `:connect_options`, `plug:` for
                     `Req.Test`); see `Ankusa.HttpClient`.

• :batch — default true; false sends every delivery in its

                     own `SendMessage` and counts it against
                     `dispatch.sink_concurrency` as before

• :batch_linger_ms — default 20 (1 to 60,000), how long a batch waits to

                     fill

• :batch_max_entries — default and maximum 10
• :batch_max_bytes — default 262,144, at most 1,048,576 (SQS's limit

                     for a batch's messages together)

• :batch_queue_max, :batch_queue_max_bytes, :batch_queue_tenant_max

                   — the queue's ceilings, default 4,096 entries,
                     128 MiB and 1,024 entries a tenant

• :batch_max_inflight — requests in flight per queue; default the

                     instance's `dispatch.sink_concurrency` (8)