NATS Event Bus

The @noddde/nats package uses the nats client with JetStream to provide at-least-once delivery, durable consumers, and natural backpressure.

Installation

yarn add @noddde/nats nats

Configuration

import { NatsEventBus } from "@noddde/nats";

const eventBus = new NatsEventBus({
  servers: "localhost:4222",
  consumerGroup: "my-service", // required — scopes durable consumer names
  streamName: "noddde-events", // optional — enables JetStream durable subscriptions
  subjectPrefix: "noddde.", // optional — subjects become "noddde.AccountCreated"
  prefetchCount: 256, // optional — backpressure control
  resilience: { maxAttempts: -1, initialDelayMs: 2000, maxRetries: 10 }, // optional — connection + delivery resilience
});

How It Works

• Publishing: Events are serialized as JSON, encoded as Uint8Array, and published to a NATS subject (${subjectPrefix}${event.name}). When a streamName is configured, JetStream publish acknowledgment is awaited.
• Subscribing: JetStream consumers with durable subscriptions are created for each registered event name, using the durable name pattern ${consumerGroup}_${eventName}. Two services with different consumerGroup values get independent consumers on the same stream. Handlers registered before connect() are buffered; subscriptions are activated during connect() — if any subscription fails, connect() rejects immediately (fail-fast). Handlers registered after connect() create subscriptions immediately (failures are logged, not thrown). Multiple handlers for the same event are invoked in parallel via Promise.allSettled() — every handler runs to completion even when some fail, and each failure is logged individually with structured fields (see Handler Error Isolation).
• Delivery: At-least-once with JetStream. Messages are acknowledged only after all handlers complete successfully. If any handler fails, the first rejection is re-thrown after all siblings have settled, and the consumer loop calls msg.nak() for immediate redelivery (capped by maxDeliver). Consumers must be idempotent — handlers that completed before a sibling failure will re-execute on redelivery.
• Poison message protection: Malformed messages that fail JSON deserialization are permanently discarded via msg.term(), preventing corrupt data from blocking the subscription via infinite redelivery. If resilience.maxRetries is configured, it maps to the JetStream maxDeliver consumer option, limiting how many times NATS will redeliver a message before discarding it server-side.
• Backpressure: The prefetchCount option controls how many unacknowledged messages the server delivers to each consumer (mapped to JetStream maxAckPending), providing natural backpressure when handlers are slow.
• Connection resilience: NATS native reconnection is enabled by default. The optional resilience config maps maxAttempts to NATS maxReconnectAttempts and initialDelayMs to reconnectTimeWait for automatic recovery on connection loss. maxDelayMs is ignored because NATS uses fixed-interval reconnection (no exponential backoff).
• Logging: All internal logging uses the framework Logger interface from @noddde/core. Pass a custom logger in the config or accept the default (NodddeLogger("warn", "noddde:nats")). All log calls include structured context data (event name, error details).
• Shutdown: close() calls nc.drain() to process in-flight messages before disconnecting.

Wiring with Domain

import { wireDomain } from "@noddde/engine";
import { NatsEventBus } from "@noddde/nats";

const domain = await wireDomain(myDomain, {
  buses: () => ({
    eventBus: new NatsEventBus({
      servers: "localhost:4222",
      consumerGroup: "my-service",
      streamName: "noddde-events",
    }),
  }),
});