ankusa.sdk.message — io.github.jamescarr/ankusa-clj 0.1.0

Decode and verify a queue message: the JSON payload Ankusa's broker sinks (RabbitMQ, Kafka, NATS) publish.

(let [message (message/decode payload)
      hook (message/->hook message gateway)]
  (handle hook (idempotency/key hook)))

A message carries its body inline (body_base64) or by reference (claim, redeemed from the claim-check gateway). decode never touches the network: an inline body is decoded and verified, a claim is only parsed.

Every failure is an :ankusa.sdk/InvalidMessageError with a stable :code ("invalid_json", "not_an_object", "unsupported_version", "invalid_field", "ambiguous_body", "missing_body", "invalid_body_base64", "size_mismatch", "integrity", "tenant_mismatch") and, for "invalid_field", the offending :field (otherwise nil). It is never retryable: the same bytes fail the same way.

->hook clj

(->hook message claim-check-client)

Turn a decoded message into the hook a handler takes:

{:id :source-id :tenant-id :content-type :body :received-at :size :dedupe-key :replay-id :idempotency-key :verification :headers}

claim-check-client (from ankusa.sdk.claim-check/client) is required even for an inline message, so a consumer cannot forget to supply one and then fail the day a payload outgrows the sink's inline limit. An inline message makes no request; a claim is redeemed, and a redemption failure propagates unchanged: its :retryable is the ack-or-requeue decision.

Delivery is at-least-once, so the handler still dedupes on ankusa.sdk.idempotency/key.

decode clj

(decode data)

Decode and verify a message of any supported v (min-supported-version to max-supported-version) from data, a string or a byte[] of UTF-8 JSON. Returns

{:v :id :source-id :tenant-id :received-at :content-type :size :body :claim :sha256 :dedupe-key :replay-id :idempotency-key :headers :verification :traceparent :tracestate :instance :region}

where :body is a byte[] for an inline message and nil for a claim, and :claim the claim ref for a claim and nil otherwise. :headers keeps its string keys. :instance and :region are the name of the Ankusa instance that published the message and the region it is configured with (nil when the message says null or predates the fields). Unknown keys are ignored.

The rules run in order and the first failure wins: valid JSON, an object, a supported v, the typed fields, the body form, then (inline) size and digest or (claim) tenant. Anything but a string or byte[] throws IllegalArgumentException.

max-supported-version clj

The highest message v decode accepts.

min-supported-version clj

The lowest message v decode accepts. v is the wire format's major version: a v: 2 message is a v: 1 message plus additive fields, decoded exactly like it (the keys this decoder does not know are ignored) into the one message map, with :v as the wire said it. Raising max-supported-version is the whole change an SDK needs for a producer that starts writing the next additive version.