Packages

@xtaskjs/cqrs

Command, query, and event buses with handler decorators, read or write datasource aliases, and lifecycle-managed projection workflows.

Back to catalog View source

npm install @xtaskjs/cqrs @xtaskjs/typeorm reflect-metadata typeorm Package path: packages/cqrs

Overview

What this package owns in the runtime

CQRS adds message buses, handler discovery, read or write repository injection, process managers, projection rebuilders, and idempotent command execution to xtaskjs. It builds on @xtaskjs/typeorm so the container can expose read and write datasources and repositories without hard-coding infrastructure details into handlers.

What it provides

CommandBus, QueryBus, and EventBus are registered in the xtaskjs container and can be injected into controllers and services.
CommandHandler(), QueryHandler(), EventHandler(), ProcessManager(), Saga, and ProjectionRebuilder() declare the CQRS runtime with decorators.
InjectReadRepository(), InjectWriteRepository(), InjectReadDataSource(), and InjectWriteDataSource() resolve aliases backed by @xtaskjs/typeorm.
IdempotentCommand() adds built-in in-memory command deduplication with an overridable idempotency store.

How it fits

Configured through configureCqrs() or @Cqrs(...) before CreateApplication() so read and write datasource names are known during bootstrap.
Initializes after TypeORM registration and publishes buses, lifecycle services, datasource aliases, and repository aliases into the same DI container.
Demonstrated by the 19-cqrs_app and 20-cqrs_postgres_replication_app samples, and used by this website to route admin, registration, and news flows through command and query buses.

Usage Chart

How strongly this package shapes the runtime

Bootstrap

4/5

Dependency Injection

5/5

Persistence

5/5

Messaging

5/5

Operations

5/5

Package Flow

How this package moves through xtaskjs runtime phases

Before startup

Register read and write datasources with @xtaskjs/typeorm, then call configureCqrs() or decorate a configuration class with @Cqrs(...) so datasource aliases and idempotency settings are ready before handler discovery.

During CreateApplication()

The CQRS lifecycle registers command, query, and event buses; discovers decorated handlers from the container; and binds read or write datasources and repositories into the same DI runtime.

During app.close()

CQRS clears handler registries, process managers, projection rebuilders, and in-memory idempotency state while TypeORM remains responsible for tearing down the underlying datasources.

API Surface

Representative exports from the upstream package

Configuration and buses

configureCqrs
Cqrs
CommandBus
QueryBus
EventBus

Handlers, injectors, and idempotency

CommandHandler
QueryHandler
EventHandler
ProcessManager
Saga
ProjectionRebuilder
IdempotentCommand
InjectCommandBus
InjectQueryBus
InjectEventBus
InjectIdempotencyStore
InjectCqrsLifecycleManager
InjectReadDataSource
InjectWriteDataSource
InjectReadRepository
InjectWriteRepository

Lifecycle, tokens, and types

CqrsLifecycleManager
initializeCqrsIntegration
shutdownCqrsIntegration
resetCqrsIntegration
getCqrsLifecycleManager
getCommandBusToken
getQueryBusToken
getEventBusToken
getIdempotencyStoreToken
getReadDataSourceToken
getWriteDataSourceToken
getReadRepositoryToken
getWriteRepositoryToken
ICommandHandler
IQueryHandler
IEventHandler
IProcessManager
ISaga
IProjectionRebuilder
ProcessManagerContext
ProjectionRebuildContext
IIdempotencyStore
CqrsOptions

Usage

Typical adoption flow

1. Configure read and write aliases

Start by naming the datasources you want CQRS to treat as read and write sides, even if both names point to the same TypeORM datasource in a smaller application.

2. Decorate handlers and projections

Use CommandHandler, QueryHandler, EventHandler, ProcessManager, ProjectionRebuilder, and IdempotentCommand to describe each message flow instead of manually wiring buses and registries.

3. Inject buses or repositories at the edges

Inject CommandBus and QueryBus into controllers, and inject read or write repositories into handlers so HTTP delivery, orchestration, and persistence boundaries stay explicit.

Example

Reference snippet

Separate read and write models with injected buses

import { Service } from "@xtaskjs/core";
import {
  CommandBus,
  CommandHandler,
  Cqrs,
  EventBus,
  IdempotentCommand,
  InjectCommandBus,
  InjectEventBus,
  InjectReadRepository,
  InjectWriteRepository,
  QueryBus,
  QueryHandler,
} from "@xtaskjs/cqrs";
import { DataSource, Repository, TypeOrmDataSource } from "@xtaskjs/typeorm";

@TypeOrmDataSource({ name: "write-db", type: "sqlite", database: "write.sqlite", entities: [UserEntity], synchronize: true })
class WriteDatabase {}

@TypeOrmDataSource({ name: "read-db", type: "sqlite", database: "read.sqlite", entities: [UserProjection], synchronize: true })
class ReadDatabase {}

@Cqrs({ writeDataSourceName: "write-db", readDataSourceName: "read-db" })
class CqrsConfiguration {}

class CreateUserCommand {
  constructor(public readonly name: string, public readonly idempotencyKey: string) {}
}

class ListUsersQuery {}

class UserCreatedEvent {
  constructor(public readonly id: number, public readonly name: string) {}
}

@Service()
@IdempotentCommand<CreateUserCommand>({ key: (command) => command.idempotencyKey })
@CommandHandler(CreateUserCommand)
class CreateUserHandler {
  constructor(
    @InjectWriteRepository(UserEntity)
    private readonly writeUsers: Repository<UserEntity>,
    @InjectEventBus()
    private readonly events: EventBus
  ) {}

  async execute(command: CreateUserCommand) {
    const created = await this.writeUsers.save(this.writeUsers.create({ name: command.name }));
    await this.events.publish(new UserCreatedEvent(created.id, created.name));
    return created.id;
  }
}

@Service()
@QueryHandler(ListUsersQuery)
class ListUsersHandler {
  constructor(
    @InjectReadRepository(UserProjection)
    private readonly readUsers: Repository<UserProjection>
  ) {}

  execute() {
    return this.readUsers.find({ order: { id: "ASC" } });
  }
}

@Service()
class UsersFacade {
  constructor(
    @InjectCommandBus() private readonly commandBus: CommandBus,
    @InjectQueryBus() private readonly queryBus: QueryBus
  ) {}
}

Samples

Official samples to inspect next

Reference samples: 19-cqrs_app and 20-cqrs_postgres_replication_app