cqrs
Command Query Responsibility Segregation — separate read/write models, event buses, and projections.
SKILL.md.claude/skills/cqrs/SKILL.md
GitHubCQRS Patterns
Purpose
Provide expert guidance on Command Query Responsibility Segregation (CQRS) — separating read and write models, implementing command/query buses, building projections, handling eventual consistency, and knowing when CQRS is (and is not) appropriate. Covers both simple CQRS (separate read/write repos) and full CQRS with event sourcing.
When to Use CQRS
Good fit:
- Read and write workloads have vastly different scaling needs
- Complex domain logic on writes, simple reads
- Multiple read representations of the same data (list view, dashboard, search)
- Event sourcing is already in use
- Microservice with distinct read/write patterns
Bad fit:
- Simple CRUD with no complex business logic
- Small-scale app with uniform read/write patterns
- Team unfamiliar with eventual consistency tradeoffs
Architecture Overview
┌──────────┐ Command ┌─────────────┐ Events ┌─────────────┐
│ Client │ ─────────────→ │ Write Side │ ─────────────→ │ Event Bus │
│ │ │ (Commands) │ │ │
│ │ Query ├─────────────┤ ├─────────────┤
│ │ ←───────────── │ Read Side │ ←───────────── │ Projections │
│ │ │ (Queries) │ │ │
└──────────┘ └─────────────┘ └─────────────┘
Write DB Read DB(s)
Command Side (Write Model)
Command definition:
// commands/create-order.command.ts
interface Command {
readonly type: string;
readonly timestamp: Date;
readonly metadata: { userId: string; correlationId: string };
}
interface CreateOrderCommand extends Command {
type: 'CreateOrder';
payload: {
customerId: string;
items: Array<{ productId: string; quantity: number; price: number }>;
shippingAddress: Address;
};
}
interface CancelOrderCommand extends Command {
type: 'CancelOrder';
payload: { orderId: string; reason: string };
}
Command handler:
// handlers/create-order.handler.ts
class CreateOrderHandler implements CommandHandler<CreateOrderCommand> {
constructor(
private readonly orderRepo: OrderWriteRepository,
private readonly eventBus: EventBus,
private readonly inventoryService: InventoryService,
) {}
async execute(command: CreateOrderCommand): Promise<string> {
// Validate business rules against the write model
const { customerId, items, shippingAddress } = command.payload;
// Check inventory (synchronous consistency boundary)
for (const item of items) {
const available = await this.inventoryService.checkAvailability(
item.productId,
item.quantity,
);
if (!available) {
throw new InsufficientInventoryError(item.productId);
}
}
// Create aggregate
const order = Order.create({
customerId,
items,
shippingAddress,
});
// Persist write model
await this.orderRepo.save(order);
// Publish domain events (async projection update)
await this.eventBus.publish(
new OrderCreatedEvent({
orderId: order.id,
customerId,
items,
total: order.total,
createdAt: order.createdAt,
}),
);
return order.id;
}
}
Command bus:
// bus/command-bus.ts
type CommandHandler<T extends Command> = {
execute(command: T): Promise<any>;
};
class CommandBus {
private handlers = new Map<string, CommandHandler<any>>();
register<T extends Command>(type: string, handler: CommandHandler<T>) {
if (this.handlers.has(type)) {
throw new Error(`Handler already registered for command: ${type}`);
}
this.handlers.set(type, handler);
}
async dispatch<T extends Command>(command: T): Promise<any> {
const handler = this.handlers.get(command.type);
if (!handler) {
throw new Error(`No handler for command: ${command.type}`);
}
return handler.execute(command);
}
}
Query Side (Read Model)
Query definition and handler:
// queries/get-order-summary.query.ts
interface GetOrderSummaryQuery {
type: 'GetOrderSummary';
orderId: string;
}
interface OrderSummaryDto {
orderId: string;
customerName: string;
itemCount: number;
total: number;
status: string;
createdAt: Date;
}
class GetOrderSummaryHandler implements QueryHandler<GetOrderSummaryQuery, OrderSummaryDto> {
constructor(private readonly readDb: ReadDatabase) {}
async execute(query: GetOrderSummaryQuery): Promise<OrderSummaryDto> {
// Read from the denormalized read model — optimized for this query
const row = await this.readDb.query(
`SELECT order_id, customer_name, item_count, total, status, created_at
FROM order_summaries
WHERE order_id = $1`,
[query.orderId],
);
if (!row) throw new NotFoundException(`Order ${query.orderId} not found`);
return row;
}
}
Query bus:
class QueryBus {
private handlers = new Map<string, QueryHandler<any, any>>();
register<Q, R>(type: string, handler: QueryHandler<Q, R>) {
this.handlers.set(type, handler);
}
async execute<Q extends { type: string }, R>(query: Q): Promise<R> {
const handler = this.handlers.get(query.type);
if (!handler) throw new Error(`No handler for query: ${query.type}`);
return handler.execute(query);
}
}
Projections
Projections transform domain events into read-optimized views:
// projections/order-summary.projection.ts
class OrderSummaryProjection implements EventHandler {
constructor(private readonly readDb: ReadDatabase) {}
// Each method handles one event type
async onOrderCreated(event: OrderCreatedEvent) {
await this.readDb.query(
`INSERT INTO order_summaries (order_id, customer_name, item_count, total, status, created_at)
VALUES ($1, $2, $3, $4, $5, $6)`,
[
event.orderId,
event.customerName,
event.items.length,
event.total,
'pending',
event.createdAt,
],
);
}
async onOrderShipped(event: OrderShippedEvent) {
await this.readDb.query(
`UPDATE order_summaries SET status = 'shipped', shipped_at = $2 WHERE order_id = $1`,
[event.orderId, event.shippedAt],
);
}
async onOrderCancelled(event: OrderCancelledEvent) {
await this.readDb.query(
`UPDATE order_summaries SET status = 'cancelled', cancel_reason = $2 WHERE order_id = $1`,
[event.orderId, event.reason],
);
}
}
// Multiple projections from the same events
class CustomerDashboardProjection implements EventHandler {
async onOrderCreated(event: OrderCreatedEvent) {
await this.readDb.query(
`UPDATE customer_dashboards
SET total_orders = total_orders + 1,
total_spent = total_spent + $2,
last_order_at = $3
WHERE customer_id = $1`,
[event.customerId, event.total, event.createdAt],
);
}
}
Event Bus Implementation
// bus/event-bus.ts
type EventSubscriber = (event: DomainEvent) => Promise<void>;
class InMemoryEventBus implements EventBus {
private subscribers = new Map<string, EventSubscriber[]>();
subscribe(eventType: string, handler: EventSubscriber) {
const handlers = this.subscribers.get(eventType) ?? [];
handlers.push(handler);
this.subscribers.set(eventType, handlers);
}
async publish(event: DomainEvent) {
const handlers = this.subscribers.get(event.type) ?? [];
// Run projections in parallel — each is independent
await Promise.allSettled(handlers.map((h) => h(event)));
}
}
// For production: use a persistent message broker
class RabbitMQEventBus implements EventBus {
async publish(event: DomainEvent) {
await this.channel.publish(
'domain-events',
event.type,
Buffer.from(JSON.stringify(event)),
{ persistent: true, messageId: event.id },
);
}
}
Eventual Consistency Handling
Client-side strategies:
// 1. Optimistic UI — update immediately, reconcile later
async function createOrder(data: CreateOrderInput) {
// Optimistically add to local state
addOptimisticOrder(data);
// Send command
const orderId = await commandBus.dispatch({
type: 'CreateOrder',
payload: data,
});
// Poll or use WebSocket for projection to catch up
await waitForProjection('order_summaries', orderId);
}
// 2. Return write model result for the creating user
async function createOrderEndpoint(req, res) {
const order = await commandBus.dispatch(createOrderCommand);
// Return the write model result directly — no eventual consistency for creator
res.status(201).json({
orderId: order.id,
status: order.status,
total: order.total,
});
}
// 3. Versioned reads — client sends last-known version
async function getOrder(req, res) {
const minVersion = parseInt(req.headers['if-none-match'] ?? '0');
const order = await queryBus.execute({
type: 'GetOrderSummary',
orderId: req.params.id,
});
if (order.version <= minVersion) {
// Projection hasn't caught up yet — retry or return 304
return res.status(304).end();
}
res.set('ETag', String(order.version));
res.json(order);
}
Best Practices
- Start simple — Separate read/write repos before adding event buses and projections.
- One command, one handler — Each command has exactly one handler.
- Queries are side-effect free — Read operations never modify state.
- Commands return minimal data — Typically just the ID; clients query for details.
- Projections must be idempotent — Events may be replayed during rebuilds.
- Use persistent event storage — Required for projection rebuilds and auditing.
- Handle projection lag — Clients must tolerate eventual consistency.
- Separate read databases — Denormalized, optimized per query pattern.
- Version projections — Allow rebuilding from events without downtime.
- Monitor projection lag — Alert when read models fall behind the write model.
Common Pitfalls
| Pitfall | Problem | Fix |
|---|---|---|
| CQRS for simple CRUD | Unnecessary complexity | Use CQRS only when read/write patterns diverge significantly |
| Querying the write model | Defeats the purpose of separation | Read side should have its own optimized store |
| Non-idempotent projections | Replaying events corrupts read model | Use upserts or track processed event IDs |
| Synchronous projections | Command latency includes projection time | Project asynchronously via event bus |
| Ignoring eventual consistency | UI shows stale data | Use optimistic UI, polling, or WebSocket notifications |
| Missing correlation IDs | Cannot trace command through projections | Include correlationId in all commands and events |