@setu-ts/cqrs-plugin

Command Query Responsibility Segregation. Registers an ICqrsFacade under CAPABILITIES.CQRS ('cqrs'), exposing a commandBus and a queryBus.

Both buses share one request pipeline, so a behavior registered once applies to commands and queries alike.

Installation

import { CqrsPlugin } from '@setu-ts/cqrs-plugin';

Usage

import { CqrsPlugin } from '@setu-ts/cqrs-plugin';
import {
  CAPABILITIES,
  type CqrsCommand,
  type CqrsQuery,
  type ICqrsFacade,
  type IPipelineBehavior,
} from '@setu-ts/common';

const timing: IPipelineBehavior = {
  async handle(request, next) {
    const started = runtime.hrtime();
    const result = await next();
    logger.debug('handled', { type: request.type, ms: runtime.hrtime() - started });
    return result;
  },
};

app.register(CqrsPlugin({ behaviors: [timing] }));

const cqrs = app.services.get<ICqrsFacade>(CAPABILITIES.CQRS);

cqrs.commandBus.register('CreateUser', {
  handle: (command: CqrsCommand<{ email: string }>) => ({ id: '123', name: 'Ada' }),
});
cqrs.queryBus.register('GetUser', {
  handle: (query: CqrsQuery<{ id: string }>) => ({ id: query.data.id, name: 'Ada' }),
});

await cqrs.commandBus.execute({ type: 'CreateUser', data: { email: 'ada@example.com' } });
const user = await cqrs.queryBus.execute<User>({ type: 'GetUser', data: { id: '123' } });

Options

A behavior that returns without calling next() short-circuits the pipeline — the handler and all later behaviors do not run.

Errors

HandlerNotFoundError is exported for instanceof handling; it is thrown when a request reaches a bus with no registered handler.

Exports

Generated from the package barrel by pnpm run docs:exports; pnpm run check:docs fails when it drifts.

Full API

Every export and option is documented in PUBLIC_API.md.

API reference

CQRS (@setu-ts/cqrs-plugin)

Provides command/query separation with buses.

Registration

import { CqrsPlugin } from '@setu-ts/cqrs-plugin';
import type { CqrsRequest, IPipelineBehavior } from '@setu-ts/common';

// Example behavior implementations
const loggingBehavior: IPipelineBehavior = {
  handle: async (request: CqrsRequest, next: () => Promise<unknown>) => {
    console.log(`Executing ${request.type}`);
    const result = await next();
    console.log(`Completed ${request.type}`);
    return result;
  },
};

const timingBehavior: IPipelineBehavior = {
  handle: async (request: CqrsRequest, next: () => Promise<unknown>) => {
    const start = Date.now();
    const result = await next();
    console.log(`${request.type} took ${Date.now() - start}ms`);
    return result;
  },
};

app.register(
  CqrsPlugin({
    behaviors: [loggingBehavior, timingBehavior],
  }),
);

Defining Commands and Queries

import type { CqrsCommand, CqrsQuery } from '@setu-ts/cqrs-plugin';

// A request is routed by its string `type`; `data` carries the payload.
// Both a class instance (below) and a plain `{ type, data }` object satisfy the contract.
class CreateUserCommand implements CqrsCommand<{ name: string; email: string }> {
  readonly type = 'CreateUserCommand';
  constructor(public readonly data: { name: string; email: string }) {}
}

class GetUserQuery implements CqrsQuery<{ id: string }> {
  readonly type = 'GetUserQuery';
  constructor(public readonly data: { id: string }) {}
}

Implementing Handlers

import type { ICommandHandler, IQueryHandler } from '@setu-ts/cqrs-plugin';

class CreateUserHandler implements ICommandHandler<CreateUserCommand, string> {
  constructor(private db: IDatabaseService) {}

  async handle(command: CreateUserCommand): Promise<string> {
    const user = await this.db.getRepository<User>('User').create(command.data);
    return user.id;
  }
}

class GetUserHandler implements IQueryHandler<GetUserQuery, User> {
  constructor(private db: IDatabaseService) {}

  async handle(query: GetUserQuery): Promise<User> {
    const user = await this.db.getRepository<User>('User').findById(query.data.id);
    if (!user) throw new Error('User not found');
    return user;
  }
}

Registering Handlers

Two ways, and both end at the same commandBus.register(type, handler) call.

Declaratively, through the plugin’s options — which is what a handler needing no constructor dependencies wants, and what setu generate command-handler wires for you:

import type { CommandHandlerRegistration, QueryHandlerRegistration } from '@setu-ts/cqrs-plugin';

const commandHandlers: readonly CommandHandlerRegistration[] = [
  { type: 'CreateUserCommand', handler: new CreateUserHandler() },
  // Or a factory that builds the handler from the service registry:
  { type: 'PlaceOrderCommand', handler: createPlaceOrderCommandHandler },
];
const queryHandlers: readonly QueryHandlerRegistration[] = [
  { type: 'GetUserQuery', handler: new GetUserHandler() },
];

app.register(CqrsPlugin({ commandHandlers, queryHandlers }));

Each entry is a { type, handler } pair because that is exactly what the bus takes — the type is a string the request carries, not something derivable from the handler’s class. The handler accepts an instance or a RegistryFactory ((services: IServiceRegistry) => handler), and behaviors accepts the same instance | RegistryFactory<IPipelineBehavior> union. Factories are called at the onInit phase — the first phase at which the registry holds every capability — so a factory can resolve any capability (the event bus, a database, the logger) and build the handler or behavior with it. A factory that throws rejects start(), naming the option and the entry, with the original error preserved as cause. Instance entries keep their register() timing byte-identically.

Imperatively, from a plugin — the route to take when a handler needs a capability, since the buses only exist once CqrsPlugin has registered and IApplication exposes no lifecycle hook:

app.register({
  name: 'user-handlers',
  version: '1.0.0',
  dependencies: ['cqrs', 'database'],
  register(ctx) {
    const commandBus = ctx.services.get<ICommandBus>('command-bus');
    const queryBus = ctx.services.get<IQueryBus>('query-bus');
    const db = ctx.services.get<IDatabaseService>('database');

    commandBus.register('CreateUserCommand', new CreateUserHandler(db));
    queryBus.register('GetUserQuery', new GetUserHandler(db));
  },
});

Using in Routes

app.router.post('/users', async (ctx) => {
  const commandBus = ctx.services.get<ICommandBus>('command-bus');
  const userId = await commandBus.execute<string>(new CreateUserCommand(ctx.request.body));
  return ctx.response.status(201).json({ id: userId });
});

app.router.get('/users/:id', async (ctx) => {
  const queryBus = ctx.services.get<IQueryBus>('query-bus');
  const user = await queryBus.execute<User>(new GetUserQuery({ id: ctx.params.id }));
  return ctx.response.json(user);
});