@nestjs/cqrs

To use the CQRS module in your NestJS application, install the @nestjs/cqrs package using npm.

$ npm install --save @nestjs/cqrs

The CommandBus uses an ICommandPublisher to publish commands. By default, it uses DefaultCommandPubSub, which is an in-memory implementation.

You can customize the publisher in two ways:

1. Via Module Options: Provide a commandPublisher in the CqrsModuleOptions during module initialization.
2. Directly on the Instance: Use the publisher setter to replace the current publisher on an existing CommandBus instance.

// Setting a custom publisher directly
commandBus.publisher = myCustomPublisher;

The @nestjs/cqrs package exports several specialized exceptions used to identify errors within the Command Query Responsibility Segregation pattern. These exceptions are typically thrown by the internal CQRS bus when handlers or sagas are misconfigured or missing.

Available exceptions include:

• CommandNotFoundException: Thrown when a command is dispatched but no handler is registered for it.
• QueryNotFoundException: Thrown when a query is dispatched but no handler is registered for it.
• InvalidCommandHandlerException: Thrown when a command handler is invalid.
• InvalidQueryHandlerException: Thrown when a query handler is invalid.
• InvalidEventsHandlerException: Thrown when an event handler is invalid.
• InvalidSagaException: Thrown when a saga is invalid.
• UnsupportedSagaScopeException: Thrown when a saga is used with an unsupported scope.

The @nestjs/cqrs package provides core abstractions for Command and Query patterns. Use the main entry point to access Command and Query base classes, which are used to define the intent of operations within your application. Commands represent actions that change state, while Queries represent requests for data.

import { Command, Query } from '@nestjs/cqrs';

To configure the CqrsModule using asynchronous providers (e.g., fetching configuration from a database or external service), use the CqrsModuleAsyncOptions interface. This follows the standard NestJS dynamic module pattern.

export interface CqrsModuleAsyncOptions {
  imports?: any[];
  useExisting?: Type<CqrsModuleOptionsFactory>;
  useClass?: Type<CqrsModuleOptionsFactory>;
  useFactory?: (...args: any[]) => Promise<CqrsModuleOptions> | CqrsModuleOptions;
  useValue?: CqrsModuleOptions;
  inject?: any[];
  extraProviders?: Provider[];
}

The CommandBus is the primary entry point for dispatching commands in the CQRS module. You use the execute method to trigger a command, which will then be routed to its registered handler. The execute method returns a Promise that resolves with the result returned by the command handler.

If no handler is registered for the provided command, it throws a CommandHandlerNotFoundException.

// Assuming 'command' is an instance of a class implementing ICommand
const result = await commandBus.execute(command);

// You can also pass an optional AsyncContext
const result = await commandBus.execute(command, context);

The QueryBus uses an IQueryPublisher to publish queries. By default, it uses DefaultQueryPubSub (an in-memory implementation). You can customize the publisher by either:

1. Providing a queryPublisher in the CqrsModuleOptions during module configuration.
2. Using the publisher setter on the QueryBus instance directly.

Access the current publisher via the publisher getter.

// Setting a custom publisher directly
this.queryBus.publisher = myCustomPublisher;

The ofType operator allows you to filter events on the EventBus so that a handler only reacts to events of a specific type. This is useful when you want to subscribe to a subset of all events flowing through the bus.

To use it, you pass the class constructor of the event type you want to listen for to the ofType function. When used in conjunction with an event listener, it ensures the callback is only executed for events that are instances of that specific class.

// Example usage pattern (conceptual based on export)
// eventBus.subscribe(ofType(MySpecificEvent))
//   .subscribe(event => {
//     // This only runs for MySpecificEvent
//   });

Sagas are long-running processes that listen to events and trigger commands. You can register multiple sagas at once using registerSagas, which looks for methods decorated with @Saga (using SAGA_METADATA). Note that sagas must be part of a static dependency tree to be registered this way; otherwise, an UnsupportedSagaScopeException is thrown.

// Registering multiple sagas from an array of wrappers
this.eventBus.registerSagas(sagaWrappers);

// Registering a single saga function
this.eventBus.registerSaga(mySagaFunction);

The IEventBus interface defines the contract for publishing events within the CQRS module. It supports publishing single events or multiple events simultaneously. You can optionally provide an AsyncContext or a custom dispatcherContext to manage execution context during event dispatching.

Key methods:

• publish(event: TEvent, ...): Dispatches a single event.
• publishAll(events: TEvent[], ...): Dispatches an array of events.

Both methods support overloads to include AsyncContext and TContext (dispatcher context) for advanced context propagation.

// Example of publishing a single event
eventBus.publish(new MyEvent());

// Example of publishing multiple events
eventBus.publishAll([new MyEvent(), new AnotherEvent()]);

// Example of publishing with context
eventBus.publish(new MyEvent(), dispatcherContext, asyncContext);