@setu-ts/messaging-plugin

Cross-service messaging. Registers an IMessageBroker under CAPABILITIES.MESSAGING ('messaging').

Eight brokers ship: InMemoryBroker (zero-dependency default), RedisStreamsBroker, RabbitMqBroker, NatsBroker (JetStream), KafkaBroker, GcpPubSubBroker (@google-cloud/pubsub), ServiceBusBroker (@azure/service-bus), and a custom-injected arm. Each cloud client is an optional dependency, lazily imported or injected.

Installation

import { MessagingPlugin } from '@setu-ts/messaging-plugin';

Usage

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { MessagingPlugin } from '@setu-ts/messaging-plugin';
import { CAPABILITIES, type IMessageBroker } from '@setu-ts/common';

// Your application's own work — a stand-in so this example compiles as written.
declare function provisionAccount(userId: string): Promise<void>;

const app = createApplication({
  plugins: [RuntimePlugin(), MessagingPlugin({ broker: 'rabbitmq', url: 'amqp://localhost:5672' })],
});

// Plugins register during `start()`, so the capability is resolvable only after it.
await app.start({ port: 3000 });

const broker = app.services.get<IMessageBroker>(CAPABILITIES.MESSAGING);

await broker.subscribe<{ userId: string }>('user.created', async (message) => {
  await provisionAccount(message.userId);
});

await broker.publish('user.created', { userId: '123' });

Publish timing

publish resolves once every matching subscription’s work item has been handed to dispatch — never once every handler has returned. This is the guarantee real brokers give (a RabbitMQ or Redis publish returns before delivery), so the in-memory default honours it too, and a plugin that publishes during its own register() cannot deadlock startup against the behaviour-chain gate. A handler that rejects never rejects the publish and never surfaces as an unhandled rejection: on the in-memory broker — which has no ack model and no redelivery — the failure path terminates in a report through the application’s logger. An application constructing InMemoryBroker directly supplies its own reporter via InMemoryBrokerOptions.onDispatchError; absent one the rejection is observed and dropped. One slow or throwing fan-out handler also no longer delays or aborts delivery to its siblings.

Publish options and trust

broker.publish(topic, message, options) takes an optional third argument:

Every option is carried as a transport header beside any native mapping, so it is observable through MessageMetadata.headers on every broker and never silently dropped. The constants ORDERING_KEY_HEADER and DEDUPLICATION_ID_HEADER (from @setu-ts/common) name the two headers. The Cloudflare Workers broker carries orderingKey, deduplicationId and the caller’s headers as envelope fields, because a Cloudflare queue has no transport header channel; on delivery all of them are surfaced as the same transport headers. Invalid options are refused on publish; on delivery, an envelope entry failing the shared rules is dropped. A NatsBroker with an injected connection and no headersFactory drops caller headers and the ordering key (reported once through the logger) but still applies the de-duplication id as nats.js’s native msgID.

What orderingKey promises

orderingKey decides placement — the same key reaches the same partition, retry queue or ordered subscription where the broker has one — not the order handlers finish in. Measured 2026-10-07 against real backends:

A consumer that needs order compares the delivered envelope’s version and drops or defers a stale message.

What a consumer may trust, and what a producer must not derive

1. A delivered x-setu-ordering-key or x-setu-deduplication-id header is a hint written by whoever published the message — validation runs on the publish side only — so a foreign or compromised producer can send any value under these names. Use it to order or de-duplicate your own work, never to authorize anything.
2. A deduplicationId derived from request input lets the caller who chooses it suppress another message with the same id for the broker’s window (NATS’s duplicate_window, 120 s by default; Service Bus’s configured detection window). Derive it from a producer-assigned id — the envelope id publishIntegrationEvent uses by default is one.
3. An orderingKey derived from request input lets a caller concentrate load on one Kafka partition or one Pub/Sub ordering key (1 MB/s per key), and on a log-compacted Kafka topic it lets the caller erase an earlier message that carries the same key. Derive it from an aggregate the application owns.

Invalid options are rejected with a RangeError as a rejected promise, naming the field and the rule and never echoing the refused value. A header name a broker or its server acts on (traceparent, tracestate, cc, bcc, payload, nats-*, x-setu-*, RabbitMQ’s x-death / x-delivery-count / x-acquired-count / x-delay and the x-first-death-* / x-last-death-* forms) is refused on every broker, compared case-insensitively. A goog prefix is refused too, as a precaution: Pub/Sub’s reservation of it is stated only by third-party documentation, and the emulator accepts goog attributes.

Options

MessagingPluginOptions is a union discriminated on broker. Two options are shared by every arm:

Omitting name registers under the bare CAPABILITIES.MESSAGING token as plugin messaging-plugin. Supplying one derives both — token messaging.<name>, plugin messaging-plugin.<name> — so several brokers can coexist in one application.

Every other option is arm-specific — url/client for 'redis-streams', credentials for the cloud arms, an injected IMessageBroker for 'custom'. A missing per-arm field is a compile error rather than a startup throw. See Brokers for the full arm list.

Broker-specific logger options are string sinks ({ error: (message: string) => void }). Broker failures are rendered into that message with the error name, message, safe classifier fields, cause chain, and bounded aggregate members; the sink never receives a driver object directly.

Declare subscriptions where the plugin is composed, instead of resolving the broker after start():

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { MessagingPlugin } from '@setu-ts/messaging-plugin';
import type { IIngressBehavior } from '@setu-ts/common';

/** Runs ahead of every subscribe handler; `next()` continues the chain. */
const auditEveryMessage: IIngressBehavior = {
  handle: (ctx, next) => {
    console.log(`${ctx.kind} ${ctx.name}`, ctx.headers ?? {});
    return next();
  },
};

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    MessagingPlugin({
      broker: 'memory',
      behaviors: [auditEveryMessage],
      subscriptions: [
        {
          topic: 'orders',
          handler: (message) => {
            console.log('order received', message);
          },
        },
      ],
    }),
  ],
});

// Declared subscriptions are established during startup, so nothing is
// subscribed until now.
await app.start({ port: 3000 });

The behaviour chain wraps subscribe() handlers only. respond() remains unwrapped and has no registration arm: its request handler returns a value, unlike the void-returning subscription handler. With no behaviours configured, no PipelinedBroker decorator is applied.

Brokers

¹ Kafka needs its reply topic to exist — see below. ² Cloud brokers need their reply topic to pre-exist (GCP) or require Manage right for admin (Azure). ³ Custom brokers carry whatever RPC capability their adapter provides.

“Supported” for 'nats' and 'kafka' meant “shipped” for four releases while neither could start against its real backend — both were driven for the first time in M90d, which is why the column above is stated per broker rather than left implied.

With 'redis-streams', connection errors of the ioredis client the broker builds go to the application logger — the first error of an outage at warn, identical repeats at debug, the recovery at info — instead of ioredis printing every reconnect failure to console.error. An injected client gets no listener: it belongs to the caller.

node-redis. To use node-redis instead of ioredis, pass { broker: 'redis-streams', client: fromNodeRedis(createClient({ url })) } (fromNodeRedis from @setu-ts/common, createClient from redis, installed in your application). The broker’s blocking read (XREADGROUP … BLOCK blockSizeMs, default 100 ms) shares that connection, as it does with ioredis, so keep the adapter’s commandTimeoutMs above blockSizeMs, and keep blockSizeMs positive: 0 blocks indefinitely, and a read the adapter abandoned would stay open and could still take a message. See Using node-redis.

RabbitMQ durability

Consumer-group queues are declared durable, and every message is published persistent — a durable queue keeps only persistent messages across a broker restart. Before 0.9.0 messages were published transient, so a RabbitMQ restart emptied every group queue of the messages waiting in it. Publishes also go through a confirm channel: publish() resolves only once RabbitMQ has accepted the message, and rejects when it refuses it (or the channel closes first).

Each publish() is bounded by publishTimeoutMs (default 15000, 0 unbounded). The bound covers every broker round trip the publish makes, so a paused broker — which keeps its socket open and answers nothing — rejects the call instead of leaving it pending forever. A rejection is not proof the message was dropped: RabbitMQ may still accept it after the bound.

import { MessagingPlugin } from '@setu-ts/messaging-plugin';

MessagingPlugin({
  broker: 'rabbitmq',
  url: 'amqp://localhost:5672',
  // Both values are the defaults, shown for reference.
  persistentMessages: true,
  publishTimeoutMs: 15_000,
});

persistentMessages: false restores the transient behaviour, for deliberately ephemeral traffic where losing in-flight messages on a restart is acceptable. An injected client without createConfirmChannel() keeps a plain channel — publish() then resolves before RabbitMQ has stored anything — and the broker logs one warning saying so; a real amqplib connection always has it.

RabbitMQ consumer recovery

Since 0.9.0, durable consumer-group handlers must be idempotent: failures redeliver, and a crash after a confirmed copy but before the original ack can duplicate a message.

These numeric options throw RangeError at construction for invalid values, including NaN, fractions, empty delay arrays and decreasing tiers. ConsumerRetryOptions is shared with Redis Streams; each broker retains its own defaults and lease semantics.

For group queue Q, each distinct delay d creates a durable quorum queue Q.retry.<d>ms, with queue-level x-message-ttl: d and default-exchange dead-letter routing back to Q. It uses at-least-once dead-lettering (x-dead-letter-strategy: at-least-once, x-overflow: reject-publish) because the original is acked once the copy is confirmed: classic at-most-once dead-lettering drops a copy that Q cannot take when it expires (measured on RabbitMQ 4: Q absent at expiry received nothing in 240 s; the quorum queue held the copy and delivered it about 180 s after Q reappeared). Changing delays creates new queues without a 406 redeclaration conflict; old retry queues drain into Q. No per-message expiration is copied. The default budget gives about 12.6 minutes of backoff before dead-lettering.

With retries enabled, group names ending in .dead or .retry.<digits>ms are reserved for helper queues. Every generated queue name must fit 255 UTF-8 bytes. Declarative names are checked at plugin construction; imperative and factory subscriptions are checked before declaring anything. Rename conflicting groups, or set consumerRetry: false to keep their existing names. Before enabling retries, migrate any existing queues occupying Q.dead or Q.retry.<delay>ms in the same vhost.

Deserialize failures and IntegrationEventRejectedError go straight to Q.dead. Other failures run the application classifier, then retry below the total attempt budget or dead-letter when exhausted. Retry copies preserve body bytes, messageId, timestamp and transport headers (including traceparent), adding x-setu-attempt (absent means initial attempt 1). Copies omit expiration, userId and the CC/BCC headers: RabbitMQ refuses another user’s user_id by closing the channel, and CC/BCC would deliver the copy to the queues they name. Malformed attempt headers dead-letter immediately. Dead letters add x-setu-attempts, x-setu-topic and x-setu-error, rendered via describeError and bounded to 1 KiB UTF-8. Error descriptions and payloads may contain sensitive data: grant queue and diagnostic-log access only to readers trusted with that data, and apply retention policy. Dead-letter log lines normalize control and format characters in queue, topic and error text, and cap the complete diagnostic at 8192 Unicode code points with a visible truncation marker. Routing names and retained topic headers preserve their original values. When the dead queue exceeds its cap, RabbitMQ drops its oldest ready messages. Drain and delete Q.dead before changing deadLetterMaxLength; RabbitMQ rejects different x-max-length arguments with 406.

The copy is always persistent, even if original publishes use persistentMessages: false. It uses the same confirmed publish path and timeout as normal publishes, and the original is acked only after that publish succeeds. Copies use mandatory publishing: an unroutable basic.return rejects the disposition even if RabbitMQ sends a positive confirm. The framework-owned x-setu-disposition-id header is replaced for each copy and correlates concurrent returns without changing the original message ID. Return listeners are removed on confirm, return, close or timeout. A failed disposition leaves the original unacked and logs the failure, then the broker closes that channel (returning the original to Q) and runs its reconnect-and-replay recovery, which re-declares the retry and dead queues before consuming again; otherwise the unacked originals would fill prefetch on a live channel and stall the consumer. The RabbitMQ user needs configure permission on Q.dead and Q.retry.<delay>ms, read permission on Q.retry.<delay>ms, and write permission on the default exchange, amq.default, which carries every retry and dead-letter copy (measured on RabbitMQ 4: without them subscribe() fails with 403 ACCESS_REFUSED on the first helper declaration, and a copy published without amq.default write closes the channel). Recovery requires confirm channels with channel on/off return listeners and close(); subscribe() refuses a retrying consumer group on an injected facade missing any of them, before declaring anything, and names consumerRetry: false as the alternative. Normal publishes retain their documented unconfirmed-publish limitation. Use consumerRetry: false for legacy nack behavior.

Private exclusive fan-out queues and RPC reply inboxes keep nack with requeue disabled. Measured: TTL dead-lettering reaches an exclusive queue while its connection lives, but discards the copy once the connection is gone. Durable retries therefore apply only to consumer groups. Messages unacked behind a slow handler stay leased until channel closure; elapsed time alone does not cause redelivery. A large prefetch allows more messages to wait in that consumer’s memory and can increase their latency and reduce fairness across replicas. Handlers may run concurrently up to prefetch.

Local RabbitMQ 4 measurement (2,000 persistent messages, 2 ms async handler): prefetch 1/8/32/128 took 6554/813/209/65 ms. Default 32 bounds local backlog while reaching about 9,570 messages/s in this probe; these numbers describe this machine, not a throughput guarantee.

Redis Streams recovery

Redis 6.2 or newer is required (XPENDING IDLE). Handlers must be idempotent: failed messages now retry, including after an application restart. New and reclaimed entries use the same payload, headers, metadata, and acknowledgement path. Reclaim uses XPENDING IDLE and atomic XCLAIM; it reads at most ten pending entries per pass, rotating the cursor so later entries cannot starve.

Counts must be positive safe integers. Millisecond values must be positive integers no greater than 2147483647; delays must be nonempty and nondecreasing. Invalid values, including NaN, throw RangeError at broker construction. There is no consumerRetry: false arm.

The first delay is also the lease for in-flight handlers: set it above the longest handler runtime, or another replica may reclaim a slow handler’s work and run it concurrently. Effective retry latency is the tier plus up to one reclaim interval when the pass keeps pace with the backlog; rotating through a larger backlog adds passes. Reclaimed entries may arrive after newer entries; use aggregateVersion when the application needs ordered effects.

Deserialize failures, missing payloads, and IntegrationEventRejectedError go directly to <topic>.dead.<group>. Exhausted deliveries go there too. The stream carries the original fields plus x-setu-source-id and x-setu-deliveries, written with XADD MAXLEN ~ before source XACK. A crash between those commands can duplicate a dead-letter entry; a failed write preserves the source pending entry. Approximate trimming can exceed the configured length by Redis allocation blocks. Dead-letter streams retain message data; apply suitable Redis ACLs and retention.

Each pass adds a pending query, up to ten claims, and a consumer-info query; cleanup adds deletion commands as needed. Clean unsubscribe/disconnect deletes its own consumer only when pending count is zero. Pending consumers survive shutdown for another replica to reclaim. Redis 6.2 uses consumer idle for the sweep; Redis 7.2+ uses inactive (falling back to idle for consumers that have never delivered).

Injected IRedisStreamsClient facades must supply xpending, xclaim, and xinfo as well as the existing methods, and support xgroup('DELCONSUMER', ...).

NATS prerequisites

The nats broker requires a JetStream-enabled server — start it with the -js flag (or set jetstream in its configuration). connect() fails with a named error rather than a bare platform code:

• JetStreamUnavailableError — the server has no JetStream; the raw 503 the server answered is carried as the error’s cause.
• JetStreamStreamError — the configured streamName does not exist on the server and no streamSubjects was supplied (the message names both remedies: create the stream out of band, or supply streamSubjects), or the server refused the stream read/create (the platform error is carried as cause).

The broker never creates a stream with a catch-all subject: NATS refuses subjects: ['>'] without no_ack, and no_ack makes every JetStream publish reject unobserved. Supply explicit subjects when you want the broker to create the stream:

import { MessagingPlugin } from '@setu-ts/messaging-plugin';

MessagingPlugin({
  broker: 'nats',
  streamSubjects: ['orders.>', 'billing.>'], // used only when the stream is absent
});

An existing stream is never touched, with or without streamSubjects.

A lazily built NATS connection reconnects forever by default (reconnect: { maxAttempts: -1, waitMs: 2000 }). Set a finite maxAttempts to cap attempts; when exhausted, the connection closes and health stays down until the application replaces it. maxAttempts must be an integer ≥ -1; waitMs an integer 1–2147483647. Invalid values throw RangeError at plugin or broker construction. reconnect with an injected client throws TypeError: configure that connection in the application instead.

publish() resolves once JetStream has stored the message. It waits for the server’s acknowledgement, so a publish to a subject no stream captures rejects with an error naming the subject (the server answers 503), and an unresponsive server rejects when publishTimeoutMs expires (default 15 s; 0 unbounded) with PublishTimeoutError. Before 0.9.0 the acknowledgement was not awaited: such a publish resolved as a success and the refusal surfaced as an unhandled rejection, which terminates a Deno or Node process by default.

Request-reply needs its subjects in the stream too. A responder consumes the derived rr.req.<topic> subject and the requester’s reply inbox is rr.inbox.<uuid>, so an RPC-capable stream must cover both — for example streamSubjects: ['orders.>', 'rr.req.orders.>', 'rr.inbox.>']. NATS refuses two streams whose subjects overlap, so only one stream per account can own rr.inbox.>: applications that use RPC on one server must share that stream.

Queue names become consumer names. A SubscribeOptions.queue is the JetStream durable consumer name, and the nats client refuses ., *, >, /, \, space, tab, CR and LF in one before anything reaches the server — which is why the reply inbox’s dotted rr.inbox.<uuid> queue made every request() fail. Each refused character is now escaped as _ plus two hex digits (orders.eu → orders_2eeu), so the name stays readable in nats consumer ls; a queue with none of them is used unchanged, so no existing consumer is renamed. The consumer records its raw queue as setu.queue metadata, which needs NATS 2.10 or later.

The escape is not injective: the legal queue orders_2eeu encodes like orders.eu. Rather than let two independent queues split one consumer’s deliveries, subscribe() rejects with NatsConsumerNameCollisionError naming both queues — in one process, and across processes by reading the existing consumer’s recorded queue. The same error refuses a queue reused across two topics, which used to attach silently to the first topic’s consumer. A consumer created before this release carries no setu.queue; it is accepted when its durable name equals the requested queue and its filter matches the topic.

A queue-less subscription is a shared durable group. Without a queue — subscribe(), and respond() with no queue included — the subscription uses ONE durable JetStream consumer per topic, named from defaultQueue (default 'messaging-consumers'): the group is <defaultQueue>.<topic>. Every replica shares the one consumer, so each message is delivered to ONE replica; a restart re-attaches to the existing consumer and resumes from its acknowledgement floor. A group created for the first time starts at messages published after it, as a new Kafka group does; an explicit queue is unchanged. Before 0.11.0 each process created its own durable messaging-<uuid> consumer that replayed the whole stream on every start and was never removed. To have every replica receive every message, pass a queue that is stable and unique per replica (for example the pod name); a queue made from a random per-start value would leave a durable consumer behind on every start.

The reply inbox is ephemeral. The request() reply inbox (rr.inbox.<uuid>) is an ephemeral consumer: it is deleted on shutdown, and if the server removes it the broker re-creates it, so replies keep arriving after an outage that reaps it.

Consumers earlier releases leaked get a reaping threshold. On connect() the broker sets a 5-minute inactive_threshold on the durable consumers earlier releases leaked — named exactly messaging-<uuid> or rr_2einbox_2e<uuid> whose recorded setu.queue metadata is absent or matches — so the server removes them once nothing reads from them. A replica still running an older version keeps its consumer while it reads. A user queue literally named messaging-<uuid> is therefore a reserved shape: such a consumer is reaped only after 5 min with no client attached.

Pub/Sub subscriptions

When queue is not supplied, each subscription is derived per topic — <defaultQueue>.<topic ID> (default prefix messaging-consumers; a fully-qualified projects/<p>/topics/<id> name contributes only its ID, because / is illegal in a subscription ID) — because a Pub/Sub subscription name is project-global: the old single shared default attached a second topic to the first topic’s subscription, so one topic’s handler consumed the other’s messages with no log, and the RPC channel of one run attached to the previous run’s. A caller-supplied queue is used verbatim, so competing consumers of one topic keep sharing a subscription.

The broker creates a subscription when it is absent. When it already exists, the broker reads which topic it is bound to and refuses one bound elsewhere with PubSubSubscriptionBoundElsewhereError naming the subscription, its topic and the requested topic — pass a distinct queue, or delete the subscription. Pub/Sub reserves no character in a name, so unlike Kafka’s colon the . separator is not recoverable by splitting. A derived or supplied name over 255 characters is refused at subscribe(). Topics must already exist; the broker creates none.

Publish bounds

Every broker arm — Kafka, GCP Pub/Sub, RabbitMQ, Redis Streams and NATS — bounds each transport publish at publishTimeoutMs (default 15000 ms, 0 unbounded), including RPC requests/replies and outbox relay publishes. A value outside 0–2147483647, NaN or Infinity throws RangeError at plugin and broker construction.

An expired publish rejects with the exported PublishTimeoutError, which carries the published topic and the configured bound as the readonly fields topic and timeoutMs:

import type { IMessageBroker } from '@setu-ts/common';
import { PublishTimeoutError } from '@setu-ts/messaging-plugin';

declare const broker: IMessageBroker;

try {
  await broker.publish('orders', { id: 'o-1' });
} catch (error) {
  if (!(error instanceof PublishTimeoutError)) throw error;
  // Not acknowledged within error.timeoutMs; the message may still be delivered.
  console.warn(`publish to ${error.topic} unconfirmed after ${error.timeoutMs} ms`);
}

The timeout does not cancel the transport or prove the message was dropped, so account for possible delivery before retrying. The Kafka, Pub/Sub and RabbitMQ message texts are unchanged from before, so a caller matching on the text keeps working. On Redis Streams the bound covers the publish’s XADD only; consumer reads are unaffected. On NATS it is passed to the client as its own publish timeout and held by the framework’s deadline as well; before 0.11.0 the NATS bound was the client’s fixed 5 s. Configure a longer bound for slow brokers, or 0 for an unbounded wait (the previous behaviour on Redis Streams).

Kafka consumer groups

When queue is not supplied, each subscription’s consumer group is derived per topic — <defaultQueue>:<topic> — because members of one Kafka group must subscribe the same topics: shared groups collapse to empty assignments and stop delivering entirely. A caller-supplied queue names the group itself, so competing consumers of one topic keep load-balancing.

The separator is a colon rather than a hyphen so the derivation cannot collide. A Kafka topic name may not contain : (the broker refuses one at creation), while a group id may, so the (defaultQueue, topic) pair is recoverable by splitting at the last colon. With a hyphen, defaultQueue: 'orders-eu' + topic created and defaultQueue: 'orders' + topic eu-created both produce orders-eu-created, putting two differently-subscribed consumers into one group and restoring the empty-assignment failure this derivation exists to prevent.

A new group with no committed partition offsets captures the topic high watermarks during subscribe(), before run(). On its first group join each member seeks only its assigned, still-uncommitted partitions to the minimum start offset advertised by the current members. A message published after subscribe() resolves on any member is delivered to the group, including when several members first start together; nothing published before the first member’s subscribe() is replayed. subscribe() returns without waiting for the join. A group with any committed offset captures nothing; committed partitions and subsequent joins are left untouched.

Group members are trusted. The minimum is taken over what every current member advertises, so any client able to join the group can move a new group’s start earlier, which replays history to it; it can never move the start later. Kafka authorizes joining a group with Read on the group alone, while committing offsets also needs Read on the topic, so where group and topic ACLs are split a principal that cannot commit the group’s offsets (and cannot read the topic) can still make a new group replay the topic’s retained history. The lowering is not logged. Restrict group membership (Kafka Read ACLs on the group, not only on the topic) where replay matters. Advertisements are keyed by topic, member metadata above 64 KiB is skipped, and an offset longer than 19 digits is refused; only the first refusal per subscription is logged.

This requires an admin-capable factory and a loadable kafkajs module. With an injected factory, the module is tried once at connect() for the round-robin wrapper and metadata codec. If absent, each member seeks to its own watermark, so concurrent first starts can miss messages. Without admin() the step is skipped and kafkajs latest-offset behavior retains its start-position window. A member that crashes between subscribing and joining takes its start offsets with it. Admin or metadata failures warn once per subscription with topic and group (falling back to logger.error); bad member metadata is skipped individually while other usable snapshots still contribute to the minimum. Admin or seek failures leave affected partitions with latest-offset behavior.

Kafka topics

The broker creates no topic. On a broker with auto.create.topics.enable (Kafka’s default), the metadata request that creates a topic answers UNKNOWN_TOPIC_OR_PARTITION — measured on Kafka 4.0 in KRaft mode — and kafkajs does not retry that answer, so a subscription to a topic that did not exist yet used to die at boot with a raw KafkaJSProtocolError naming no topic. subscribe() now retries that one error with exponential backoff within retry (kafkajs’s defaults: 5 retries from 300 ms, about 9 s in all), and the next attempt finds the new topic. When the topic is still unknown after the budget — a broker that does not auto-create — it rejects with KafkaTopicUnavailableError naming the topic and the consumer group; retry: { retries: 0 } names it at once. Pre-create the topic, or enable auto-creation.

retry is also forwarded to new Kafka({ retry }) for kafkajs’s own retries, unless a client is injected. A consumer’s run() can reject — kafkajs’s crash handler rethrows a disconnect that fails — and the broker reports that through the logger instead of letting it become an unhandled rejection.

A consumer that crashes with an error kafkajs would retry is restarted by the broker, after the crash’s own retryTime, else retry.initialRetryTime, else 300 ms — read the way kafkajs’s own restart reads them, so initialRetryTime: 0 waits 300 ms rather than restarting at once. The retry read is KafkaOptions.retry, not an injected client’s own. The broker declines kafkajs’s restart in favour of its own so that stopping can always reach the restarted consumer.

Stopping releases every consumer: disconnect() and unsubscribe() cancel a scheduled restart, wait up to 10 s for an in-flight group join to settle, then disconnect the consumer. A join still pending at 10 s is not disconnected under — that neither stops a join that later succeeds nor returns before kafkajs’s pending JoinGroup is answered — so the release returns and the consumer is disconnected the moment its join settles or fails. Until then the process stays alive, and a record delivered in that window is left uncommitted for the group to redeliver. On a broker with Kafka’s default group.initial.rebalance.delay.ms (3 s), stopping an application moments after it started takes a few seconds.

Request-reply

request() / respond() carry correlation inside a message envelope over each broker’s ordinary publish/subscribe. Correlation is carried in the message envelope independently of transport headers.

RPC rides a channel derived from the topic, so it never collides with plain pub/sub: a subscribe('orders', …) consumer never sees a request envelope, and a publish('orders', …) is never consumed by a responder on 'orders'.

Each broker decides what its reply inbox is. The four whose topics are cheap mint a fresh per-instance one. Kafka cannot — a topic there is a durable, partitioned cluster resource — so it reads a shared replyTopic (default 'messaging.replies') under a consumer group unique to each instance:

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { MessagingPlugin } from '@setu-ts/messaging-plugin';

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    MessagingPlugin({
      broker: 'kafka',
      brokers: ['localhost:9092'],
      replyTopic: 'orders.replies', // must already exist; the broker creates no topics
    }),
  ],
});

Every instance reads every reply on that topic and discards the ones it did not originate, so give a high-traffic service its own replyTopic to bound the fan-out.

RequestTimeoutError and RemoteHandlerError are exported for instanceof handling. MessagingNotSupportedError is also still exported but deprecated — it existed for the Kafka broker’s former refusal and no broker throws it now.

Trace propagation

With TelemetryPlugin registered and tracing enabled, every first-party broker sends W3C traceparent with published messages and reads it on delivery. The plugin creates publish <topic> producer and receive <topic> consumer spans. MessageMetadata.headers is always an object for the first-party transports ({} when the message has no headers); custom brokers retain their own metadata behavior. An injected NATS client must supply headersFactory to construct NATS headers.

Message identity

Every first-party broker populates MessageMetadata.messageId and .timestamp from what its transport actually assigns — read the delivered metadata, not a per-broker guess:

An absent member means the transport carried none — never “the adapter did not look” (M90d / X28-4 closed the two cloud brokers that were not reading what their platforms assign). A consumer reading metadata.messageId for de-duplication on an at-least-once transport gets a value on every first-party broker.

Integration events

The transport primitives above carry any payload; an integration event is the contract layer on top of them. defineIntegrationEvent names an event, versions it, and pairs it with a parser; publishIntegrationEvent and onIntegrationEvent put a portable envelope on the wire and take it off again. No IMessageBroker method changes, no broker adapter is rewritten, and no capability token is added: the envelope is payload data, the whole message argument to publish, because payload is the one channel every broker arm carries while headers are not universally available.

The envelope’s fields:

occurredAt is an ISO-8601 string, not a Date: a payload round-trips through the serializer’s JSON.parse on every transport, so a Date would arrive at the consumer as a string regardless of what the producer put in. The string is the honest type. correlationId, causationId and aggregateId are strings; aggregateVersion is the only number, and it must be FINITE, because JSON serialization maps NaN/Infinity to null — a non-finite one would arrive as null. Each is checked on both sides: refused at the producer, and refused as a malformed rejection at the consumer.

Publish options. publishIntegrationEvent takes the same PublishOptions as broker.publish as an optional sixth argument (after the causal metadata), with two defaults of its own: deduplicationId defaults to the envelope id, so re-publishing the same envelope is de-duplicated where the broker supports it; and orderingKey comes from the caller, else from an optional orderingKey: (envelope) => string | undefined selector on the definition, else none. The selector runs only when the caller supplied no key; if it throws or returns a value the rules refuse, the publish rejects.

One definition serves both directions — the producer reads its type/version/topic, the consumer the same three plus parse:

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { LoggerPlugin } from '@setu-ts/logger-plugin';
import {
  defineIntegrationEvent,
  MessagingPlugin,
  onIntegrationEvent,
  publishIntegrationEvent,
} from '@setu-ts/messaging-plugin';
import { CAPABILITIES, type IMessageBroker, type IRuntimeServices } from '@setu-ts/common';

// Your application's own work — a stand-in so this example compiles as written.
declare function provisionOrder(orderId: string): Promise<void>;

const orderPlaced = defineIntegrationEvent<{ orderId: string }>({
  type: 'orders.placed',
  version: 1,
  topic: 'orders.placed.v1', // MUST end with `.v${version}` — enforced at definition time
  // `parse` is the payload validation boundary, and it runs on a value that has
  // crossed a process. A bare `value as T` is a compile-time assertion that
  // checks NOTHING at runtime, so a wrong payload would reach the handler typed
  // as if it were right. Narrow it for real:
  parse: (value) => {
    const candidate = value as { orderId?: unknown };
    if (typeof candidate?.orderId !== 'string') {
      throw new TypeError('orderPlaced: "orderId" must be a string');
    }
    return { orderId: candidate.orderId };
  },
});

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    LoggerPlugin(),
    MessagingPlugin({
      subscriptions: [
        // Consumer: the handler receives the PARSED payload, the envelope, and
        // the transport metadata. A malformed envelope never reaches it.
        onIntegrationEvent(
          orderPlaced,
          async (payload) => {
            await provisionOrder(payload.orderId);
          },
          { queue: 'billing' },
        ),
      ],
    }),
  ],
});

await app.start();

const broker = app.services.get<IMessageBroker>(CAPABILITIES.MESSAGING);
const runtime = app.services.get<IRuntimeServices>(CAPABILITIES.RUNTIME);

await publishIntegrationEvent(runtime, broker, orderPlaced, { orderId: '123' });

Correlation propagation

A handler that publishes the next event derives the causal fields from the envelope it was handed — causedBy is the whole chain-root rule, extracted so no handler copies it by hand. This is an application-level causal chain carried in the payload; it is a different thing from the W3C traceparent the telemetry layer propagates in transport headers (see Trace propagation), and neither replaces the other.

import {
  causedBy,
  defineIntegrationEvent,
  publishIntegrationEvent,
} from '@setu-ts/messaging-plugin';
import type { IntegrationEventEnvelope } from '@setu-ts/messaging-plugin';
import { CAPABILITIES, type IMessageBroker, type IRuntimeServices } from '@setu-ts/common';

const orderCharged = defineIntegrationEvent<{ orderId: string }>({
  type: 'orders.charged',
  version: 1,
  topic: 'orders.charged.v1',
  parse: (value) => value as { orderId: string }, // terse here; validate for real (see above)
});

// Inside a handler: the consumed envelope's correlationId (or its own id,
// when it is the chain root) plus its id as the direct cause.
async function charge(
  runtime: IRuntimeServices,
  broker: IMessageBroker,
  consumed: IntegrationEventEnvelope<{ orderId: string }>,
): Promise<void> {
  await publishIntegrationEvent(runtime, broker, orderCharged, consumed.data, {
    ...causedBy(consumed),
  });
}

declare const runtime: IRuntimeServices;
declare const broker: IMessageBroker;
declare const consumed: IntegrationEventEnvelope<{ orderId: string }>;
await charge(runtime, broker, consumed);

The versioned-topic rollout policy

Each incompatible version owns a distinct topic whose name ends in .v<version>, and the factory enforces this rather than documenting it: defineIntegrationEvent refuses a topic that does not end with the exact string .v${version}, at definition time — at module load, loudly, with the expected suffix named. A version bump with an unchanged topic is precisely the change that breaks every deployed consumer, and no type checker can see it.

Consumers subscribe explicitly to every version they support, one definition per topic, so they never receive and reject an unsupported version from a shared topic. A producer rolls forward by dual-publishing both topics until every required consumer has deployed onto the new one, then stops the old publication after the agreed migration window:

import { defineIntegrationEvent, publishIntegrationEvent } from '@setu-ts/messaging-plugin';
import { CAPABILITIES, type IMessageBroker, type IRuntimeServices } from '@setu-ts/common';

const ordersPlacedV1 = defineIntegrationEvent<{ orderId: string }>({
  type: 'orders.placed',
  version: 1,
  topic: 'orders.placed.v1',
  parse: (value) => value as { orderId: string }, // terse here; validate for real (see above)
});

const ordersPlacedV2 = defineIntegrationEvent<{ orderId: string; totalCents: number }>({
  type: 'orders.placed',
  version: 2,
  topic: 'orders.placed.v2',
  parse: (value) => value as { orderId: string; totalCents: number }, // terse; validate for real
});

// The migration window: both topics are published; the v1 publication stops
// only after every required consumer has deployed onto v2.
async function publishBoth(
  runtime: IRuntimeServices,
  broker: IMessageBroker,
  order: { orderId: string; totalCents: number },
): Promise<void> {
  await publishIntegrationEvent(runtime, broker, ordersPlacedV1, { orderId: order.orderId });
  await publishIntegrationEvent(runtime, broker, ordersPlacedV2, order);
}

declare const runtime: IRuntimeServices;
declare const broker: IMessageBroker;
await publishBoth(runtime, broker, { orderId: '123', totalCents: 9900 });

The suffix guard locks out a topic that predates this policy — a live orders.created with no version suffix cannot be expressed as a definition. The raw broker.publish / broker.subscribe surface is unchanged and remains the documented route for such a topic.

What the publisher does NOT do

publishIntegrationEvent never runs parse. The parser is a narrowing function unknown → T, and a realistic one (a schema with defaults, coercion, or stripping) returns a different object than it was given — running it on publish would silently change what the producer asked to send. The honest consequence: a producer can publish a payload its own consumers reject, and that surfaces at the consumer as a reason: 'parse' rejection. What the consumer parses is the value after a JSON round trip, which the producer’s call never saw.

Rejections

onIntegrationEvent’s wrapper validates the delivered envelope structurally, checks type and version for exact equality, and runs parse — all before the application handler runs. A refusal throws IntegrationEventRejectedError, discriminated by reason:

Unknown EXTRA envelope fields are accepted and forwarded — a producer on a later framework version may add a field this build does not know about, and refusing it would make every additive envelope change a coordinated deployment. Payload strictness is the parser’s job, where the application owns the policy.

A payload of undefined is refused at the producer: JSON.stringify drops a key whose value is undefined, so such an event would publish cleanly and then arrive with no data at all, and every consumer would refuse it as malformed — silently, on a composition with no logger. A payloadless integration event publishes null. For the same reason a non-finite aggregateVersion (which serializes to null) is refused, and on the consumer side a present correlationId/causationId/aggregateId must be a string, a present aggregateVersion a finite number, and occurredAt a real ISO-8601 instant — otherwise the envelope’s declared types would be a lie one hop before causedBy copies correlationId into the next event.

occurredAt must be an ISO-8601 instant in the interoperable RFC 3339 profile — a full date, a time to at least seconds, and an explicit Z or a numeric offset (2026-01-01T00:00:00Z, 2026-01-01T00:00:00.000Z, 2026-01-01T00:00:00+05:30). A date-only value, an RFC 2822 date, or a zone-LESS timestamp is refused: Date.parse accepts all three, and the zone-less form is read in each engine’s own local time — measured, 2026-01-01T00:00:00 becomes 2025-12-31T18:30:00.000Z on a +05:30 host — so one event would mean a different instant on every consumer.

The rejection follows the broker’s OWN failure path, and that path differs per arm — it is not a retry guarantee. RabbitMQ durable groups now send deterministic rejections directly to Q.dead and log the failure; private queues, reply inboxes and consumerRetry: false retain nack with requeue disabled (operator DLX if configured, otherwise discard). NATS naks, which redelivers while the stream retains the message, and (since PR #287) also logs it. The in-memory broker reports through onDispatchError and drops. Because a rejection here is deterministic — the same envelope fails the same way on every delivery — redelivery cannot resolve it, so a dead-letter queue rather than a retry is where a refused event is inspected. The error’s message carries the whole diagnostic (reason, topic, expected against observed), because the in-memory default composition logs exactly error.message through the application’s logger; the structured fields serve an instanceof branch on a path that surfaces the error object, such as a dead-letter consumer or a bespoke sink on the 'custom' broker arm.

Behaviour ordering

When MessagingPlugin({ behaviors }) is configured, the ingress behaviour chain runs before the integration-event wrapper, so IngressContext.payload is the raw envelope and never the parsed payload. That ordering is deliberate: a behaviour that short-circuits (a tenant guard, an auth check) must be able to refuse a message without the framework parsing it first, and a behaviour reading envelope.correlationId off the raw object is doing something useful. A behaviour therefore cannot depend on payload being the parsed T.

The dispatch consumer identity. Every behaviour also sees IngressContext.consumer — the identity of the subscription the message was dispatched to. A subscription created with a queue carries that queue name; a queue-less subscription gets a per-process subscription:<prefix>:<n>. It is always present for a broker-delivered message, so a behaviour may key per-consumer state on it — the idempotentIngress() behaviour uses it so two subscribers on one topic each run once. On a topic you make idempotent, pass queue to subscribe (together with your broker’s competing-consumer configuration) so a redelivery reaches the SAME consumer identity; a queue-less subscription’s identity is per process, so on a broker whose queue-less default is a shared group a redelivery to another replica is not de-duplicated.

Mapping a local domain event to an integration event

Mapping a fact an aggregate recorded (see @setu-ts/events-plugin) onto a published integration event is APPLICATION policy, written out explicitly — no framework path performs it. Two traps get named: event.type is the domain fact’s name and deliberately NOT the integration type (the contract owns its own type/version pair, so refactoring an internal class name cannot break the wire), and event.occurredOn is when the fact HAPPENED while the envelope’s occurredAt is set at publish time — different instants, never conflated.

import { createDomainEvents } from '@setu-ts/events-plugin';
import { defineIntegrationEvent, publishIntegrationEvent } from '@setu-ts/messaging-plugin';
import {
  CAPABILITIES,
  type IDomainEvent,
  type IMessageBroker,
  type IRuntimeServices,
} from '@setu-ts/common';

const orderPlaced = defineIntegrationEvent<{ orderId: string }>({
  type: 'orders.placed',
  version: 1,
  topic: 'orders.placed.v1',
  parse: (value) => value as { orderId: string }, // terse here; validate for real (see above)
});

// The facts an aggregate recorded locally, read at the application boundary.
const events = createDomainEvents();

// The contract's own parser is the narrowing step for the payload; the causal
// fields come from the domain fact.
function metadataFor(event: IDomainEvent): {
  causationId: string;
  aggregateId?: string;
  aggregateVersion?: number;
} {
  const metadata: { causationId: string; aggregateId?: string; aggregateVersion?: number } = {
    causationId: event.id,
  };
  if (event.aggregateId !== undefined) metadata.aggregateId = event.aggregateId;
  if (event.version !== undefined) metadata.aggregateVersion = event.version;
  return metadata;
}

async function publishPending(runtime: IRuntimeServices, broker: IMessageBroker): Promise<void> {
  for (const event of events.pending()) {
    await publishIntegrationEvent(runtime, broker, orderPlaced, orderPlaced.parse(event.data), {
      ...metadataFor(event),
    });
  }
}

declare const runtime: IRuntimeServices;
declare const broker: IMessageBroker;
await publishPending(runtime, broker);
events.clear(); // after the application's own confirmed dispatch policy

Publishing this way says nothing about any consumer finishing: as Publish timing documents, a transport accepting a publish resolves on dispatch hand-off, not on handler completion.

Transactional outbox

A service that changes its own data and announces the change must do both or neither. The outbox writes an integration event as a ROW in the same database transaction as the business change, through your own unit of work, and a relay publishes the persisted rows afterwards. If the transaction rolls back, no row exists and nothing is ever published; if it commits, the event is published even when the process dies before the relay runs.

import { CAPABILITIES } from '@setu-ts/common';
import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { SchedulerPlugin } from '@setu-ts/scheduler-plugin';
import {
  createDatabaseOutboxStore,
  DatabasePlugin,
  type IDatabaseService,
} from '@setu-ts/database-plugin';
import { defineIntegrationEvent, type IOutbox, MessagingPlugin } from '@setu-ts/messaging-plugin';

const orderPlaced = defineIntegrationEvent<{ orderId: string }>({
  type: 'orders.placed',
  version: 1,
  topic: 'orders.placed.v1',
  parse: (value) => value as { orderId: string },
  // Rows of one ordering key are relayed in write order (see "The promise").
  orderingKey: (envelope) => envelope.data.orderId,
});

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    SchedulerPlugin(), // runs the relay every `relay.intervalMs`
    DatabasePlugin({ type: 'memory' }),
    MessagingPlugin({ outbox: { store: createDatabaseOutboxStore() } }),
  ],
});
await app.start();

const db = app.services.get<IDatabaseService>(CAPABILITIES.DATABASE);
const outbox = app.services.get<IOutbox>(CAPABILITIES.OUTBOX);

await db.transaction(async (uow) => {
  await uow.getRepository('Order').create({ id: 'o-1', total: 42 });
  // The SAME unit of work: the row commits or rolls back with the order.
  await outbox.write(
    uow,
    orderPlaced,
    { orderId: 'o-1' },
    {
      metadata: { aggregateId: 'o-1', aggregateVersion: 1 },
    },
  );
});
// Optional: request a sweep now instead of waiting for the next interval.
outbox.dispatch();

MessagingPlugin({ outbox }) registers an IOutbox under CAPABILITIES.OUTBOX (outbox.<name> for a named instance) and an outbox health indicator. At onInit it resolves the store, runs its verify() (a backend that cannot serve an outbox fails start() by name), and schedules the relay (outbox-relay) and the retention purge (outbox-purge) on CAPABILITIES.SCHEDULER. With the default relay.schedule: true and no scheduler registered, start() rejects OutboxRelayUnscheduledError.

write uses ONE implementation of the publish precedence with publishIntegrationEvent: the caller’s orderingKey beats the definition’s selector, and deduplicationId defaults to the envelope id, so every re-send of a row carries the same deduplication id. An envelope larger than maxEnvelopeBytes (default 262 144 UTF-8 bytes) is refused with OutboxEnvelopeTooLargeError INSIDE the transaction, so the business write rolls back with it rather than committing a row that could never be sent. A write before onInit completes rejects OutboxNotReadyError.

The promise

At-least-once delivery of every committed row. At most one relay holds a row’s claim at a time, provided the relays’ wall clocks agree within relay.maxClockSkewMs. A relay starts publishing only while its claim leaves room for the publish, status write and clock skew. Any number of relays keep per-key order among FIRST publishes, provided rows of one key commit in write order and the writers’ clocks agree. Delivery order is what the broker gives it (see What orderingKey promises); consumers compare aggregateVersion.

Never exactly once: a repeat carries the same envelope and deduplication id, which a consumer inbox absorbs. Two windows remain: a relay can pause AFTER the fence check and resume after its lease; and a publish the broker accepts after it was abandoned can arrive late. Those repeats can arrive out of order. A relay paused past its claim BEFORE the fence check publishes nothing.

The ordering limit is measured, not theoretical. On real PostgreSQL 16, transaction A wrote its row and stayed open while B wrote and committed; the relay published B; then A committed. The relay published A on its next tick — a watermark relay (id > lastSeen or createdAt > lastSeen) would never have published A at all, which is why the relay reads the PENDING set rather than a watermark — but B, written second, was published first, because it committed first. A relay can order only the rows it can see. Clock skew is the cross-replica version of the same limit: a row’s position is taken from its writer’s clock, clamped per process, so two writers whose clocks disagree interleave by clock rather than by real time.

The mid-batch and after-publish rows are driven against real PostgreSQL with real RabbitMQ 4 and real Redis Streams in test/integration/outbox-real.test.ts.

Backends

Use createDatabaseOutboxStore() from @setu-ts/database-plugin (any IOutboxStore works — see A custom store). Its verify() refuses a backend that cannot serve the relay at startup with OutboxStoreUnavailableError, naming the reason:

On DynamoDB the relay’s position > after page runs as a KEY condition on that GSI (status = :pending AND position > :after, with kind as a filter), recorded against DynamoDB Local, so paging never rescans. A transaction there is one TransactWriteItems, at most 100 writes INCLUDING outbox rows.

The discriminator. Every row carries kind: 'setu-outbox'; every read, count, transition and purge requires it, and a lookup by id treats a row of another kind as missing. An outbox sharing a table, collection or container with business documents — Cosmos queries a container as a whole — never reads, counts, transitions or deletes a business document that happens to carry status: 'pending' or 'sent'. Driven on the Cosmos emulator with business documents carrying both.

Provisioning

No portable migration exists (DatabaseService.migrate() always rejects), so create the table yourself. PostgreSQL (the exact file the real suite applies):

CREATE TABLE setu_outbox (
  id            text    PRIMARY KEY,
  kind          text    NOT NULL,
  topic         text    NOT NULL,
  envelope      text    NOT NULL,
  options       text    NOT NULL,
  ordering_key  text,
  tenant_id     text,
  traceparent   text,
  position      text    NOT NULL,
  created_at    bigint  NOT NULL,
  status        text    NOT NULL,
  attempts      integer NOT NULL,
  available_at  bigint  NOT NULL,
  claim_version bigint  NOT NULL,
  lease_until   bigint  NOT NULL,
  last_error    text,
  settled_at    bigint,
  sent_by       text
);
CREATE INDEX setu_outbox_relay ON setu_outbox (kind, status, position);
CREATE INDEX setu_outbox_purge ON setu_outbox (kind, status, settled_at);

SQLite and Cloudflare D1, where D1Adapter uses the field names as column names (the file the D1 test applies):

CREATE TABLE setu_outbox (
  id          TEXT    PRIMARY KEY,
  kind        TEXT    NOT NULL,
  topic       TEXT    NOT NULL,
  envelope    TEXT    NOT NULL,
  options     TEXT    NOT NULL,
  orderingKey TEXT,
  tenantId    TEXT,
  traceparent TEXT,
  position    TEXT    NOT NULL,
  createdAt   INTEGER NOT NULL,
  status      TEXT    NOT NULL,
  attempts    INTEGER NOT NULL,
  availableAt INTEGER NOT NULL,
  claimVersion INTEGER NOT NULL,
  leaseUntil  INTEGER NOT NULL,
  lastError   TEXT,
  settledAt   INTEGER,
  sentBy      TEXT
);
CREATE INDEX setu_outbox_relay ON setu_outbox (kind, status, position);
CREATE INDEX setu_outbox_purge ON setu_outbox (kind, status, settledAt);

MySQL takes the PostgreSQL columns with VARCHAR(64) for id, kind, status and position (an index needs a bounded key) and TEXT elsewhere; this repository has no MySQL backend, so that form is unverified. Keep position in a binary or C collation-equivalent ordering: it is fixed width with no separators, so a collation that ignores punctuation cannot reorder it.

The mappings for the other adapters:

import { bigint, integer, pgTable, text } from 'drizzle-orm/pg-core';
import { CosmosAdapter, DynamoAdapter, MongoAdapter } from '@setu-ts/database-plugin';

// Drizzle: register this table as `drizzleTables: { Outbox: outbox }`.
export const outbox = pgTable('setu_outbox', {
  id: text('id').primaryKey(),
  kind: text('kind').notNull(),
  topic: text('topic').notNull(),
  envelope: text('envelope').notNull(),
  options: text('options').notNull(),
  orderingKey: text('ordering_key'),
  tenantId: text('tenant_id'),
  traceparent: text('traceparent'),
  position: text('position').notNull(),
  createdAt: bigint('created_at', { mode: 'number' }).notNull(),
  status: text('status').notNull(),
  attempts: integer('attempts').notNull(),
  availableAt: bigint('available_at', { mode: 'number' }).notNull(),
  claimVersion: bigint('claim_version', { mode: 'number' }).notNull(),
  leaseUntil: bigint('lease_until', { mode: 'number' }).notNull(),
  lastError: text('last_error'),
  settledAt: bigint('settled_at', { mode: 'number' }),
  sentBy: text('sent_by'),
});

// MongoDB (a replica set): the envelope id is the `_id`, stored as a string.
export const mongo = new MongoAdapter({
  url: 'mongodb://127.0.0.1:27017/?replicaSet=rs0',
  database: 'shop',
  collections: { Outbox: { collection: 'outbox', idType: 'raw' } },
});

// DynamoDB: a table keyed by `id`, and the GSI the relay pages through.
export const dynamo = new DynamoAdapter({
  region: 'eu-west-1',
  entities: {
    Outbox: {
      table: 'outbox',
      partitionKey: 'id',
      indexes: { 'by-status-position': { partitionKey: 'status', sortKey: 'position' } },
    },
  },
});

// Cosmos: the outbox shares the business CONTAINER and partition-key path, so
// one batch carries the business write and the row. Write with `tenantId`.
export const cosmos = new CosmosAdapter({
  endpoint: 'https://shop.documents.azure.com:443/',
  key: '…',
  database: 'shop',
  containers: {
    Order: { container: 'orders', partitionKey: 'tenantId' },
    Outbox: { container: 'orders', partitionKey: 'tenantId' },
  },
});

MongoDB indexes, for the relay and the purge:

db.outbox.createIndex({ kind: 1, status: 1, position: 1 });
db.outbox.createIndex({ kind: 1, status: 1, settledAt: 1 });

The DynamoDB table needs id as its hash key and the GSI by-status-position (hash status, range position, both S) with "Projection": { "ProjectionType": "ALL" } — a narrower projection is not detected by verify(). A Prisma model maps the PostgreSQL columns with @map (orderingKey String? @map("ordering_key"), createdAt BigInt @map("created_at"), …) and @@map("setu_outbox"). Prisma Client returns a BigInt column as a JS bigint; the bridge converts every integer column it reads to a number. A value too large for a safe integer stays too large after conversion, so the relay still refuses it as invalid-row in claimVersion and leaseUntil. The Prisma path is not driven in this repository.

The relay

The relay walks the pending set in LAPS, keyset-paged on position, examining rows one at a time. A failed row, a row in retry backoff, a row held by another relay, or a row whose status write failed BLOCKS its ordering key for the rest of the lap, so a later row of the same key is never published first; an unkeyed row is never blocked. Two budgets per sweep keep a stuck key from starving others: scanLimit (rows examined, default 1000) and publishLimit (publishes, default 100); the lap carries across sweeps, so the scan moves past a blocked key’s rows and reaches an unblocked key behind them.

A publish failure is retried with backoff (baseBackoffMs doubled per attempt up to maxBackoffMs); after maxAttempts (default 10) the row becomes failed. A row that cannot be decoded — its envelope is not JSON, too large or carries another id; its options are longer than any the outbox writes, fail validation or disagree with its orderingKey column; or its topic is empty, over 255 bytes or carries a control character — is failed at once with lastError: 'invalid-row'. Invalid claim fields also become failed: a stored version must be an integer in [0, Number.MAX_SAFE_INTEGER - 1], and a lease must be a nonnegative safe integer no farther than now + 3 660 000 ms. This global horizon accepts other replicas’ longer leases during a rollout. MAX_SAFE_INTEGER is exhausted and never incremented; MAX_SAFE_INTEGER - 1 is claimed normally. markInvalid preserves attempts, clears the lease and writes invalid-row. Either kind of failed row blocks its key until an operator calls release.

release carries no built-in authorization. It is an operator capability: gate the route that calls it. 'retry' returns the row to pending with zero attempts and leaseUntil: 0, keeping claimVersion; 'discard' settles it without publishing (and the purge deletes it later). A release takes effect at the relay’s next lap.

One deadline per sweep. relay.sweepDeadlineMs (default 30 000), on the monotonic clock, bounds the whole sweep and purge, EXCEPT that the status write settling an already-invoked publish runs to its own storeTimeoutMs even when a pause has exhausted the budget: a sweep that overran by a pause can outlive the deadline by at most one storeTimeoutMs, and the shutdown drain that waits for it is bounded by the same amount. Each publish and store call has its own bound (default 5 000 each). A row starts only while publishTimeoutMs + 2 × storeTimeoutMs fits: the claim, publish and status write reserve 15 000 ms by default, leaving the first 15 000 ms for row starts. Construction requires publishTimeoutMs + 2 × storeTimeoutMs < sweepDeadlineMs; equality is refused naming all three options.

Several relays may sweep one store. Each row is claimed through an atomic compare-and-set on claimVersion, incrementing it once and recording leaseUntil. Another relay waits until now ≥ leaseUntil + maxClockSkewMs; every status write, including immediate deletion, requires the held version. relay.claimLeaseMs defaults to 30 000 (integer 1–3 600 000) and must cover publishTimeoutMs + 2 × storeTimeoutMs + maxClockSkewMs. relay.maxClockSkewMs defaults to 5 000 (integer 0–60 000); zero assumes perfectly agreeing clocks and is unsafe across hosts.

A shared scheduler lock saves redundant scans: without it each replica scans every interval. dispatch() runs outside the lock. The former sweepDeadlineMs ≤ distributedLock.ttlMs − 10 000 rule no longer bears on correctness. Claims use the same protocol on every supported backend; the bridge refuses sources without native conditional writes.

The overlap kinds are fenced (the pre-publish check refused an expired claim), claim-lost (a status write found another version) and duplicate (markSent found the row already sent, or markFailure did after the relay’s publish timed out — the copy may have duplicated). A settlement answer is recorded even when it arrives after the call’s own bound or the sweep deadline; a relay paused inside its publish or at a settlement write (markSent or markFailure) may also report the store-write-failing health reason for that sweep. Any kind, including a dispatch sweep, adds the health reason relay-overlap inside the configured window. Losing at CLAIM time is ordinary contention and has no shipped signal. retainSentMs: 0 disables only duplicate; fenced and claim-lost still work. sentBy remains an operator diagnostic, and is absent from transition outcomes.

On DynamoDB a stale GSI page cannot cause a publish: the claim’s conditional write checks the latest base item. This follows AWS’s documented model and is exercised against DynamoDB Local, not the service.

On shutdown the relay drains in an onShutdown hook, before any onClose hook closes the broker or the database: the jobs are removed, dispatch() becomes a no-op, and the in-flight sweep finishes within its deadline plus, when a pause made it overrun, one storeTimeoutMs. A failure caused by the application stopping is never counted against maxAttempts.

Tenancy

Column isolation: one store; write(…, { tenantId }) records the tenant in the row, blocking keys include it (one tenant’s poisoned row never blocks another tenant’s key of the same value), and the relay sweeps every tenant together. Database per tenant: stores: { [tenantId]: entry } (typically createDatabaseOutboxStore({ database: '<name>' })); write and release select the store by tenantId and reject OutboxUnknownTenantError for an absent or unknown one. store and stores together are a compile error.

The store must read the database your unit of work writes to — a caller obligation the outbox cannot check, because a unit of work carries no database identity. A per-tenant store for ANOTHER tenant, or a createDatabaseOutboxStore({ database }) naming a different connection than the transaction’s, puts the row in the transaction’s database where the selected store’s relay never looks: it is relayed under the wrong tenant if that database has its own relay, and never if it does not. The tenant is never a published header: an event that needs its tenant carries it in the payload.

Retention

purge() runs on its own interval, purgeIntervalMs (default 60 000), never per sweep, and deletes sent and discarded rows whose settledAt is older than retainSentMs (default 7 days), at most purgeBatch (default 100) per status per run. Settled rows hold event payloads, which may be personal data — set retainSentMs to what your retention policy allows. retainSentMs: 0 deletes a row at mark-sent (and disables only duplicate detection). failed rows are never purged: they block their key until released. On DynamoDB the purge is a status-partition query with settledAt as a filter, which READS (and bills for) every settled item it passes over; purgeIntervalMs bounds how often.

Trust

Protect the outbox table like the broker credentials. The relay publishes whatever a row says: its topic, envelope and options. Rows are re-validated on every read — size, id, options, ordering key and topic format — and a row failing any check is invalid-row, never published; but a VALID row edited to name another topic is published there. Anyone with write access to the table can therefore publish under this service’s broker identity, which is a stronger capability than the broker’s own trust boundary assumes.

A failed row’s lastError column holds the broker’s error message, with control characters removed and cut to 1 024 characters. A broker may quote part of the message it refused in that text, so the column can carry payload data: it is covered by the same protection as the rest of the table.

Startup is bounded. Each store’s verify() runs under relay.storeTimeoutMs (default 5 000). A database that accepts the connection and never answers fails start() with OutboxStoreVerifyTimeoutError instead of waiting on the driver’s own timeout, which pg does not set by default.

Health and metrics

The outbox indicator reads the store’s counts through a cached, bounded probe (5 s TTL, 2 s bound): down when the store does not answer, otherwise up with data.reasons from a fixed vocabulary, else up. data carries counts, ages in ms and the last local sweep’s counts — never a tenant id, ordering key, topic or error text.

A replica that never wins the scheduler’s slot never sweeps, so it reports none of the per-instance reasons. With CAPABILITIES.METRICS registered the outbox adds counters outbox_published_total, outbox_publish_failures_total, outbox_poisoned_total (labelled topic, capped at 100 distinct values then other) and outbox_overlaps_total (kind: fenced, claim-lost, duplicate), and gauges outbox_pending_rows and outbox_oldest_pending_seconds, each labelled outbox with the instance’s token. The two gauges are written only when the health indicator reads the store — they are as fresh as the last /health poll, and stale if nothing polls it.

Trace continuity

write stores the active traceparent; the relay publishes each row inside an outbox relay <topic> span whose parent is that stored context, so request → relay → publish → receive is one trace even though the sweep runs later, from a clean context. A row with no valid traceparent — written outside a span, or edited — gets a ROOT relay span (SpanOptions.root), never the span of whatever request happened to dispatch the sweep. Both are driven against the real OpenTelemetry SDK in test/integration/outbox-trace-real.test.ts. The relay span is active only when a context manager is registered (TelemetryPlugin’s contextPropagation, on by default); without one the producer span is a root. A custom broker without the framework-header channel — WorkersBroker today — drops the producer’s traceparent, so the consumer cannot continue the trace.

Cloudflare Workers

Workers has no scheduler (SchedulerPlugin refuses to start there): set relay: { schedule: false }, hand waitUntil to background so a dispatch()ed sweep survives the response, and drive the relay and the purge from Cron Triggers. The outbox table is a D1 table through D1Adapter:

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { createDatabaseOutboxStore, DatabasePlugin } from '@setu-ts/database-plugin';
import {
  createScheduledHandler,
  D1Adapter,
  type ID1Database,
  type IScheduledController,
  WorkersCron,
} from '@setu-ts/cloudflare-plugin';
import { MessagingPlugin } from '@setu-ts/messaging-plugin';
import { CAPABILITIES } from '@setu-ts/common';
import type { IOutbox } from '@setu-ts/messaging-plugin';

// From `cloudflare:workers` in a real Worker.
declare const env: { readonly DB: ID1Database };
declare function waitUntil(promise: Promise<unknown>): void;

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    DatabasePlugin({
      type: 'custom',
      adapter: new D1Adapter(env.DB, { tables: { Outbox: { table: 'setu_outbox' } } }),
    }),
    MessagingPlugin({
      outbox: {
        store: createDatabaseOutboxStore(),
        relay: { schedule: false },
        background: waitUntil,
      },
    }),
  ],
});
await app.start();
const outbox = app.services.get<IOutbox>(CAPABILITIES.OUTBOX);

const cron = new WorkersCron()
  .on('* * * * *', async () => {
    await outbox.sweep();
  })
  .on('0 * * * *', async () => {
    await outbox.purge();
  });

export function scheduled(controller: IScheduledController): Promise<void> {
  return createScheduledHandler(cron)(controller);
}

Cron Triggers fire at most once a minute, so call outbox.dispatch() after each transaction for latency and keep the cron for reliability. This composition is driven at unit level (the shipped D1Adapter over real SQLite, WorkersCron and createScheduledHandler) in test/integration/outbox-workers.test.ts; it has NOT been driven on real workerd.

A custom store

Any IOutboxStore from @setu-ts/common can back the outbox — an instance or a RegistryFactory. It must keep the bridge’s contract: append writes inside the caller’s scope; scanPending returns pending rows of its own kind in position order, strictly after the cursor, without filtering on availableAt; every transition atomically requires the discriminator and expected status; claim also compares the read version and increments it, while markSent/markFailure require the held version and answer claim-lost on a version miss; markInvalid uses a status-only guard and preserves attempts; purge deletes only settled rows older than the cutoff; verify rejects when the backend cannot serve the relay; and every method rejects rather than throwing synchronously.

Consumer inbox

At-least-once delivery means duplicates by design: an outbox relay that crashes after publishing sends the row again, and every broker redelivers after a lost acknowledgement. The inbox records each event a consumer applied, keyed by (consumer, topic, envelope id), in the SAME database transaction as the handler’s own writes. A duplicate delivery is acknowledged without running the handler; a handler that fails leaves no record, so the redelivery runs it again.

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { SchedulerPlugin } from '@setu-ts/scheduler-plugin';
import {
  createDatabaseInboxStore,
  DatabasePlugin,
  type IUnitOfWork,
} from '@setu-ts/database-plugin';
import {
  defineIntegrationEvent,
  MessagingPlugin,
  onIntegrationEvent,
} from '@setu-ts/messaging-plugin';

const personHired = defineIntegrationEvent<{ personId: string }>({
  type: 'people.hired',
  version: 1,
  topic: 'people.hired.v1',
  parse: (value) => value as { personId: string },
});

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    SchedulerPlugin(), // runs the retention purge
    DatabasePlugin({ type: 'memory' }),
    MessagingPlugin({
      inbox: { store: createDatabaseInboxStore() },
      subscriptions: [
        onIntegrationEvent(
          personHired,
          // The fourth argument is the inbox transaction's unit of work: write
          // through it, and the writes commit with the inbox marker or not at all.
          async (payload, _envelope, _metadata, uow: IUnitOfWork) => {
            await uow.getRepository('PayrollRecord').create({ id: payload.personId });
          },
          { inbox: { consumer: 'payroll' } },
        ),
      ],
    }),
  ],
});
await app.start();

With inbox, onIntegrationEvent returns a RegistryFactory rather than a SubscriptionDefinition — still a valid subscriptions entry, resolved at onInit after the store is verified. A plain subscription without inbox is unchanged. To subscribe imperatively, call the factory with a registry at or after onInit: const d = onIntegrationEvent(def, handler, { inbox })(ctx.services); await broker.subscribe(d.topic, d.handler, d.options);.

MessagingPlugin({ inbox }) registers an IInbox under CAPABILITIES.INBOX (inbox.<name> for a named instance; an inbox subscription names it with inbox.instance) and an inbox health indicator. At onInit it resolves the store, runs its verify() (an unusable backend fails start() by name), and schedules the retention purge (inbox-purge) on CAPABILITIES.SCHEDULER; with no scheduler registered start() rejects InboxPurgeUnscheduledError, unless purge: { schedule: false } (then call inbox.purge() yourself — from a Cron Trigger on Workers).

The consumer name

inbox.consumer is part of the key — with the topic and the envelope id — so it must be the SAME on every replica and every deployment of the consumer: a name that changes makes every redelivered event look new. The topic in the key keeps one consumer’s topics apart, so an event on one topic never suppresses an event with the same id on another. Unless you set queue, the broker queue is inbox. plus 16 hex characters of a 64-bit FNV-1a hash of the consumer and the topic (one queue per pair, legal on every broker; set queue for a readable name). The hash guards against accidental collisions only, so build neither name from untrusted input. Service Bus creates no subscriptions: set queue explicitly there and create that subscription. A named queue matters because on RabbitMQ a queue-less subscriber gets a private queue whose failed messages are discarded, which would leave the inbox nothing to de-duplicate, and a queue shared by two topics hands each handler the other topic’s messages. Two inbox subscriptions with the same consumer name on one topic in one application are refused with InboxConsumerConflictError, because each would skip the other’s events. Across processes this cannot be detected: give each consumer its own name.

The promise

For one consumer name, the handler’s writes through the supplied unit of work are committed at most once per topic and envelope id while the marker is retained; a delivery after the marker is purged is processed again. Nothing is promised about effects outside that unit of work, about two processes using one consumer name with different handlers, or about a database other than the store’s. The inbox is not authentication: anyone who can publish to the topic can still publish a new event.

How it decides: it reads the marker first and acknowledges if one exists; otherwise it opens ONE transaction, creates the marker FIRST, runs the handler, and commits. A concurrent duplicate loses that race on the marker’s primary key — PostgreSQL blocks the second insert until the first transaction ends, a MongoDB replica set rejects the loser with a write conflict, and the deferred backends (memory, D1, DynamoDB) refuse it at commit — and its business writes roll back with it. After ANY rejection the inbox reads the marker again: present means another delivery handled the event, so it acknowledges; absent means the handler failed (a duplicate key in your OWN writes included), and the failure is reported. Both race shapes are driven against real PostgreSQL and a real MongoDB replica set in test/integration/inbox-backends-real.test.ts.

Effects outside the database

Sending an email or calling a provider cannot join the transaction, and the inbox runs nothing after the commit — so an effect performed inside the handler is AT LEAST ONCE: if the commit is refused or the process dies, the effect has already happened and the redelivery runs it again. Choose per effect:

On the backends that defer writes to commit (memory, D1, DynamoDB) the handler runs BEFORE a concurrent duplicate can be refused, so its outside effects run for both; and DynamoDB’s reads are eventually consistent, so a sequential duplicate inside the consistency window can do the same.

Failures and parking

Without maxAttempts (the default) a handler failure is rethrown and the broker’s own budget applies:

With maxAttempts, each failure is counted on a separate row OUTSIDE the rolled-back transaction, and at the limit the delivery is PARKED and acknowledged: its marker becomes parked with the envelope (up to maxParkedEnvelopeBytes), so the partition moves on and every redelivery is skipped. A parse rejection parks at once — redelivery cannot fix it. The count is a lower bound (two concurrent failures can record one increment), so parking comes one attempt late, never early. On RabbitMQ and Redis Streams consumerRetry (default 5 attempts) also applies, and whichever budget runs out first wins: set maxAttempts at or below consumerRetry.maxAttempts, or set consumerRetry: false, or the broker dead-letters the delivery before it is ever parked. On a backend that defers writes, a concurrent delivery’s park can commit while a sibling’s successful transaction is still open; the sibling then rolls back and acknowledges, and the event ends parked — recoverable through release('retry').

import { CAPABILITIES, type IMessageBroker } from '@setu-ts/common';
import type { IInbox } from '@setu-ts/messaging-plugin';

const inbox = app.services.get<IInbox>(CAPABILITIES.INBOX);
const broker = app.services.get<IMessageBroker>(CAPABILITIES.MESSAGING);

for (const parked of await inbox.parked()) {
  // `retry` deletes the parked marker and hands back the stored envelope.
  const { topic, envelope } = await inbox.release(parked.rowId, 'retry');
  // Re-publish WITHOUT a deduplicationId: NATS and Service Bus would drop it
  // inside their de-duplication window.
  if (envelope !== undefined) await broker.publish(topic, envelope);
}

release(rowId, 'discard') keeps the marker as discarded with its envelope cleared, so redeliveries stay skipped. The inbox never re-runs a handler itself — that would bypass the messaging behaviour chain. A re-publish reaches every consumer group of the topic: groups using an inbox skip it, groups without one process it again. parked() never returns an envelope, and the health indicator carries a count only. Parked rows are purged by retainMs like every other row, envelope included, so a parked row does not outlive the window: release or discard a parked delivery within that window (the inbox indicator stays up with parked rows in data.parked and data.reasons), and discard clears one sooner. These are operator capabilities — gate any route that calls them.

Backends

Use createDatabaseInboxStore() from @setu-ts/database-plugin (any IInboxStore works — see A custom inbox store). Its verify() refuses a backend that cannot serve the inbox at startup with InboxStoreUnavailableError:

Every row carries kind: 'setu-inbox', and every read, count, release and purge requires it, so an inbox sharing a table or collection with business documents never touches them. Row ids are a SHA-256 of (consumer, topic, envelope id) — 64 hex characters, or 73 for the failure-count row — whatever id a producer sent; the raw envelope id is stored only when it is a valid publish id (at most 128 UTF-8 bytes, no control or format characters). Protect the inbox table like the broker credentials: anyone who can write to it can mark an event handled.

PostgreSQL (the exact file the real suite applies):

CREATE TABLE setu_inbox (
  id          text    PRIMARY KEY,
  kind        text    NOT NULL,
  consumer    text    NOT NULL,
  topic       text    NOT NULL,
  envelope_id text,
  status      text    NOT NULL,
  attempts    integer NOT NULL,
  updated_at  bigint  NOT NULL,
  last_error  text,
  envelope    text
);
CREATE INDEX setu_inbox_status ON setu_inbox (kind, status, updated_at);

SQLite and Cloudflare D1, where D1Adapter uses the field names as column names:

CREATE TABLE setu_inbox (
  id          TEXT    PRIMARY KEY,
  kind        TEXT    NOT NULL,
  consumer    TEXT    NOT NULL,
  topic       TEXT    NOT NULL,
  envelopeId  TEXT,
  status      TEXT    NOT NULL,
  attempts    INTEGER NOT NULL,
  updatedAt   INTEGER NOT NULL,
  lastError   TEXT,
  envelope    TEXT
);
CREATE INDEX setu_inbox_status ON setu_inbox (kind, status, updatedAt);

Retention

retainMs (default 7 days, at least 60 000) is how long a processed marker is kept. A redelivery older than that is processed again, so it must exceed the broker’s redelivery window plus the outbox’s re-send window (a crashed relay re-sends within its next sweep; an operator’s release('retry') of a failed outbox row re-sends a row no consumer has seen). The purge deletes rows of every status older than the window — processed, discarded and parked markers and failure-count rows — but each run deletes at most purge.batch rows per status (default 100 a minute), so the table stays within the window only while inflow per status stays below that rate — raise purge.batch or shorten purge.intervalMs above it.

Options

Every numeric option must be a finite integer in its range; a violation is refused when MessagingPlugin(...) is called. The handler’s transaction is NOT bounded: an abandoned transaction cannot be cancelled from outside, so the database’s own statement and transaction timeouts govern it.

A custom inbox store

Any IInboxStore from @setu-ts/common can back the inbox. It must keep the bridge’s contract: run creates the marker FIRST inside one transaction and rolls it back with the work on any rejection; a second marker with the same id is refused; find returns undefined for a row of another kind; recordFailure writes outside any transaction; park answers 'exists' when a marker is already present; parked never returns the envelope; release writes only from parked; purge deletes rows of every status, parked included, at most limit per status; verify rejects when the backend cannot serve the inbox; and every method rejects rather than throwing synchronously.

Bridging in-process events

EventsMessagingBridge forwards selected events from @setu-ts/events-plugin onto the broker:

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { EventsMessagingBridge, MessagingPlugin } from '@setu-ts/messaging-plugin';
import { EventsPlugin } from '@setu-ts/events-plugin';

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    EventsPlugin(),
    MessagingPlugin({ broker: 'memory' }),
    EventsMessagingBridge({ eventTypes: ['user.created', 'user.updated'] }),
  ],
});

Health indicator

Registered under the broker’s capability token. Since M70c it reports two signals: the broker’s lifecycle (isReady()) and its reachability. A ready-but-unreachable broker is down with data.reachable: false — the distinction an operator needs to tell “we never started” from “the broker restarted under us”. An unprobeable broker (e.g. the custom arm without isHealthy) is up with data.reachable: 'unknown', honestly reporting “we did not check”.

data reports { broker, reachable }, where reachable is true, false, or 'unknown'.

Since M95b reachability reads the plane the application actually uses. The Service Bus broker records the outcome of every real publish — the data plane — as evidence and reachability() consults it FIRST: a positive success resolves true for ServiceBusOptions.dataPlaneEvidenceMs (default 5000), while a network-layer failure (a rejection carrying no statusCode; a rejected topic or a quota error is an application-level fact, never an outage) resolves false until a successful publish or positive management probe contradicts it. Since M101a that retained failure is answered at once: the management probe runs in the background, and only a true answer clears the outcome, for the NEXT read — before, reachability() awaited the probe, whose own 2 s bound tied the indicator’s, so the indicator’s bound fired first and a recorded outage was reported up. A failure inside a window the indicator has already cached as up is reported at the first poll after that 5 s cache expires. The plane distinction is the substance: the management round trip proves the management plane is reachable — evidence about the data plane, never proof of it — and that gap is what let a stopped namespace report up while every publish threw. Two further changes, and the first has two layers because the health indicator is not the only caller. The indicator wraps every arm’s reachability() in its own createCachedProbe (5-second TTL, 2-second bound), which is the universal outer bound: a probe that cannot answer — a hung broker, the condition a stopped one never produces — settles reachable: 'unknown' instead of holding /health open. RabbitMqBroker, ServiceBusBroker, KafkaBroker and GcpPubSubBroker additionally build their own cached, bounded probes, which is what isHealthy() reads; that inner layer is what protects a direct caller, such as realtime-backplane-plugin’s 'messaging' transport, which resolves the broker itself and never passes through this indicator. Second, the RabbitMQ probe is a real round trip (a throwaway channel open/close), replacing the connection-fault flag read that a hung broker never trips. Residual exposure, stated rather than implied: a deployment whose Service Bus management plane is unreachable and that publishes nothing keeps reporting reachable: 'unknown' with status up until its first publish — an operator who needs the signal can publish synthetically.

A Kafka publish bound covers everything send() waits for, including leader election for a topic the broker auto-creates on first use — measured at over 2 s right after a broker restart — so keep publishTimeoutMs well above that when topics are created on demand.

The Kafka probe uses a second client (<clientId>-health) with zero retries, a 1500 ms request timeout and 1000 ms connection timeout; one probe admin connects at startup and closes at shutdown. An injected factory supplies admin({ retry: { retries: 0 } }) with its own timeouts; without admin, Kafka reachability is unknown. A separate group admin connects on the first subscription, shared across concurrent subscriptions, and closes at shutdown.

Pub/Sub treats any other numeric gRPC code, including 7 (permission denied) and 16 (unauthenticated), as reachable: a publisher-only account need not have pubsub.topics.list, and an authorization response proves the server answered. A failure without a numeric code is false. Unknown reachability still reports up; changing readiness for unknown is outside this milestone.

Exports

Generated from the package barrel by pnpm run docs:exports; pnpm run check:docs fails when it drifts.

Full API

Every export and option is documented in PUBLIC_API.md.

Readiness and dependency outages (M113b)

Inbox parked rows and all six outbox reasons keep the indicator up and /ready at 200. Alert on inbox data.parked / data.reasons and outbox data.reasons (oldest-pending-age, failed-rows, failed-scan-cap, blocked-key-cap, store-write-failing, relay-overlap). Lifecycle and unreachable-store down arms remain.

API reference

Messaging (@setu-ts/messaging-plugin)

Provides message broker abstraction for cross-service integration events.

Registration

import { MessagingPlugin } from '@setu-ts/messaging-plugin';

// In-memory broker (for development/testing)
app.register(
  MessagingPlugin({
    broker: 'memory',
  }),
);

// Redis Streams broker
app.register(
  MessagingPlugin({
    broker: 'redis-streams',
    url: config.get('REDIS_URL'),
    defaultQueue: 'myapp-events',
  }),
);

Connection errors. The 'redis-streams' broker routes the 'error' events of the ioredis client it builds to the application logger through createConnectionErrorReporter (common): the first error of an outage at warn, identical repeats at debug, the recovery at info. Without a listener ioredis prints every reconnect failure to console.error, bypassing the logger and redaction. An injected client gets no listener — it belongs to the caller. Health semantics are unchanged; the indicator still reports the outage.

RabbitMQ consumer recovery (0.9.0). Durable consumer groups retry transient handler failures through durable quorum Q.retry.<delay>ms queues (queue-level TTL, default-exchange DLX back to Q, at-least-once dead-lettering with reject-publish overflow, so a copy is held rather than dropped while Q cannot take it). Delays must be nonempty, nondecreasing positive integer milliseconds ≤2147483647. Total attempts and dead-letter cap must be positive safe integers. Invalid options throw RangeError at construction. Deserialize failures and IntegrationEventRejectedError dead-letter immediately; then isRetryable runs (false dead-letters, throwing logs and retries), then the attempt budget applies. The original is acked after the persistent retry/dead-letter publish is accepted on the same captured channel. Disposition failure leaves it unacked, closes that channel (returning it to Q) and starts the reconnect-and-replay recovery, which re-declares the helper queues; a late confirm or a crash between publish and ack can duplicate it. Handlers must be idempotent. Q.dead carries preserved properties/headers (less expiration, userId, CC and BCC, which the broker would act on) plus x-setu-attempts, x-setu-topic, and a 1 KiB UTF-8 x-setu-error description. Changing the cap requires draining and deleting Q.dead. Private exclusive queues and RPC reply inboxes retain nack-and-discard. consumerRetry: false retains the operator-DLX path. Prefetch is re-applied on reconnect, which re-declares retry and dead queues before consuming. Retry/dead copies are mandatory; correlated basic.return rejects disposition despite a positive confirm. The framework replaces x-setu-disposition-id per copy, preserving the message ID. Recovery requires confirm channels with on/off return listeners and close(), including injected clients: subscribe() refuses a retrying consumer group without them, before declaring anything, rather than letting every failure stay unacked until prefetch stalls the consumer. With retries enabled, group names ending in .dead or .retry.<digits>ms are reserved, and helper names must fit 255 UTF-8 bytes. Names are validated before declaration (at construction for declarative instances). Rename conflicting groups or set consumerRetry: false; migrate existing queues occupying helper names before enabling retries.

Redis Streams recovery (0.9.0). Requires Redis ≥6.2 (XPENDING IDLE). Failed messages redeliver, including after restart, so handlers must be idempotent. The first retry delay must exceed the longest handler runtime: it also guards against another replica stealing in-flight work. Reclaim atomically claims up to ten idle pending entries per pass; delivery metadata and acknowledgement share the new-entry path. Deserialize/missing-payload failures and IntegrationEventRejectedError dead-letter immediately. Other failures use the classifier and total delivery budget. Dead-letter XADD MAXLEN ~ to <topic>.dead.<group> precedes source XACK, with original fields plus x-setu-source-id and x-setu-deliveries; crashes between writes can duplicate entries. Pending consumers survive shutdown, while empty consumers are deleted and old foreign empty consumers are swept.

All counts are positive safe integers; all new millisecond options are positive integers ≤2147483647. Empty/decreasing delay arrays and invalid numeric values throw RangeError at construction. No consumerRetry: false arm exists. Injected IRedisStreamsClient facades require xpending(...args) returning [id, owner, idleMs, deliveries][], xclaim(...args) returning ([id, fields] | null)[], and xinfo(...args) returning consumer field/value arrays, plus xgroup accepting DELCONSUMER and returning a number for that command. Redis 6.2 null claims of trimmed entries are acknowledged; Redis 7 removes them server-side. Empty race-loser claims are never acknowledged. See the messaging README for defaults, retention, ordering, and command cost.

Plugin Options

MessagingPluginOptions is a discriminated union keyed on broker — exactly mirroring packages/messaging-plugin/src/interfaces/index.ts, which is the source of truth. Each arm carries the fields the source defines; the cloud brokers split into an injected-transport arm (no production credentials) and a production arm (credentials required, client typed never). The shared MessagingBrokerType has eight literals.

type MessagingBrokerType =
  'memory' | 'redis-streams' | 'rabbitmq' | 'nats' | 'kafka' | 'pubsub' | 'service-bus' | 'custom';

/** Present on every arm. */
interface MessagingCommonOptions {
  /** Instance name for multi-instance support (registers under messaging.<name>). */
  name?: string;
  /** Serializer for message payloads. @defaultValue new JsonSerializer() */
  serializer?: ISerializer;
  /** Create producer and consumer spans when telemetry is registered. @defaultValue true */
  tracing?: boolean;
}

// ── Default (in-memory). `broker` is optional so MessagingPlugin() and {} are valid. ──
interface MemoryMessagingOptions extends MessagingCommonOptions {
  broker?: 'memory';
}

// version:history
/** Shared retry shape; defaults and lease semantics depend on the broker. @since 0.9.0 */
interface ConsumerRetryOptions {
  readonly maxAttempts?: number;
  readonly delaysMs?: readonly number[];
  readonly isRetryable?: (error: unknown) => boolean;
}

// ── Redis Streams ────────────────────────────────────────────────────────────────────
interface RedisStreamsMessagingOptions extends MessagingCommonOptions {
  broker: 'redis-streams';
  /** Connection URL. */
  url?: string;
  /** Injected client — bypasses the lazy npm import. */
  client?: IRedisStreamsClient;
  /** Default consumer group / queue name. */
  defaultQueue?: string;
  /** XREADGROUP poll interval in ms. */
  pollIntervalMs?: number;
  /** XREADGROUP block timeout in ms. */
  blockSizeMs?: number;
  /** Total delivery budget, including initial delivery. Default 5. */
  consumerRetry?: ConsumerRetryOptions;
  /** Pending reclaim interval, default 5000 ms. */
  reclaimIntervalMs?: number;
  /** Approximate per-group dead-letter MAXLEN, default 10000. */
  deadLetterMaxLen?: number;
  /** Foreign empty consumer inactivity threshold, default 3600000 ms. */
  consumerIdleSweepMs?: number;
  /**
   * Publish acknowledgement bound in ms, from 0 through 2147483647. Zero disables the bound;
   * an expired publish may still be delivered. Consumer reads are unaffected.
   * @defaultValue 15000 @since 0.11.0
   */
  publishTimeoutMs?: number;
}

// ── RabbitMQ ─────────────────────────────────────────────────────────────────────────
interface RabbitMqMessagingOptions extends MessagingCommonOptions {
  // version:history
  /** Group retry policy, default 5 attempts and [5000, 30000, 120000, 600000] ms. @since 0.9.0 */
  consumerRetry?: false | ConsumerRetryOptions;
  // version:history
  /** Q.dead retention cap, default 10000; changing requires draining/deleting Q.dead. @since 0.9.0 */
  deadLetterMaxLength?: number;
  // version:history
  /** Unacked delivery limit per consumer, integer 1–65535, default 32. @since 0.9.0 */
  prefetch?: number;
  broker: 'rabbitmq';
  /** AMQP connection URL. */
  url?: string;
  /** Injected AMQP connection. */
  client?: IAmqpConnection;
  /**
   * Topic exchange name, at most 255 UTF-8 bytes (an AMQP short string; longer
   * is refused at construction). @defaultValue 'messaging'
   */
  exchangeName?: string;
  /**
   * Default consumer group / queue name. A private queue is named
   * `<defaultQueue>-<uuid>`, so it must be at most 218 UTF-8 bytes; longer is
   * refused at construction.
   */
  defaultQueue?: string;
  // version:history
  /**
   * Publish every message persistent (`delivery_mode` 2), so it survives a broker restart.
   * `false` restores the pre-0.9.0 transient publishes. @defaultValue true @since 0.9.0
   */
  persistentMessages?: boolean;
  // version:history
  /**
   * Bound on one `publish()` in ms — every broker round trip, including the publisher confirm.
   * `0` disables it; outside `0`–`2147483647` throws `RangeError` at `MessagingPlugin(...)`.
   * @defaultValue 15000 @since 0.9.0
   */
  publishTimeoutMs?: number;
}

// ── NATS (JetStream) ─────────────────────────────────────────────────────────────────
// version:history — the `@since` below records when the member appeared, not the
// current release; an `@since` is never bumped.
interface NatsMessagingOptions extends MessagingCommonOptions {
  /**
   * Publish acknowledgement bound in ms, from 0 through 2147483647. Zero disables the bound;
   * an expired publish may still be delivered. Passed to the client as its own publish
   * `timeout` and held by the framework's deadline as well; consumer reads are unaffected.
   * @defaultValue 15000 @since 0.11.0
   */
  publishTimeoutMs?: number;
  /** Lazy connection only: defaults -1 (forever) / 2000 ms. @since 0.10.0 */
  reconnect?: { maxAttempts?: number; waitMs?: number };
  broker: 'nats';
  /** NATS connection URL. */
  url?: string;
  /** Injected NATS connection. */
  client?: INatsConnection;
  /** Header factory required for propagation on an injected NATS connection. */
  headersFactory?: () => INatsHeaders;
  /** JetStream stream name. @defaultValue 'MESSAGING' */
  streamName?: string;
  /**
   * Subjects the broker may create the stream with when the stream is absent.
   * No default and no catch-all: NATS refuses `subjects: ['>']` without
   * `no_ack`, and `no_ack` makes every JetStream publish reject unobserved
   * (M90d / X28-2). Supplied and the stream is absent → the broker creates it
   * with exactly these subjects. Absent and the stream is absent → startup
   * rejects with `JetStreamStreamError`, naming the stream and both remedies.
   * An existing stream is never touched either way. An RPC-capable stream must
   * also cover `rr.req.<topic>` and `rr.inbox.>` (M101b); NATS refuses two
   * streams with overlapping subjects, so one stream per account owns
   * `rr.inbox.>`.
   * @since 0.5.0
   */
  streamSubjects?: readonly string[];
  /**
   * Shared queue-less group prefix; the group is `<defaultQueue>.<topic>`.
   * A newly created group starts at new messages; later subscribers resume it.
   * An explicit queue becomes the JetStream durable consumer name with each
   * character the nats client refuses (`.`, `*`, `>`, `/`, `\`, whitespace)
   * escaped as `_` + two hex digits — `orders.eu` → `orders_2eeu` (M101b). A
   * queue whose encoding collides with another queue's, or that is reused
   * across topics, rejects `subscribe()` with
   * `NatsConsumerNameCollisionError`. The raw queue is recorded as consumer
   * metadata, which needs NATS 2.10+.
   * @defaultValue 'messaging-consumers'
   */
  defaultQueue?: string;
}

// ── Kafka ────────────────────────────────────────────────────────────────────────────
interface KafkaMessagingOptions extends MessagingCommonOptions {
  // version:history
  /** Publish bound in ms; default 15000, 0 unbounded. Timeout may still deliver. @since 0.10.0 */
  publishTimeoutMs?: number;

  broker: 'kafka';
  /** Kafka bootstrap brokers. */
  brokers?: readonly string[];
  /** Injected Kafka client factory. */
  client?: IKafkaFactory;
  /** Kafka client ID. @defaultValue 'messaging-client' */
  clientId?: string;
  /**
   * Prefix of the derived default consumer group. Since M90d a subscription
   * with no `queue` uses `<defaultQueue>:<topic>`: members of one consumer
   * group must subscribe the same topics, so the previously shared group
   * collapsed to empty assignments (nothing delivered) the moment an
   * application subscribed two topics. Same topic across instances still
   * load-balances; a caller-supplied `queue` names the group itself.
   * @defaultValue 'messaging-consumers'
   */
  defaultQueue?: string;
  /** Request-reply topic; must already exist on the broker. @defaultValue 'messaging.replies' */
  replyTopic?: string;
  // version:history
  /**
   * Retry budget (M101b). Forwarded to `new Kafka({ retry })` unless `client` is
   * injected, and read by `subscribe()` to retry `UNKNOWN_TOPIC_OR_PARTITION` —
   * which a KRaft broker answers to the request that auto-creates a topic, and
   * which kafkajs does not retry. Exponential: `retries` after the first attempt,
   * from `initialRetryTime`, growing by `multiplier` up to `maxRetryTime`
   * (kafkajs's defaults: 5 / 300 ms / 2 / 30 000 ms). Still unknown after the
   * budget → `KafkaTopicUnavailableError`. Every field is validated at
   * construction, `factor` (kafkajs's jitter) included, held to [0, 1].
   * @since 0.9.0
   */
  retry?: {
    maxRetryTime?: number;
    initialRetryTime?: number;
    /** kafkajs's jitter, between 0 and 1. */
    factor?: number;
    multiplier?: number;
    retries?: number;
  };
}

// ── GCP Pub/Sub — injected transport (client required; credentials optional) ─────────
interface PubSubMessagingOptionsInjected extends MessagingCommonOptions {
  // version:history
  /** Publish bound in ms; default 15000, 0 unbounded. Timeout may still deliver. @since 0.10.0 */
  publishTimeoutMs?: number;

  broker: 'pubsub';
  /** Injected transport (bypasses the lazy SDK load). Required for this arm. */
  client: IPubSubTransport;
  /** GCP project ID. Optional when client is injected. */
  projectId?: string;
  /** Service-account credentials. Optional when client is injected. */
  credentials?: unknown;
  /**
   * Prefix of the per-topic default subscription. Since M101b a subscription with
   * no `queue` uses `<defaultQueue>.<topic ID>` (a fully-qualified topic name
   * contributes only its ID): Pub/Sub subscription names are
   * project-global, so the previously shared name attached a second topic to the
   * first topic's subscription. An existing subscription bound to another topic is
   * refused with `PubSubSubscriptionBoundElsewhereError`; a name over 255
   * characters is refused at `subscribe()`.
   * @defaultValue 'messaging-consumers'
   */
  defaultQueue?: string;
  replyTopic?: string;
  /**
   * Refused on this arm (compile error, and `RangeError` from a standalone
   * `GcpPubSubBroker`): the injected transport creates its own subscriptions, so
   * the broker cannot switch ordering on. Pass `enableMessageOrdering` to
   * `adaptPubSubModule` when building the transport instead.
   */
  enableMessageOrdering?: never;
}

// ── GCP Pub/Sub — production (projectId required; client must be omitted) ────────────
interface PubSubMessagingOptionsProduction extends MessagingCommonOptions {
  // version:history
  /** Publish bound in ms; default 15000, 0 unbounded. Timeout may still deliver. @since 0.10.0 */
  publishTimeoutMs?: number;

  broker: 'pubsub';
  /** GCP project ID. Required for production. */
  projectId: string;
  /** Service-account credentials. SDK ADC is used when omitted. */
  credentials?: unknown;
  /** Mutually exclusive with the injected arm — client?: never. */
  client?: never;
  /** Per-topic default subscription prefix — see the injected arm. @defaultValue 'messaging-consumers' */
  defaultQueue?: string;
  replyTopic?: string;
  /**
   * Create the broker's own subscriptions with message ordering (M106), which a
   * native `orderingKey` needs to be delivered in order. Fixed when a
   * subscription is created and it costs throughput, so it is opt-in.
   * @defaultValue false
   */
  enableMessageOrdering?: boolean;
}

/** GCP Pub/Sub options — exclusive union of injected and production arms. */
type PubSubMessagingOptions = PubSubMessagingOptionsInjected | PubSubMessagingOptionsProduction;

// ── Azure Service Bus — injected transport (client required; credentials optional) ───
interface ServiceBusMessagingOptionsInjected extends MessagingCommonOptions {
  broker: 'service-bus';
  /** Injected transport (bypasses the lazy SDK load). Required for this arm. */
  client: IServiceBusTransport;
  /** Connection string. Optional when client is injected. */
  connectionString?: string;
  /** Administration connection string. Optional when client is injected. */
  adminConnectionString?: string;
  defaultQueue?: string;
  replyTopic?: string;
}

// ── Azure Service Bus — production (connectionString required; client must be omitted) ─
interface ServiceBusMessagingOptionsProduction extends MessagingCommonOptions {
  broker: 'service-bus';
  /** Connection string for the Service Bus namespace. Required for production. */
  connectionString: string;
  /** Connection string for the administration client. Defaults to connectionString. */
  adminConnectionString?: string;
  /** Mutually exclusive with the injected arm — client?: never. */
  client?: never;
  defaultQueue?: string;
  replyTopic?: string;
  /** SDK retry budget for the data client. Optional on this arm only (M90b / X28-6). */
  retryOptions?: ServiceBusRetryOptions;
}

/** Azure Service Bus SDK retry budget — passed ONLY to `ServiceBusClient`, never to the
 * administration client. `mode` is translated to the SDK's numeric `RetryMode` before the
 * client is constructed (the SDK compares the value with `===` against its enum). Omission
 * preserves the Azure SDK default (`maxRetries: 3`, `retryDelayInMs: 30000`,
 * `maxRetryDelayInMs: 90000`, `mode: 'fixed'`, `timeoutInMs: 60000`); `maxRetries: 0` is the
 * documented short budget for a deployment that must fail fast toward a dead broker rather
 * than hold a request for the default chain.
 */
interface ServiceBusRetryOptions {
  maxRetries?: number;
  retryDelayInMs?: number;
  maxRetryDelayInMs?: number;
  mode?: 'fixed' | 'exponential';
  timeoutInMs?: number;
}

ServiceBusOptions — the exported options type of a directly constructed ServiceBusBroker — carries one more member (since M95b): dataPlaneEvidenceMs?: number, how long a positive data-plane outcome stays authoritative. Default 5000. A negative outcome remains authoritative until a successful publish or a positive management probe contradicts it; elapsed time alone never reports a known outage as healthy. It is a broker option rather than a plugin-wide one because no other broker has two planes to choose between; the plugin arm does not forward it.

It must be a safe positive integer, and the constructor REFUSES anything else rather than accepting a value that reads as configured while the window misbehaves. NaN — what Number(env.X) yields for an unset or misspelled variable — would freeze the positive-evidence window, pinning reachability() at one boot-time successful publish; 0 or a negative value would discard every positive outcome instantly. The option therefore has no disable arm: omit it for the default.

/** Azure Service Bus options — exclusive union of injected and production arms. */
type ServiceBusMessagingOptions =
  ServiceBusMessagingOptionsInjected | ServiceBusMessagingOptionsProduction;

// ── Custom — inject any IMessageBroker implementation ────────────────────────────────
interface CustomMessagingOptions extends MessagingCommonOptions {
  broker: 'custom';
  /** Pre-built IMessageBroker instance. Required for this arm. */
  instance: IMessageBroker;
}

/** The factory's single options parameter. */
type MessagingPluginOptions =
  | MemoryMessagingOptions
  | RedisStreamsMessagingOptions
  | RabbitMqMessagingOptions
  | NatsMessagingOptions
  | KafkaMessagingOptions
  | PubSubMessagingOptions
  | ServiceBusMessagingOptions
  | CustomMessagingOptions;

Broker-specific logger options remain string sinks ({ error: (message: string) => void }). When a broker catches a failure it renders the error name, message, safe classifier fields, bounded cause chain, and bounded aggregate members into that string; no structured driver object is passed to the application callback.

Every MessagingPluginOptions arm also inherits these ingress-registration options:

The chain deliberately does not wrap or add a registration arm for respond(): its handler returns a value, unlike the void-returning subscribe handlers, and remains forwarded unchanged.

Publish timing. publish resolves once every matching subscription’s work item has been handed to dispatch — never once every handler has returned — the same guarantee real brokers give, so a plugin publishing during its own register() cannot deadlock startup against the gate. A handler that rejects never rejects the publish and never becomes an unhandled rejection: on the in-memory broker, which has no ack model and no redelivery, the failure path terminates in a report through the application’s logger. InMemoryBrokerOptions.onDispatchError supplies that reporter for applications constructing InMemoryBroker directly (the plugin always supplies one); absent it, the rejection is observed and dropped. A supplied reporter may return a promise; synchronous throws and asynchronous rejections from it are swallowed as the failure-path terminus.

Cloud brokers require production credentials OR an injected transport:

// GCP Pub/Sub — production (requires projectId)
app.register(
  MessagingPlugin({
    broker: 'pubsub',
    projectId: 'my-gcp-project',
  }),
);

// GCP Pub/Sub — injected transport
app.register(
  MessagingPlugin({
    broker: 'pubsub',
    client: myPubSubTransport,
  }),
);

// Azure Service Bus — production (requires connectionString)
app.register(
  MessagingPlugin({
    broker: 'service-bus',
    connectionString: 'Endpoint=sb://...',
  }),
);

// Azure Service Bus — injected transport
app.register(
  MessagingPlugin({
    broker: 'service-bus',
    client: myServiceBusTransport,
  }),
);

Publishing Messages

import { CAPABILITIES } from '@setu-ts/common';

app.router.post('/orders', async (ctx) => {
  const broker = ctx.services.get<IMessageBroker>(CAPABILITIES.MESSAGING);
  const order = await createOrder(ctx.request.body);

  await broker.publish('order.created', {
    orderId: order.id,
    total: order.total,
    customerId: order.customerId,
  });

  return ctx.response.status(201).json(order);
});

Subscribing to Messages

app.register({
  name: 'order-processor',
  version: '1.0.0',
  dependencies: [CAPABILITIES.MESSAGING],
  register(ctx) {
    const broker = ctx.services.get<IMessageBroker>(CAPABILITIES.MESSAGING);

    broker.subscribe(
      'order.created',
      async (message, metadata) => {
        console.log('Processing order', message.orderId);
        await processOrder(message);
      },
      {
        queue: 'order-processor',
      },
    );
  },
});

Request-Reply (RPC)

Beyond fire-and-forget publish/subscribe, the broker supports brokered request-reply: a caller sends a request and awaits a single correlated reply. A responder registered with respond returns the reply value; correlation, the reply channel, and timeout are handled internally.

// Responder side — its resolved value is returned to the caller.
const broker = ctx.services.get<IMessageBroker>(CAPABILITIES.MESSAGING);
await broker.respond<{ userId: string }, { name: string }>('user.lookup', async (req) => {
  const user = await users.findById(req.userId);
  return { name: user.name };
});

// Caller side — resolves with the responder's reply.
const reply = await broker.request<{ userId: string }, { name: string }>(
  'user.lookup',
  { userId: '42' },
  { timeoutMs: 3000 }, // defaults to 5000 when omitted
);

Signatures (on IMessageBroker, from @setu-ts/common):

interface RequestOptions {
  /** Reply wait budget in ms. @defaultValue 5000 */
  readonly timeoutMs?: number;
}

type RequestHandler<TReq = unknown, TRes = unknown> = (
  message: TReq,
  metadata: MessageMetadata,
) => TRes | Promise<TRes>;

request<TReq, TRes>(topic: string, message: TReq, options?: RequestOptions): Promise<TRes>;
respond<TReq, TRes>(
  topic: string,
  handler: RequestHandler<TReq, TRes>,
  options?: SubscribeOptions,
): Promise<ISubscription>;

Pass options.queue to respond to load-balance requests across competing responders.

request rejects with one of four exported error classes (import from @setu-ts/messaging-plugin for instanceof handling):

Two more error classes are exported for the NATS broker’s startup prerequisites (M90d / X28-3). Each carries the platform error as cause when there is one — the absent-stream refusal below is a configuration verdict this broker reaches on its own, so that one arm has no cause:

Three more name failures the real brokers impose and a permissive fake accepts (M101b). Each is thrown by subscribe(), so a declared subscription rejects start() with it:

Broker support. Request-reply is available on all supported broker types — in-memory, Redis Streams, RabbitMQ, NATS, Kafka, GCP Pub/Sub, Azure Service Bus, and custom (which delegates to the injected IMessageBroker).

Kafka has one operational prerequisite. Replies travel on a shared reply topic (replyTopic, default 'messaging.replies') which must already exist — the broker creates no topics, so either pre-create it or enable auto.create.topics.enable; the same holds for every subscribed topic, whose absence on a non-auto-creating broker rejects with KafkaTopicUnavailableError. Each broker instance reads that topic under its own consumer group, so every instance receives every reply and discards those it did not originate; give a high-traffic service its own replyTopic to bound that fan-out.

RPC and pub/sub are separate channels. request/respond travel on a channel derived from the topic, not on the topic itself. A plain subscribe('orders', …) therefore never observes an RPC request, and a plain publish('orders', …) is never consumed by a responder on 'orders'. The two can share a topic name safely.

Multiple Broker Instances

import { CAPABILITIES } from '@setu-ts/common';

app.register(
  MessagingPlugin({
    broker: 'redis-streams',
    name: 'events',
    url: config.get('EVENTS_REDIS_URL'),
    defaultQueue: 'events',
  }),
);

app.register(
  MessagingPlugin({
    broker: 'redis-streams',
    name: 'audit',
    url: config.get('AUDIT_REDIS_URL'),
    defaultQueue: 'audit',
  }),
);

// Access by namespaced token
const eventsBroker = ctx.services.get<IMessageBroker>('messaging.events');
const auditBroker = ctx.services.get<IMessageBroker>('messaging.audit');

Events Messaging Bridge

The EventsMessagingBridge forwards domain events from EventsPlugin to a messaging broker:

import { EventsMessagingBridge } from '@setu-ts/messaging-plugin';

app.register(
  EventsMessagingBridge({
    eventTypes: ['user.created', 'user.updated'],
    token: CAPABILITIES.MESSAGING,
    topicMapping: (eventType) => eventType.toLowerCase(),
    errorHandler: (error, eventType) => {
      console.error(`Failed to forward ${eventType}:`, error);
    },
  }),
);

Exports

// Plugin factories
export { MessagingPlugin } from '@setu-ts/messaging-plugin';
export { EventsMessagingBridge } from '@setu-ts/messaging-plugin';

// Broker implementations
export { InMemoryBroker } from '@setu-ts/messaging-plugin';
export { RedisStreamsBroker } from '@setu-ts/messaging-plugin';
export { RabbitMqBroker } from '@setu-ts/messaging-plugin';
export { NatsBroker } from '@setu-ts/messaging-plugin';
export { KafkaBroker } from '@setu-ts/messaging-plugin';
export { GcpPubSubBroker } from '@setu-ts/messaging-plugin';
export { ServiceBusBroker } from '@setu-ts/messaging-plugin';

// Adapter / load helpers. The `*SdkModule` types describe the SDK shape each
// `adapt*` consumes, so a consumer can type a substitute module.
export { adaptPubSubModule, loadPubSubModule } from '@setu-ts/messaging-plugin';
export { adaptServiceBusModule, loadServiceBusModule } from '@setu-ts/messaging-plugin';
export type { PubSubSdkModule, ServiceBusSdkModule } from '@setu-ts/messaging-plugin';

// Serializer
export { JsonSerializer } from '@setu-ts/messaging-plugin';
export type { ISerializer } from '@setu-ts/messaging-plugin';

// Request-reply, gate, publish-timeout, and NATS-prerequisite error classes
export {
  ChainGateTimeoutError,
  CloudBrokerUnavailableError,
  JetStreamStreamError,
  JetStreamUnavailableError,
  KafkaTopicUnavailableError,
  MessagingNotSupportedError,
  NatsConsumerNameCollisionError,
  PubSubSubscriptionBoundElsewhereError,
  PublishTimeoutError,
  RemoteHandlerError,
  ReplyInboxUnavailableError,
  RequestTimeoutError,
} from '@setu-ts/messaging-plugin';

// Option types
export type {
  ConsumerRetryOptions,
  CustomMessagingOptions,
  EventsMessagingBridgeOptions,
  InMemoryBrokerOptions,
  KafkaMessagingOptions,
  KafkaOptions,
  MemoryMessagingOptions,
  MessagingBrokerType,
  MessagingCommonOptions,
  MessagingPluginOptions,
  NatsMessagingOptions,
  NatsOptions,
  PubSubMessagingOptions,
  RabbitMqMessagingOptions,
  RabbitMqOptions,
  RedisStreamsMessagingOptions,
  RedisStreamsOptions,
  ServiceBusMessagingOptions,
  ServiceBusRetryOptions,
} from '@setu-ts/messaging-plugin';

// Port types (structural)
export type {
  IPubSubSubscription,
  IPubSubTransport,
  PubSubOptions,
} from '@setu-ts/messaging-plugin';
export type {
  IServiceBusProcessErrorArgs,
  IServiceBusReceiver,
  IServiceBusSubscribeOptions,
  IServiceBusSubscription,
  IServiceBusTransport,
  ServiceBusOptions,
} from '@setu-ts/messaging-plugin';

// Transactional outbox (M107)
export {
  OutboxEnvelopeTooLargeError,
  OutboxNotReadyError,
  OutboxRelayUnscheduledError,
  OutboxRowStateError,
  OutboxStoreVerifyTimeoutError,
  OutboxUnknownTenantError,
} from '@setu-ts/messaging-plugin';
export type {
  IOutbox,
  OutboxCommonOptions,
  OutboxHealthOptions,
  OutboxOptions,
  OutboxRelayOptions,
  OutboxStoreEntry,
  OutboxSweepResult,
  OutboxWriteInput,
} from '@setu-ts/messaging-plugin';

// Consumer inbox (M108)
export {
  InboxConsumerConflictError,
  InboxNotConfiguredError,
  InboxNotReadyError,
  InboxPurgeUnscheduledError,
  InboxRowStateError,
  InboxStoreVerifyTimeoutError,
} from '@setu-ts/messaging-plugin';
export type {
  IInbox,
  InboxOptions,
  InboxPurgeOptions,
  InboxReleaseResult,
  InboxStoreEntry,
  IntegrationEventInboxHandler,
  IntegrationEventInboxOptions,
  IntegrationEventSubscribeOptions,
  ParkedInboxEntry,
} from '@setu-ts/messaging-plugin';

// Re-exported types from @setu-ts/common
export type {
  IMessageBroker,
  ISubscription,
  MessageHandler,
  MessageMetadata,
  SubscribeOptions,
} from '@setu-ts/messaging-plugin';

Kafka Commit Model: Kafka uses the producer/consumer commit model — handler success auto-commits; a thrown handler prevents commit.

Publish options — ordering key, de-duplication ID and headers

publish takes an optional trailing PublishOptions (from @setu-ts/common):

interface PublishOptions {
  readonly orderingKey?: string; // ≤ 128 UTF-8 bytes
  readonly deduplicationId?: string; // ≤ 128 UTF-8 bytes
  readonly headers?: Readonly<Record<string, string>>; // ≤ 32 headers, ≤ 1024 bytes each
}

publish<T>(topic: string, message: T, options?: PublishOptions): Promise<void>;

Every option is carried as a transport header beside any native mapping, so it is observable through MessageMetadata.headers on every broker and never silently dropped. Use the exported constants ORDERING_KEY_HEADER ('x-setu-ordering-key') and DEDUPLICATION_ID_HEADER ('x-setu-deduplication-id') to read them.

What orderingKey promises. It decides placement — the same key reaches the same partition, retry queue or ordered subscription where the broker has one — not the order handlers finish in. What a handler failure does to order differs per broker:

• Order kept by blocking: Kafka (a throwing handler leaves the offset uncommitted, so one failing message stalls its whole partition until it succeeds) and Pub/Sub on an ordering subscription.
• Service Bus — the condition decides, so it is always stated. Measured 2026-10-07 both ways: at the broker’s default maxConcurrentCalls (1) the abandoned message is redelivered before the next is handled (order kept by blocking); at maxConcurrentCalls: 2 the later message is handled first (order lost).
• Order lost on retry: RabbitMQ, Redis Streams and NATS redelivery — later messages for the key are handled while the failed one waits.

A consumer that needs order compares the delivered envelope’s version and drops or defers a stale message.

What a consumer may trust, and what a producer must not derive.

1. A delivered x-setu-ordering-key or x-setu-deduplication-id header is a hint written by whoever published the message — validation runs on the publish side only — so a foreign or compromised producer can send any value under these names. Use it to order or de-duplicate your own work, never to authorize anything.
2. A deduplicationId derived from request input lets the caller who chooses it suppress another message with the same id for the broker’s window (NATS’s duplicate_window, 120 s by default; Service Bus’s configured detection window). Derive it from a producer-assigned id — the envelope id publishIntegrationEvent uses by default is one.
3. An orderingKey derived from request input lets a caller concentrate load on one Kafka partition or one Pub/Sub ordering key (1 MB/s per key), and on a log-compacted Kafka topic it lets the caller erase an earlier message that carries the same key. Derive it from an aggregate the application owns.

Invalid options are rejected with a RangeError as a rejected promise (never a synchronous throw), naming the field and the rule; a refused value is never echoed. A header name a broker or its server acts on (traceparent, tracestate, cc, bcc, payload, nats-*, x-setu-*, RabbitMQ’s x-death / x-delivery-count / x-acquired-count / x-delay and the x-first-death-* / x-last-death-* forms) is refused on every broker, compared case-insensitively. A goog prefix is refused too, as a precaution: Pub/Sub’s reservation of it is stated only by third-party documentation, and the emulator accepts goog attributes.

The Cloudflare Workers broker has no transport header channel, so it carries all three options as envelope fields and surfaces them on delivery as the same headers. On publish it refuses invalid options with the same RangeError as every other broker; on delivery, an envelope orderingKey, deduplicationId or header entry that fails the shared rules — or every caller header, when the record holds more than MAX_PUBLISH_HEADERS entries — is dropped, never surfaced, since a foreign producer may have written the envelope.

One exception to “never silently dropped” above: a NatsBroker given an injected connection and no headersFactory has nothing to build NATS headers with, so caller headers and the ordering key are dropped and reported once through the logger; the de-duplication id is still applied, as nats.js’s native msgID.

Trace context across the broker

When CAPABILITIES.TELEMETRY is registered, the plugin wraps the broker so publish, subscribe, request and respond create producer and consumer spans, writing a W3C traceparent on publish and parenting delivery from the header it reads back. MessagingPlugin({ tracing: false }) opts out; with no telemetry capability registered, behaviour is unchanged.

Each broker uses the header channel its transport actually provides:

MessageMetadata.headers is populated by every first-party broker: it carries the headers the broker read, and {} when the transport carried none. Three transports can deliver values that are not strings — AMQP field tables carry numbers, booleans, timestamps and byte arrays; Service Bus types its application properties number | boolean | string | Date | null; kafkajs delivers Buffer and permits arrays — so each is normalized to satisfy the declared Record<string, string>. A byte value is decoded as UTF-8, a number or boolean is stringified, a Date becomes ISO-8601, and the first element of a repeated header is taken. A value with no faithful string form (a nested table, null, NaN) is dropped rather than rendered as [object Object], so a missing key reads as absent rather than as a corrupted value. It is absent only for a 'custom' broker that does not supply it. Branch on emptiness rather than on the member’s presence.

Two cases drop the header rather than propagating, both by construction:

• NATS with an injected client. A MsgHdrs can only be built by the nats module’s headers() function, and an injected connection carries no module. Pass NatsOptions.headersFactory (for example () => nats.headers()) alongside the client. Without it the broker publishes normally and reports the dropped headers once through its logger.
• A Pub/Sub or Service Bus transport injected via client that predates this release. IPubSubTransport.publish and IServiceBusTransport.send gained an optional third parameter (attributes / applicationProperties), and their delivered-message callbacks gained a matching optional member. A two-parameter implementation stays assignable and simply ignores the header. M90d widened the delivered callbacks further with optional messageId and timestamp members (X28-4); the same assignability holds — an implementation supplying neither simply produces metadata without them.

Message identity on the metadata

Every first-party broker populates MessageMetadata.messageId and .timestamp from what its transport actually assigns (M90d / X28-4 closed the two cloud brokers that were not reading them): in-memory assigns runtime.uuid() and runtime.now(); Redis Streams uses the entry id and its timestamp; RabbitMQ uses properties.messageId/properties.timestamp (assigned on publish when the producer did not); NATS uses the JetStream stream sequence and the delivery’s info.timestampNanos; Kafka uses partition:offset and the record timestamp; Pub/Sub reads message.id/message.publishTime; Service Bus reads message.messageId/message.enqueuedTimeUtc. An absent member means the transport carried none — never “the adapter did not look” — so a de-duplication read of metadata.messageId is meaningful on all seven.

Health status

Since M70c the indicator reports two signals: the broker’s lifecycle (isReady()) and its reachability. A ready-but-unreachable broker is down with data.reachable: false — the distinction an operator needs to tell “we never started” from “the broker restarted under us”. An unprobeable broker (e.g. the custom arm without isHealthy) is up with data.reachable: 'unknown', honestly reporting “we did not check”.

Since M90b the Service Bus broker’s transport implements the probe it documents: one administration round trip (getNamespaceProperties()) proves the namespace is reachable, and a positively identified 401/403 counts as reachable too — the namespace ANSWERED with an auth verdict, which a send/listen-only credential legitimately produces and which is not a network outage. Every broker’s probe is cached for 5 seconds and bounded at 2 seconds through createCachedProbe, so a dead broker can no longer hold a health poll for the full transport timeout, and repeated polls cost at most one round trip per TTL.

data reports { broker, reachable }, where reachable is true, false, or 'unknown'.

Since M95b reachability reads the plane the application actually uses. The Service Bus broker records the outcome of every real publish — the data plane — and reachability() consults it FIRST: a positive success resolves true for ServiceBusOptions.dataPlaneEvidenceMs (default 5000), while a network-layer failure (a rejection carrying no statusCode; a rejected topic or a quota error is an application-level fact, never an outage) resolves false until a successful publish or a positive management probe contradicts it. Since M101a that retained failure is answered at once: the management probe runs in the background and only a true answer clears the outcome, for the NEXT read. Before, reachability() awaited the probe, whose own 2-second bound tied the indicator’s, so the indicator’s bound fired first and a recorded outage was reported up. A failure inside a window the indicator has already cached as up is reported at the first poll after that 5-second cache expires. The plane distinction is the substance: the management round trip proves the management plane is reachable — evidence about the data plane, never proof of it — and that gap is what let a stopped namespace report up while every publish threw. Two further changes: every arm’s probe is bounded by the indicator’s createCachedProbe (5-second TTL, 2-second bound), so a probe that cannot answer — a hung broker, the condition a stopped one never produces — settles reachable: 'unknown' instead of holding /health open; and the RabbitMQ probe is a real round trip (a throwaway channel open/close), replacing the connection-fault flag read that a hung broker never trips. Residual exposure, stated rather than implied: a deployment whose Service Bus management plane is unreachable and that publishes nothing keeps reporting reachable: 'unknown' with status up until its first publish — an operator who needs the signal can publish synthetically.

Kafka and Pub/Sub also own cached probes (5 s TTL, 2 s bound), protecting direct isHealthy() callers. Kafka uses describeCluster() on a second <clientId>-health client (zero retries, 1500 ms request timeout, 1000 ms connection timeout); a request-timeout error anywhere within five cause links means unknown, another rejection means false. A faulted supervisor returns false immediately. An injected IKafkaFactory.admin?(options?: { retry?: { retries: number } }): unknown provides the probe admin; injected timeouts belong to the application. Without it reachability is unknown. The probe admin connects at startup and disconnects at shutdown. The separate group admin connects lazily through one memoized promise and closes at shutdown.

Pub/Sub uses getTopics({ pageSize: 1, autoPaginate: false, gaxOpts: { timeout: 1500, retry: null } }). PubSubSdkModule includes this SDK member; IPubSubTransport.isHealthy?() returns Promise<boolean | undefined>. Resolution means true; gRPC 14 means false, 4 means unknown, any other numeric code (including 7/16 authorization responses) means true; no numeric code means false. An omitted transport probe means unknown. Unknown still reports up. See the per-broker probe table.

The Kafka, Pub/Sub, RabbitMQ, Redis Streams and NATS plugin arms — and their KafkaOptions, PubSubOptions, RabbitMqOptions, RedisStreamsOptions and NatsOptions shapes — accept publishTimeoutMs (default 15000 ms, 0 unbounded). Out-of-range or non-finite values throw RangeError at MessagingPlugin(...) and at broker construction. Every publish entry point, RPC and outbox relay publishes use the bound; the timeout is the exported PublishTimeoutError, which names the broker, topic and bound and warns that delivery may still occur. It does not cancel the transport. On Redis Streams the bound covers the publish’s XADD only; on NATS it is passed to the client as its own publish timeout as well.

Fresh Kafka groups whose every partition offset is -1 capture high watermarks before run(). On the first join, members seek assigned, still-uncommitted partitions to the minimum start offset advertised by current members. Messages published after any member’s subscribe() resolves reach the group even during concurrent first starts; history before the first member’s subscription is not replayed, provided every group member is trusted: any client able to join the group can move a new group’s start earlier (never later), without a log line. Joining needs only Read on the group while committing also needs Read on the topic, so under split ACLs a principal that cannot commit can still cause that replay; restrict group membership. Advertisements are keyed by topic; member metadata above 64 KiB and offsets longer than 19 digits are refused, and the first refusal per subscription is logged. subscribe() does not wait for the join. Existing committed groups and later joins are untouched. This requires admin and a loadable kafkajs module. Injected factories try loading the module once at connect(); without it members seek only their own offsets and concurrent first starts can miss messages. Without admin the step is skipped. A member crashing before the join takes its start offsets with it. Admin/metadata failures warn once per subscription through KafkaOptions.logger.warn? (fallback error), naming topic/group; invalid member metadata is skipped individually while usable snapshots still contribute to the minimum. Admin or seek failures leave affected partitions with latest-offset behavior.

NatsOptions.reconnect and NatsMessagingOptions.reconnect configure only lazily built connections: maxAttempts defaults to -1 (unbounded), waitMs to 2000; forwarded as maxReconnectAttempts / reconnectTimeWait. The former must be an integer ≥ -1; the latter an integer 1–2147483647, else construction throws RangeError naming the field. Supplying an injected client together with reconnect throws TypeError. A finite exhausted budget closes the connection; the indicator stays down until the application replaces it.

Integration event contracts

The contract layer ON TOP OF the transport: a declarable, versioned integration event with a portable wire envelope — payload data, never transport headers, so every broker arm carries it without an adapter change. No IMessageBroker method is added and no capability token is added.

import {
  causedBy,
  defineIntegrationEvent,
  onIntegrationEvent,
  publishIntegrationEvent,
} from '@setu-ts/messaging-plugin';

// The contract: four required fields (plus an optional `orderingKey`
// selector, below), validated eagerly. The topic MUST end with the
// exact `.v${version}` suffix — the versioned-topic rollout policy enforced
// by the factory, not documented prose. A pre-existing unversioned topic has
// no definition; keep using the raw broker.publish/subscribe surface for it.
const orderPlaced = defineIntegrationEvent<{ orderId: string }>({
  type: 'orders.placed',
  version: 1,
  topic: 'orders.placed.v1',
  // `parse` is the validation boundary: a bare `as` checks nothing at runtime.
  // Narrow it for real — see the messaging-plugin README for a worked parser.
  parse: (value) => value as { orderId: string },
});

// Producer: builds the envelope (below) and publishes it to the definition's
// topic. parse is NEVER run on publish — a producer CAN publish a payload its
// own consumers reject, and that surfaces at the consumer.
await publishIntegrationEvent(
  runtime,
  broker,
  orderPlaced,
  { orderId: '123' },
  {
    correlationId: 'root-1', // each field optional; omitted when absent
    causationId: 'e-0',
    aggregateId: 'order-1',
    aggregateVersion: 3,
  },
);

// Consumer: produces the existing SubscriptionDefinition — plugs straight
// into MessagingPlugin({ subscriptions }). The handler receives the parsed
// payload, the envelope (rebuilt with data = the parsed value), and the
// transport metadata.
const subscription = onIntegrationEvent(
  orderPlaced,
  async (payload, envelope) => {
    await provisionOrder(payload.orderId, envelope.id);
  },
  { queue: 'billing' },
);
// provisionOrder: the application's own work, resolved from the parsed
// payload; envelope carries the identity/correlation fields; the third
// handler argument is the transport MessageMetadata.

// Correlation propagation: the chain-root rule, extracted.
const causal = causedBy(envelope); // { correlationId: envelope.correlationId ?? envelope.id, causationId: envelope.id }

Publish options (M106). publishIntegrationEvent takes an optional sixth argument, PublishOptions (see “Publish options” above), validated once as on every broker. Two defaults differ from a raw broker.publish:

• deduplicationId defaults to the envelope id, so a retried publish of the SAME envelope is de-duplicated where the broker supports it. A caller-supplied deduplicationId wins.
• orderingKey comes from the caller first, then from the definition’s optional orderingKey(envelope) selector, then none. The selector is called only when the caller supplied no key; a selector that throws, or returns a value the rules refuse, rejects the publish.

const orderShipped = defineIntegrationEvent<{ orderId: string }>({
  type: 'orders.shipped',
  version: 1,
  topic: 'orders.shipped.v1',
  parse: (value) => value as { orderId: string },
  // Keep every event for one order on one partition / ordering key.
  orderingKey: (envelope) => `order-${envelope.data.orderId}`,
});

await publishIntegrationEvent(runtime, broker, orderShipped, { orderId: '123' }, undefined, {
  headers: { 'x-tenant': 'acme' },
});

The wire envelope, published as the message payload:

occurredAt is a string, not a Date: payloads round-trip through the serializer’s JSON.parse, so a Date would arrive as a string on every transport. The consumer’s structural check REQUIRES the five mandatory fields at their primitive types, checks type and version for exact equality, and IGNORES unknown extra top-level fields — an additive envelope change is never a breaking deployment. Payload strictness belongs to parse, where the application owns the policy. The optional causal fields are checked on both sides: correlationId, causationId and aggregateId must be strings, and aggregateVersion — the only number among them — must be FINITE, because JSON serialization maps NaN/Infinity to null, so a non-finite one would reach consumers as null.

occurredAt must be an ISO-8601 instant in the interoperable RFC 3339 profile — a full date, a time to at least seconds, and an explicit Z or a numeric offset (2026-01-01T00:00:00Z, 2026-01-01T00:00:00.000Z, 2026-01-01T00:00:00+05:30). A date-only value, an RFC 2822 date, or a zone-LESS timestamp is refused: Date.parse accepts all three, and the zone-less form is read in each engine’s own local time — measured, 2026-01-01T00:00:00 becomes 2025-12-31T18:30:00.000Z on a +05:30 host — so one event would mean a different instant on every consumer.

The ten exported symbols of this section: the four functions above (defineIntegrationEvent, publishIntegrationEvent, onIntegrationEvent, causedBy), the IntegrationEventRejectedError class (rejections, below), and five types:

• IntegrationEventDefinition<T> (interface) — the contract defineIntegrationEvent returns and both directions read: readonly type, version, topic, parse: (value: unknown) => T, and the optional orderingKey?(envelope): string | undefined selector (M106). parse runs on the consumer side only.
• IntegrationEventEnvelope<T> (interface) — the wire shape in the table above. The handler receives it with data rebuilt to the parsed value, so envelope.data === payload holds on every delivery.
• IntegrationEventMetadata (interface) — the optional fifth argument of publishIntegrationEvent: correlationId/causationId/aggregateId (string) and aggregateVersion (number), each omitted from the envelope when absent. causedBy() returns a value assignable to it.
• IntegrationEventHandler<T> (type) — (payload: T, envelope: IntegrationEventEnvelope<T>, metadata: MessageMetadata) => void | Promise<void>: parsed payload first, envelope second, transport metadata third.
• onIntegrationEvent overloads (M108). With options.inbox the call takes an IntegrationEventInboxHandler<T, S> — the same three arguments plus the inbox transaction’s unit of work, S inferred from the handler’s annotation — and returns a RegistryFactory<SubscriptionDefinition> (see Consumer inbox). Without it the call is unchanged and returns a SubscriptionDefinition; the inbox overload is declared first, and the legacy options take inbox?: never, so an options variable carrying inbox is typed as the factory it is.
• IntegrationEventRejectionReason (type) — 'malformed' | 'type-mismatch' | 'version-mismatch' | 'parse', the reason discriminant of IntegrationEventRejectedError (table below).