# Set Up the CQRS Bridge (/integrations/nestjs/how-to/cqrs-bridge)



Wire up `@kubemq/nestjs-transport/cqrs` to distribute `@nestjs/cqrs` commands, queries, and events across services. For how the routing works internally, see [CQRS Bridge Concepts](../concepts/cqrs-bridge); for every configuration field, see the [KubeMQCqrsOptions reference](../reference/cqrs-options).

## Prerequisites [#prerequisites]

* `@kubemq/nestjs-transport` and `kubemq-js` installed, with `KubeMQModule.forRoot()` already registered in the app (see [Getting Started with NestJS](/integrations/nestjs/tutorials/getting-started))
* The `@nestjs/cqrs` peer dependency installed (see [Setup](#setup) below)
* A running KubeMQ broker reachable on `localhost:50000`

## Setup [#setup]

Install the optional peer dependency alongside the transport:

```bash
npm install @nestjs/cqrs
```

Import order matters. `KubeMQCqrsModule.forRoot()` swaps the bus publishers during its `onModuleInit`, so both `CqrsModule` (which provides the buses) and `KubeMQModule.forRoot()` (which provides the broker connection) must be initialized first. Import them **before** `KubeMQCqrsModule.forRoot()`:

```typescript title="app.module.ts"
import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';
import { KubeMQModule } from '@kubemq/nestjs-transport';
import { KubeMQCqrsModule } from '@kubemq/nestjs-transport/cqrs';

const address = process.env.KUBEMQ_ADDRESS ?? 'localhost:50000';

@Module({
  imports: [
    KubeMQModule.forRoot({
      address,
      clientId: 'orders-app',
      isGlobal: true,
    }),
    CqrsModule,
    KubeMQCqrsModule.forRoot({
      commandChannelPrefix: 'myapp.commands',
      queryChannelPrefix: 'myapp.queries',
      eventChannelPrefix: 'myapp.events',
      commandTimeout: 10,
      queryTimeout: 10,
    }),
  ],
})
export class AppModule {}
```

`KubeMQCqrsModule` reads the broker address, credentials, TLS, and reconnection settings from the `KubeMQModule.forRoot()` registration — it does not take its own `address`. If `KubeMQModule.forRoot()` (or `forRootAsync()`) is missing, the module throws on startup because no connection configuration is available.

<Callout type="warn">
  If CQRS commands, queries, or events are not being distributed, the most common cause is import order. `KubeMQCqrsModule.forRoot()` must be imported **after** both `CqrsModule` and `KubeMQModule.forRoot()`, otherwise the buses are not yet available to have their publishers swapped.
</Callout>

You need a running KubeMQ broker. The transport (and therefore the bridge) connects over native gRPC on port `50000` — no HTTP connector is required:

<RunKubeMQ ports="[50000]" />

For async configuration (for example, pulling prefixes from `ConfigService`), use `KubeMQCqrsModule.forRootAsync({ imports, useFactory, inject })`, which returns the same providers via a factory. See the [KubeMQCqrsOptions reference](../reference/cqrs-options) for the full options table.

## Commands [#commands]

A command is a plain class. Define it, annotate a handler with the `@nestjs/cqrs` `@CommandHandler` decorator, and dispatch it with `CommandBus.execute()`. The bridge serializes the command instance and sends it over the KubeMQ command channel; the handler's return value is sent back as the command response.

```typescript title="create-order.command.ts"
export class CreateOrderCommand {
  constructor(
    public readonly productId: string,
    public readonly quantity: number,
  ) {}
}
```

```typescript title="create-order.handler.ts"
import { CommandHandler, ICommandHandler, EventBus } from '@nestjs/cqrs';
import { Logger } from '@nestjs/common';
import { CreateOrderCommand } from './create-order.command';
import { OrderCreatedEvent } from './order-created.event';

@CommandHandler(CreateOrderCommand)
export class CreateOrderHandler implements ICommandHandler<CreateOrderCommand> {
  private readonly logger = new Logger(CreateOrderHandler.name);
  private orderCounter = 0;

  constructor(private readonly eventBus: EventBus) {}

  async execute(command: CreateOrderCommand): Promise<string> {
    this.orderCounter++;
    const orderId = `ORD-${String(this.orderCounter).padStart(3, '0')}`;
    this.logger.log(`Creating order ${orderId}: product=${command.productId}, qty=${command.quantity}`);

    this.eventBus.publish(new OrderCreatedEvent(orderId, command.productId, command.quantity));
    return orderId;
  }
}
```

Dispatch from any service that injects the `CommandBus`:

```typescript title="order.service.ts"
import { Injectable } from '@nestjs/common';
import { CommandBus } from '@nestjs/cqrs';
import { CreateOrderCommand } from './create-order.command';

@Injectable()
export class OrderService {
  constructor(private readonly commandBus: CommandBus) {}

  async createOrder(productId: string, quantity: number): Promise<string> {
    // Routed to: myapp.commands.CreateOrderCommand
    return this.commandBus.execute<CreateOrderCommand, string>(
      new CreateOrderCommand(productId, quantity),
    );
  }
}
```

With the default `commandChannelPrefix` of `cqrs.commands`, this command travels to `cqrs.commands.CreateOrderCommand`. If the handler reports a non-executed response, the bridge throws an error that includes the failing channel name.

## Queries [#queries]

Queries work like commands but use the `QueryBus` and the `@QueryHandler` decorator. The query is routed to a query channel, handled remotely, and the result is returned to the caller.

```typescript title="get-order.query.ts"
export class GetOrderQuery {
  constructor(public readonly orderId: string) {}
}

export class GetOrderResult {
  constructor(
    public readonly orderId: string,
    public readonly productId: string,
    public readonly quantity: number,
    public readonly status: string,
  ) {}
}
```

```typescript title="get-order.handler.ts"
import { QueryHandler, IQueryHandler } from '@nestjs/cqrs';
import { Logger } from '@nestjs/common';
import { GetOrderQuery, GetOrderResult } from './get-order.query';
import { OrderStore } from './order.store';

@QueryHandler(GetOrderQuery)
export class GetOrderHandler implements IQueryHandler<GetOrderQuery, GetOrderResult> {
  private readonly logger = new Logger(GetOrderHandler.name);

  constructor(private readonly store: OrderStore) {}

  async execute(query: GetOrderQuery): Promise<GetOrderResult> {
    this.logger.log(`Querying order: ${query.orderId}`);
    const order = this.store.findById(query.orderId);
    if (!order) {
      throw new Error(`Order ${query.orderId} not found`);
    }
    return new GetOrderResult(order.orderId, order.productId, order.quantity, order.status);
  }
}
```

```typescript title="order.service.ts"
import { Injectable } from '@nestjs/common';
import { QueryBus } from '@nestjs/cqrs';
import { GetOrderQuery, GetOrderResult } from './get-order.query';

@Injectable()
export class OrderQueryService {
  constructor(private readonly queryBus: QueryBus) {}

  async getOrder(orderId: string): Promise<GetOrderResult> {
    // Routed to: myapp.queries.GetOrderQuery
    return this.queryBus.execute<GetOrderQuery, GetOrderResult>(
      new GetOrderQuery(orderId),
    );
  }
}
```

The `QueryBus` resolves with the deserialized response body, so the caller awaits the result exactly as it would with the in-process `@nestjs/cqrs` bus.

## Events [#events]

Events are dispatched through the `EventBus` and are fire-and-forget: `EventBus.publish()` fans the event out over KubeMQ to every subscribed handler. Use the `@nestjs/cqrs` `@EventsHandler` decorator on the receiving side.

```typescript title="order-created.event.ts"
export class OrderCreatedEvent {
  constructor(
    public readonly orderId: string,
    public readonly productId: string,
    public readonly quantity: number,
  ) {}
}
```

```typescript title="order-created.handler.ts"
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { Logger } from '@nestjs/common';
import { OrderCreatedEvent } from './order-created.event';
import { OrderStore } from './order.store';

@EventsHandler(OrderCreatedEvent)
export class OrderCreatedHandler implements IEventHandler<OrderCreatedEvent> {
  private readonly logger = new Logger(OrderCreatedHandler.name);

  constructor(private readonly store: OrderStore) {}

  handle(event: OrderCreatedEvent): void {
    this.logger.log(`Event received — storing order ${event.orderId}`);
    this.store.save({
      orderId: event.orderId,
      productId: event.productId,
      quantity: event.quantity,
      status: 'created',
    });
  }
}
```

```typescript title="order.service.ts"
import { Injectable, Logger } from '@nestjs/common';
import { EventBus } from '@nestjs/cqrs';
import { OrderCreatedEvent } from './order-created.event';

@Injectable()
export class OrderService {
  private readonly logger = new Logger(OrderService.name);

  constructor(private readonly eventBus: EventBus) {}

  async createOrder(orderId: string, product: string, quantity: number): Promise<void> {
    this.logger.log(`Publishing OrderCreatedEvent: id=${orderId}`);
    // Routed to: myapp.events.OrderCreatedEvent
    this.eventBus.publish(new OrderCreatedEvent(orderId, product, quantity));
    this.logger.log('Event published successfully');
  }
}
```

By default events use the transient Events pattern (`sendEvent`) — if no handler is subscribed when the event is published, it is not retained. For durable, replayable domain events, set `persistEvents: true` so the bridge publishes to the KubeMQ Events Store (`sendEventStore`) instead:

```typescript title="app.module.ts"
KubeMQCqrsModule.forRoot({
  eventChannelPrefix: 'myapp.events',
  persistEvents: true,
})
```

<Callout type="info">
  Events Store retains the event stream so late subscribers can replay history. See [Usage](usage#events-store-persistent-events) for the distinction between Events and Events Store, including start positions and sequence tracking.
</Callout>

## Custom Channel Resolution [#custom-channel-resolution]

The channel segment after the prefix defaults to the message's `constructor.name`. So `new CreateOrderCommand(...)` lands on `{commandChannelPrefix}.CreateOrderCommand`. The default resolver falls back to `unknown` for anonymous or plain `Object` instances.

To map message classes onto your own channel naming scheme — for example versioned or namespaced channels — provide a `channelResolver`:

```typescript title="app.module.ts"
KubeMQCqrsModule.forRoot({
  commandChannelPrefix: 'myapp.commands',
  // CreateOrderCommand -> myapp.commands.orders.create.v1
  channelResolver: (message) => {
    const map: Record<string, string> = {
      CreateOrderCommand: 'orders.create.v1',
      GetOrderQuery: 'orders.get.v1',
    };
    return map[message.constructor.name] ?? message.constructor.name;
  },
})
```

The same resolver is applied to commands, queries, and events, so make sure each message class resolves to a segment that is unique within its prefix.

## End-to-End Example [#end-to-end-example]

The repository ships a complete `full-cqrs-flow` example that wires all three buses into a single command → event → query story: a `CreateOrderCommand` is dispatched, its handler publishes an `OrderCreatedEvent` that updates a read model, and a `GetOrderQuery` then reads that model back — every hop crossing KubeMQ.

```typescript title="app.module.ts"
import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';
import { KubeMQModule } from '@kubemq/nestjs-transport';
import { KubeMQCqrsModule } from '@kubemq/nestjs-transport/cqrs';
import { CreateOrderHandler } from './create-order.handler';
import { OrderCreatedHandler } from './order-created.handler';
import { GetOrderHandler } from './get-order.handler';
import { OrderStore } from './order.store';
import { OrderService } from './order.service';

const address = process.env.KUBEMQ_ADDRESS ?? 'localhost:50000';

@Module({
  imports: [
    KubeMQModule.forRoot({
      address,
      clientId: 'nestjs-cqrs-full-flow',
      isGlobal: true,
    }),
    CqrsModule,
    KubeMQCqrsModule.forRoot({
      commandChannelPrefix: 'nestjs-cqrs.full-flow.commands',
      queryChannelPrefix: 'nestjs-cqrs.full-flow.queries',
      eventChannelPrefix: 'nestjs-cqrs.full-flow.events',
      commandTimeout: 10,
      queryTimeout: 10,
    }),
  ],
  providers: [
    OrderStore,
    CreateOrderHandler,
    OrderCreatedHandler,
    GetOrderHandler,
    OrderService,
  ],
})
export class AppModule {}
```

Run it against a live broker:

```bash
cd examples
npm install
npx tsx examples/cqrs/full-cqrs-flow/main.ts
```

Expected output:

```text
[FullCqrsFlowExample] KubeMQ microservice started
[OrderService] Step 1: Dispatching CreateOrderCommand for product=WIDGET-42
[CreateOrderHandler] Creating order ORD-001: product=WIDGET-42, qty=3
[OrderCreatedHandler] Event received — storing order ORD-001
[OrderService] Step 1 complete: orderId=ORD-001
[OrderService] Step 2: Querying order ORD-001
[GetOrderHandler] Querying order: ORD-001
[OrderService] Step 2 complete: {"orderId":"ORD-001","productId":"WIDGET-42","quantity":3,"status":"created"}
```

<Callout type="info">
  Focused single-bus examples are available too: `examples/cqrs/cqrs-commands`, `examples/cqrs/cqrs-queries`, and `examples/cqrs/cqrs-events`. Each example reads the broker address from `KUBEMQ_ADDRESS` and falls back to `localhost:50000`.
</Callout>

## Related [#related]

<Cards>
  <Card title="CQRS Bridge Concepts" href="../concepts/cqrs-bridge" description="How the bridge routes CommandBus, QueryBus, and EventBus traffic over KubeMQ." />

  <Card title="KubeMQCqrsOptions reference" href="../reference/cqrs-options" description="Every KubeMQCqrsModule.forRoot() configuration field." />

  <Card title="Usage" href="../usage" description="Commands, Queries, Events, Events Store, and Queues with the handler decorators and KubeMQRecord builder." />
</Cards>