node-cqrs: CQRS and Event Sourcing for TypeScript

Features

• Plain messages: Commands and events are ordinary typed objects without decorators or generated classes.
• Focused domain blocks: Aggregates handle commands, projections build read models, and sagas coordinate work.
• Replaceable infrastructure: Thin interfaces let applications provide their own storage, buses, locks, views, and dispatch processors.
• Concurrency handling: Commands are serialized per aggregate within a process; persistent event stores add optimistic concurrency for distributed writers.
• Projection lifecycle: Restore hooks, readiness locks, event deduplication, and checkpoints are available to persistent views.
• Selective rehydration: Aggregates can restore selected events and use optional snapshots.
• Dispatch pipelines: Event batches pass through configurable persistence and processing pipelines with concurrency limits.

Infrastructure modules can be combined according to the deployment:

• node-cqrs/sqlite - embedded event storage and relational or JSON views;
• node-cqrs/mongodb - distributed event storage and document views;
• node-cqrs/redis - distributed document projection views;
• node-cqrs/postgresql - transactional event storage and relational or JSON views;
• node-cqrs/rabbitmq - distributed command and event buses;
• node-cqrs/workers - worker-thread projections for CPU-intensive handlers.

Installation

npm install node-cqrs

The built package supports Node.js 16 and later. The TypeScript examples can be executed directly with Node.js 24 or later; earlier Node.js versions require normal TypeScript compilation or a loader.

The browser bundle exposes the browser-compatible core API. Database adapters, RabbitMQ, and Node.js worker threads are server-side modules. Infrastructure modules require their documented peer dependencies.

Quick Start

This example defines one command, one event, and one read model entirely in memory:

import { AbstractAggregate, AbstractProjection, ContainerBuilder, InMemoryEventStorage } from 'node-cqrs';
import type { IContainer, IEvent, Identifier } from 'node-cqrs';

type UserRecord = {
	username: string;
};

type UserCreatedEvent = IEvent<UserRecord>;
type UsersView = Map<Identifier, UserRecord>;

class UserAggregate extends AbstractAggregate {
	createUser(payload: UserRecord) {
		this.emit('userCreated', payload);
	}
}

class UsersProjection extends AbstractProjection<UsersView> {
	constructor() {
		super({ view: new Map() });
	}

	userCreated(event: UserCreatedEvent) {
		this.view.set(event.aggregateId!, event.payload);
	}
}

interface AppContainer extends IContainer {
	usersView: UsersView;
}

const builder = new ContainerBuilder<AppContainer>();
builder.register(InMemoryEventStorage);
builder.registerAggregate(UserAggregate);
builder.registerProjection(UsersProjection, 'usersView');

const container = builder.container();
const { usersView, commandBus } = container;

const [userCreated] = await commandBus.send('createUser', undefined, {
	payload: { username: 'alice' }
});

console.log(usersView.get(userCreated.aggregateId!)); // { username: 'alice' }

InMemoryEventStorage is useful for learning and tests; its events disappear when the process exits. Choose a persistent event store from Infrastructure for an application that must survive restarts.

How It Fits Together

Domain behavior is split into three small blocks:

• Aggregates restore write-side state, validate commands, and emit events.
• Projections consume events and update read-side views.
• Sagas react to events and enqueue commands for multi-step processes.

The default runtime flow is:

1. The command bus delivers a command to an aggregate command handler.
2. The handler restores the target aggregate and invokes its command method.
3. Emitted events pass through the event dispatch pipeline, including configured persistence.
4. The event bus delivers committed events to projections, sagas, and other subscribers.

Messages And Replacement Points

Commands and events are plain objects. A message needs only a type and payload; identifiers, context, aggregate versions, and saga origins are added when the workflow needs them:

type Message<TPayload> = {
	type: string;
	aggregateId?: Identifier;
	payload: TPayload;
	context?: unknown;
};

const command: Message<{ username: string }> = {
	type: 'createUser',
	payload: { username: 'alice' }
};

Library blocks are similarly narrow. For example, a projection only needs a view and three lifecycle methods:

interface Projection<TView> {
	readonly view: TView;
	subscribe(eventStore: IObservable): void | Promise<void>;
	restore(eventStore: IEventStorageReader): void | Promise<void>;
	project(event: IEvent): void | Promise<void>;
}

Applications can implement these contracts directly or extend the supplied base classes:

The framework-free example implements the core interfaces without using the supplied aggregate or projection base classes.

Aggregates

AbstractAggregate maps public method names to command types. This aggregate handles a createUser command and emits a userCreated event:

class UserAggregate extends AbstractAggregate {
	createUser(payload: { username: string }) {
		this.emit('userCreated', { username: payload.username });
	}
}

Override static handles when command types should be declared explicitly.

Aggregate State

State is rebuilt by applying the aggregate’s historical events. Keep mutation deterministic and validate in the command method before emitting a new event:

class UserState {
	username!: string;

	userCreated(event: IEvent<{ username: string }>) {
		this.username = event.payload.username;
	}

	userRenamed(event: IEvent<{ username: string }>) {
		this.username = event.payload.username;
	}
}

class UserAggregate extends AbstractAggregate<UserState> {
	protected readonly state = new UserState();

	renameUser(payload: { username: string }) {
		if (payload.username === this.state.username)
			throw new Error('Username is unchanged');

		this.emit('userRenamed', payload);
	}
}

Constructor dependencies are resolved from the container, so domain behavior can use application services without service locators:

type UserAggregateOptions = IAggregateConstructorParams<void> & {
	authService?: AuthService;
};

class UserAggregate extends AbstractAggregate {
	readonly #authService: AuthService;

	constructor({ authService, ...options }: UserAggregateOptions) {
		super(options);
		if (!authService)
			throw new TypeError('authService is required');

		this.#authService = authService;
	}
}

interface AggregateContainer extends IContainer {
	authService: AuthService;
}

const builder = new ContainerBuilder<AggregateContainer>();
builder.register(AuthService).as('authService');
builder.registerAggregate(UserAggregate);

Projections And Views

AbstractProjection maps event types to methods in the same way:

class UsersProjection extends AbstractProjection<Map<Identifier, UserRecord>> {
	constructor() {
		super({ view: new Map() });
	}

	userCreated(event: IEvent<UserRecord>) {
		this.view.set(event.aggregateId!, event.payload);
	}
}

Override static handles to declare event types explicitly.

Expose a projection view through the typed container and wait for startup restoration before serving reads:

interface AppContainer extends IContainer {
	usersView: Map<Identifier, UserRecord>;
}

const builder = new ContainerBuilder<AppContainer>();
builder.registerProjection(UsersProjection, 'usersView');

const container = builder.container();
await Promise.all(container.restorePromises ?? []);

const usersView = container.usersView;

Persistent projection implementations provide restore locking, event deduplication, and checkpoints. Their exact transaction and retry guarantees are documented by each infrastructure module.

Sagas

Sagas coordinate multi-step work by handling events and producing follow-up commands:

class WelcomeEmailSaga extends AbstractSaga {
	userSignedUp(event: IEvent<{ email: string }>) {
		this.enqueue('sendWelcomeEmail', undefined, {
			email: event.payload.email
		});
	}
}

builder.register(EventIdAugmentor).as('eventIdAugmenter');
builder.registerSaga(WelcomeEmailSaga);

Saga starter events require ids. Register EventIdAugmentor when the selected event storage does not assign an id before saga delivery.

By default, a saga starts when a handled event has no origin for that saga type. Use static startsWith for explicit starter event types, static handles for additional events, and static sagaDescriptor for a stable origin key independent of the class name.

The simple saga and overlapping sagas demonstrate state restoration and origin propagation.

Runtime Lifecycle And Guarantees

Container dependencies are resolved lazily. Access each exposed projection view during startup to create its projection, subscribe it to the event store, and start restoration. Then await restorePromises before accepting requests that depend on those views. Destructuring exposed views from the container, as in the quick start, performs that initial resolution.

Event dispatch has two stages:

1. commandBus.send() waits for command handling and the configured dispatch pipeline, including event storage.
2. Event-bus publication runs asynchronously after pipeline processing so command throughput is not tied to every subscriber.

Use await eventStore.drain() when a caller, test, or shutdown path must wait for all currently queued event publications. A completed command does not otherwise guarantee that every projection has finished processing its events. Configure eventPublishErrorHandler when publication failures must be logged or reported; draining waits for publication attempts but does not make subscriber handling part of the storage transaction.

Infrastructure determines distributed guarantees:

• In-memory locks and buses coordinate only one process.
• Persistent event stores define transaction boundaries and optimistic concurrency behavior.
• Persistent views define restore locking, event deduplication, retries, and checkpoint semantics.
• RabbitMQ can redeliver acknowledged-late messages, so distributed handlers should be idempotent.

Review the selected module documentation before relying on a specific failure or multi-instance behavior.

Infrastructure

Choose infrastructure by deployment need. Modules can be used independently or combined.

MongoDB, Redis, and PostgreSQL support is currently experimental and has not yet been validated in production. Their APIs may change in minor versions.

Event Storage

See the SQLite example, MongoDB event-storage example, and PostgreSQL example.

Advanced: Manual Composition

The container is optional. The same components can be assembled directly:

const commandBus = new InMemoryMessageBus();
const eventBus = new InMemoryMessageBus();
const eventStorage = new InMemoryEventStorage();
const eventStore = new EventStore({
	eventStorageReader: eventStorage,
	identifierProvider: eventStorage,
	eventDispatchPipeline: [eventStorage],
	eventBus
});

const aggregateHandler = new AggregateCommandHandler({
	aggregateType: UserAggregate,
	eventStore
});
aggregateHandler.subscribe(commandBus);

const projection = new UsersProjection();
projection.subscribe(eventStore);
await projection.restore(eventStore);

const [userCreated] = await commandBus.send('createUser', undefined, {
	payload: { username: 'alice' }
});
await eventStore.drain();

console.log(projection.view.get(userCreated.aggregateId!));

OpenTelemetry

Register a tracer factory to enable spans across commands, event dispatch, projections, sagas, storage adapters, and RabbitMQ transport. Install @opentelemetry/api alongside the library:

import { trace } from '@opentelemetry/api';

builder.register(() => (name: string) => trace.getTracer(`cqrs.${name}`)).as('tracerFactory');

See the telemetry example for a complete setup with exporters.

The project was inspired by Lokad.CQRS.