@auto-engineer/message-bus

Type-safe message bus implementing the CQRS pattern with command handling and event publishing.

Purpose

Without @auto-engineer/message-bus, you would have to implement your own command/event routing, handle handler registration, manage request/correlation ID propagation, and ensure proper error isolation between handlers.

This package provides the core messaging infrastructure for the Auto Engineer ecosystem. It enables decoupled communication through commands (write operations) and events (state changes). Commands have exactly one handler; events fan out to many subscribers. Correlation IDs flow automatically from commands to their resulting events, enabling end-to-end traceability.

Key Concepts

• One handler per command type -- ensures deterministic command processing. Registering a second handler for the same command throws.
• Multiple handlers per event type -- enables fan-out notification. Handlers run concurrently via Promise.allSettled, so one failure does not block others.
• Request/Correlation ID propagation -- when a command handler returns events, the bus copies requestId and correlationId from the command onto each event (unless the event already carries its own).
• Correlation listeners -- subscribe to events by exact correlationId or by prefix, useful for tracking the progress of a specific workflow or job graph.
• defineCommandHandler -- a factory that attaches CLI metadata (alias, description, fields, examples) to a command handler, so the same object can drive both the message bus and a CLI interface.

Installation

pnpm add @auto-engineer/message-bus

Quick Start

import { createMessageBus, defineCommandHandler } from '@auto-engineer/message-bus';
import type { Command, Event } from '@auto-engineer/message-bus';

// 1. Create the bus
const bus = createMessageBus();

// 2. Define and register a command handler
type CreateUser = Command<'CreateUser', { name: string }>;
type UserCreated = Event<'UserCreated', { userId: string; name: string }>;

const handler = defineCommandHandler<CreateUser, (cmd: CreateUser) => Promise<UserCreated>>({
  name: 'CreateUser',
  alias: 'create-user',
  description: 'Creates a new user',
  fields: {
    name: { description: 'User name' },
  },
  examples: ['create-user --name John'],
  events: ['UserCreated'],
  handle: async (cmd) => ({
    type: 'UserCreated',
    data: { userId: 'u-001', name: cmd.data.name },
  }),
});

bus.registerCommandHandler(handler);

// 3. Subscribe to the resulting event
bus.subscribeToEvent('UserCreated', {
  name: 'UserCreatedLogger',
  handle: (event) => {
    console.log('User created:', event.data);
  },
});

// 4. Send a command
await bus.sendCommand({
  type: 'CreateUser',
  data: { name: 'John' },
  requestId: 'req-001',
});

How-to Guides

Register a Command Handler

Use defineCommandHandler to create a handler with CLI metadata, then register it on the bus:

import { createMessageBus, defineCommandHandler } from '@auto-engineer/message-bus';

const handler = defineCommandHandler({
  name: 'MyCommand',
  alias: 'my-command',
  description: 'Does something',
  fields: {},
  examples: [],
  events: ['MyEvent'],
  handle: async (cmd) => ({ type: 'MyEvent', data: {} }),
});

const bus = createMessageBus();
bus.registerCommandHandler(handler);

Send a Command

sendCommand finds the registered handler, executes it, and publishes the resulting events:

await bus.sendCommand({
  type: 'CreateUser',
  data: { name: 'John', email: '[email protected]' },
  requestId: 'req-123',
  correlationId: 'signup-flow-1',
});

Subscribe to Events

const subscription = bus.subscribeToEvent('UserCreated', {
  name: 'UserCreatedNotifier',
  handle: async (event) => {
    console.log(`User created: ${event.data.userId}`);
  },
});

// Later, stop receiving events
subscription.unsubscribe();

Subscribe to All Events

const subscription = bus.subscribeAll({
  name: 'EventLogger',
  handle: (event) => {
    console.log(`[${event.type}]`, event.data);
  },
});

Return Multiple Events from a Handler

A command handler can return a single event or an array:

const handler = defineCommandHandler({
  name: 'BatchCreate',
  alias: 'batch-create',
  description: 'Creates multiple items',
  fields: {},
  examples: [],
  events: ['ItemCreated'],
  handle: async (cmd) => {
    return cmd.data.items.map((item: { id: string }) => ({
      type: 'ItemCreated' as const,
      data: item,
    }));
  },
});

Track Events by Correlation ID

Listen for events tied to a specific correlation ID:

const subscription = bus.onCorrelation('signup-flow-1', (event) => {
  console.log('Correlated event:', event.type);
});

subscription.unsubscribe();

Track Events by Correlation Prefix

Listen for all events whose correlation ID starts with a given prefix. Useful for job graphs where each job appends a suffix:

const subscription = bus.onCorrelationPrefix('graph:g1:', (event) => {
  console.log('Graph event:', event.type, event.correlationId);
});

subscription.unsubscribe();

API Reference

Package Exports

import {
  createMessageBus,
  defineCommandHandler,
} from '@auto-engineer/message-bus';

import type {
  Command,
  Event,
  CommandHandler,
  EventHandler,
  EventSubscription,
  EventDefinition,
  MessageBus,
  UnifiedCommandHandler,
  FieldDefinition,
  PackageMetadata,
} from '@auto-engineer/message-bus';

Functions

createMessageBus()

function createMessageBus(): MessageBus

Create a new message bus instance with isolated state.

defineCommandHandler(config)

function defineCommandHandler<C extends Command>(config: {
  name: string;
  alias: string;
  description: string;
  displayName?: string;
  category?: string;
  icon?: string;
  package?: PackageMetadata;
  fields: Record<string, FieldDefinition<unknown>>;
  examples: string[];
  events: EventDefinition[];
  handle: (command: C) => Promise<Event | Event[]>;
}): UnifiedCommandHandler<C>

Define a command handler with metadata for CLI integration.

| Parameter | Type | Description | |---|---|---| | name | string | Command type string (e.g. 'CreateUser') | | alias | string | CLI alias (e.g. 'create-user') | | description | string | Human-readable description | | displayName | string? | Optional display name | | category | string? | Optional grouping category | | icon | string? | Optional icon | | package | PackageMetadata? | Source package metadata | | fields | Record<string, FieldDefinition> | Parameter definitions | | examples | string[] | Usage examples | | events | EventDefinition[] | Events this handler may emit | | handle | (command) => Promise<Event \| Event[]> | The handler function |

Interfaces

Command<Type, Data>

type Command<
  Type extends string = string,
  Data extends Record<string, unknown> = Record<string, unknown>
> = Readonly<{
  type: Type;
  data: Readonly<Data>;
  timestamp?: Date;
  requestId?: string;
  correlationId?: string;
}>;

Event<Type, Data>

type Event<
  Type extends string = string,
  Data extends Record<string, unknown> = Record<string, unknown>
> = Readonly<{
  type: Type;
  data: Data;
  timestamp?: Date;
  requestId?: string;
  correlationId?: string;
}>;

CommandHandler

type CommandHandler<TCommand extends Command = Command, TEvent extends Event = Event> = {
  name: string;
  handle: (command: TCommand) => Promise<TEvent | TEvent[]>;
};

EventHandler

type EventHandler<TEvent extends Event = Event> = {
  name: string;
  handle: (event: TEvent) => Promise<void> | void;
};

EventSubscription

type EventSubscription = {
  unsubscribe: () => void;
};

MessageBus Methods

| Method | Signature | Description | |---|---|---| | registerCommandHandler | (handler: CommandHandler) => void | Register a handler for a command type. Throws if one is already registered. | | sendCommand | (command: Command) => Promise<void> | Dispatch a command to its handler. Resulting events are published automatically. | | publishEvent | (event: Event) => Promise<void> | Publish an event to all matching subscribers. | | subscribeToEvent | (eventType: string, handler: EventHandler) => EventSubscription | Subscribe to a specific event type. | | subscribeAll | (handler: EventHandler) => EventSubscription | Subscribe to all events regardless of type. | | registerEventHandler | (handler: EventHandler) => EventSubscription | Register an event handler, inferring event type from the handler name (strips Handler suffix). | | getCommandHandlers | () => Record<string, CommandHandler> | Return a shallow copy of all registered command handlers. | | onCorrelation | (correlationId: string, listener: (event: Event) => void) => EventSubscription | Listen for events with an exact correlation ID match. | | onCorrelationPrefix | (prefix: string, listener: (event: Event) => void) => EventSubscription | Listen for events whose correlation ID starts with the given prefix. |

Architecture

src/
├── index.ts              # Re-exports all public API
├── message-bus.ts         # createMessageBus factory and MessageBus type
├── define-command.ts      # defineCommandHandler and related types
├── types.ts               # Core Command, Event, Handler types
└── message-bus.specs.ts   # Tests for correlation features

flowchart LR
    A[sendCommand] --> B[CommandHandler.handle]
    B --> C[Event / Event array]
    C --> D[publishEvent]
    D --> E[Event subscribers]
    D --> F[Correlation listeners]

Dependencies

| Type | Packages | |---|---| | Runtime | debug (via createDebug) | | Dev | typescript, tsx, @types/node |

This package has no dependencies on other @auto-engineer/ packages. It is a foundational package used throughout the monorepo.