@atlas-id/event-bus

Production-ready event bus for NATS with Transactional Outbox support.

Highlights

• NATS publish/subscribe with queue groups
• Transactional Outbox processor for reliable delivery
• Dead letter queue handling and retry controls
• TypeScript-first API with shared contracts
• ESM + CJS builds

Install

pnpm add @atlas-id/event-bus

npm install @atlas-id/event-bus

Quick start

import { NatsEventBus } from "@atlas-id/event-bus";
import type { EventEnvelope } from "@atlas-id/contracts";

const eventBus = new NatsEventBus("nats://localhost:4222", {
  maxReconnectAttempts: 10,
  reconnectTimeWait: 2000,
  name: "my-service",
  logger: console,
});

await eventBus.connect();

const event: EventEnvelope<{ userId: string }> = {
  id: "event-123",
  type: "user.created",
  version: "1.0.0",
  occurredAt: new Date().toISOString(),
  payload: { userId: "user-456" },
  meta: { source: "users-service" },
};

await eventBus.publish("user", event);

const subscription = await eventBus.subscribe(
  "user",
  async (evt) => {
    console.log("Received event:", evt);
  },
  { queue: "user-processors", maxDeliveries: 3 }
);

await subscription.unsubscribe();
await eventBus.close();

Transactional Outbox

import { OutboxProcessor, PrismaOutboxRepository } from "@atlas-id/event-bus";
import { PrismaClient } from "@prisma/client";

const prisma = new PrismaClient();
const repository = new PrismaOutboxRepository(prisma);

const processor = new OutboxProcessor(eventBus, repository, {
  pollIntervalMs: 5000,
  batchSize: 10,
  maxRetries: 3,
  logger: console,
});

processor.start();

Notes:

• OutboxProcessor derives the publish topic from eventType by removing the last segment (orders.created.v1 -> orders.created).
• Failed events are marked and can be routed to a DLQ by your bus implementation.

Configuration

NatsEventBus options

• maxReconnectAttempts (default 10)
• reconnectTimeWait (default 2000ms)
• name client name used by NATS
• logger optional custom logger

OutboxProcessor options

• pollIntervalMs (default 5000ms)
• batchSize (default 10)
• maxRetries (default 3)
• logger optional custom logger

Logger interface

export interface Logger {
  debug(message: string, meta?: Record<string, unknown>): void;
  info(message: string, meta?: Record<string, unknown>): void;
  warn(message: string, meta?: Record<string, unknown>): void;
  error(message: string, meta?: Record<string, unknown>): void;
}

Development

pnpm install
pnpm run ci
pnpm run build
pnpm run sbom

Requirements

• Node.js >= 18
• NATS Server 2.x+
• Prisma 5.x (optional, only for PrismaOutboxRepository)

License

UNLICENSED - Proprietary software.