CQRS Implementation Guide

Command Query Responsibility Segregation separates read and write operations into different models. This enables independent scaling, optimization, and complexity management.

CQRS Basics#

Traditional CRUD:
- Single model for read/write
- Same database for all operations
- Trade-offs between read and write optimization

CQRS:
- Separate models for commands (writes) and queries (reads)
- Optimized data stores for each
- Can scale independently

Simple CQRS Structure#

// Commands - write operations
interface CreateOrderCommand {
  type: 'CreateOrder';
  customerId: string;
  items: OrderItem[];
}

interface CancelOrderCommand {
  type: 'CancelOrder';
  orderId: string;
  reason: string;
}

type Command = CreateOrderCommand | CancelOrderCommand;

// Command handler
class OrderCommandHandler {
  constructor(
    private repository: OrderRepository,
    private eventBus: EventBus
  ) {}

  async handle(command: Command): Promise<void> {
    switch (command.type) {
      case 'CreateOrder':
        await this.createOrder(command);
        break;
      case 'CancelOrder':
        await this.cancelOrder(command);
        break;
    }
  }

  private async createOrder(cmd: CreateOrderCommand): Promise<void> {
    const order = Order.create(cmd.customerId, cmd.items);
    await this.repository.save(order);

    await this.eventBus.publish({
      type: 'OrderCreated',
      data: {
        orderId: order.id,
        customerId: order.customerId,
        items: order.items,
        total: order.total,
      },
    });
  }

  private async cancelOrder(cmd: CancelOrderCommand): Promise<void> {
    const order = await this.repository.get(cmd.orderId);
    order.cancel(cmd.reason);
    await this.repository.save(order);

    await this.eventBus.publish({
      type: 'OrderCancelled',
      data: {
        orderId: order.id,
        reason: cmd.reason,
      },
    });
  }
}

Query Side#

// Queries - read operations
interface GetOrderQuery {
  orderId: string;
}

interface ListOrdersQuery {
  customerId: string;
  status?: string;
  limit?: number;
  cursor?: string;
}

// Read model - optimized for queries
interface OrderReadModel {
  id: string;
  customerName: string;
  customerEmail: string;
  items: {
    productName: string;
    quantity: number;
    price: number;
  }[];
  total: number;
  status: string;
  createdAt: Date;
}

// Query handler
class OrderQueryHandler {
  constructor(private readDb: ReadDatabase) {}

  async getOrder(query: GetOrderQuery): Promise<OrderReadModel | null> {
    return this.readDb.orders.findUnique({
      where: { id: query.orderId },
    });
  }

  async listOrders(query: ListOrdersQuery): Promise<PaginatedResult<OrderReadModel>> {
    const orders = await this.readDb.orders.findMany({
      where: {
        customerId: query.customerId,
        status: query.status,
      },
      take: query.limit || 20,
      cursor: query.cursor ? { id: query.cursor } : undefined,
      orderBy: { createdAt: 'desc' },
    });

    return {
      data: orders,
      hasMore: orders.length === (query.limit || 20),
      nextCursor: orders[orders.length - 1]?.id,
    };
  }
}

Projections (Updating Read Models)#

// Event handler that updates read models
class OrderProjection {
  constructor(private readDb: ReadDatabase) {}

  async handle(event: DomainEvent): Promise<void> {
    switch (event.type) {
      case 'OrderCreated':
        await this.onOrderCreated(event.data);
        break;
      case 'OrderCancelled':
        await this.onOrderCancelled(event.data);
        break;
      case 'OrderShipped':
        await this.onOrderShipped(event.data);
        break;
    }
  }

  private async onOrderCreated(data: OrderCreatedEvent): Promise<void> {
    // Fetch additional data for denormalization
    const customer = await this.customerService.get(data.customerId);
    const products = await this.productService.getMany(
      data.items.map((i) => i.productId)
    );

    await this.readDb.orders.create({
      data: {
        id: data.orderId,
        customerId: data.customerId,
        customerName: customer.name,
        customerEmail: customer.email,
        items: data.items.map((item) => ({
          productId: item.productId,
          productName: products.find((p) => p.id === item.productId)!.name,
          quantity: item.quantity,
          price: item.price,
        })),
        total: data.total,
        status: 'pending',
        createdAt: new Date(),
      },
    });
  }

  private async onOrderCancelled(data: OrderCancelledEvent): Promise<void> {
    await this.readDb.orders.update({
      where: { id: data.orderId },
      data: {
        status: 'cancelled',
        cancelledAt: new Date(),
        cancellationReason: data.reason,
      },
    });
  }

  private async onOrderShipped(data: OrderShippedEvent): Promise<void> {
    await this.readDb.orders.update({
      where: { id: data.orderId },
      data: {
        status: 'shipped',
        shippedAt: new Date(),
        trackingNumber: data.trackingNumber,
      },
    });
  }
}

Event Sourcing Integration#

// Event store
interface Event {
  id: string;
  aggregateId: string;
  type: string;
  data: any;
  version: number;
  timestamp: Date;
}

class EventStore {
  constructor(private db: Database) {}

  async save(aggregateId: string, events: Event[], expectedVersion: number): Promise<void> {
    const currentVersion = await this.getVersion(aggregateId);

    if (currentVersion !== expectedVersion) {
      throw new ConcurrencyError(
        `Expected version ${expectedVersion}, found ${currentVersion}`
      );
    }

    await this.db.transaction(async (tx) => {
      for (let i = 0; i < events.length; i++) {
        await tx.events.create({
          data: {
            ...events[i],
            version: expectedVersion + i + 1,
          },
        });
      }
    });
  }

  async getEvents(aggregateId: string, fromVersion = 0): Promise<Event[]> {
    return this.db.events.findMany({
      where: {
        aggregateId,
        version: { gt: fromVersion },
      },
      orderBy: { version: 'asc' },
    });
  }

  async getVersion(aggregateId: string): Promise<number> {
    const lastEvent = await this.db.events.findFirst({
      where: { aggregateId },
      orderBy: { version: 'desc' },
    });
    return lastEvent?.version ?? 0;
  }
}

// Rebuild aggregate from events
class OrderAggregate {
  private state: OrderState;
  private version: number = 0;
  private uncommittedEvents: Event[] = [];

  static async load(eventStore: EventStore, id: string): Promise<OrderAggregate> {
    const events = await eventStore.getEvents(id);
    const aggregate = new OrderAggregate();

    for (const event of events) {
      aggregate.apply(event);
      aggregate.version = event.version;
    }

    return aggregate;
  }

  private apply(event: Event): void {
    switch (event.type) {
      case 'OrderCreated':
        this.state = {
          id: event.data.orderId,
          status: 'pending',
          items: event.data.items,
        };
        break;
      case 'OrderCancelled':
        this.state.status = 'cancelled';
        break;
    }
  }

  cancel(reason: string): void {
    if (this.state.status !== 'pending') {
      throw new Error('Can only cancel pending orders');
    }

    const event: Event = {
      id: crypto.randomUUID(),
      aggregateId: this.state.id,
      type: 'OrderCancelled',
      data: { reason },
      version: this.version + this.uncommittedEvents.length + 1,
      timestamp: new Date(),
    };

    this.uncommittedEvents.push(event);
    this.apply(event);
  }

  getUncommittedEvents(): Event[] {
    return this.uncommittedEvents;
  }

  clearUncommittedEvents(): void {
    this.uncommittedEvents = [];
  }
}

API Layer#

// Express routes with CQRS
class OrderController {
  constructor(
    private commandHandler: OrderCommandHandler,
    private queryHandler: OrderQueryHandler
  ) {}

  // Commands
  async createOrder(req: Request, res: Response): Promise<void> {
    await this.commandHandler.handle({
      type: 'CreateOrder',
      customerId: req.user.id,
      items: req.body.items,
    });

    res.status(202).json({ message: 'Order creation initiated' });
  }

  async cancelOrder(req: Request, res: Response): Promise<void> {
    await this.commandHandler.handle({
      type: 'CancelOrder',
      orderId: req.params.id,
      reason: req.body.reason,
    });

    res.status(202).json({ message: 'Order cancellation initiated' });
  }

  // Queries
  async getOrder(req: Request, res: Response): Promise<void> {
    const order = await this.queryHandler.getOrder({
      orderId: req.params.id,
    });

    if (!order) {
      res.status(404).json({ error: 'Order not found' });
      return;
    }

    res.json(order);
  }

  async listOrders(req: Request, res: Response): Promise<void> {
    const orders = await this.queryHandler.listOrders({
      customerId: req.user.id,
      status: req.query.status as string,
      limit: parseInt(req.query.limit as string) || 20,
      cursor: req.query.cursor as string,
    });

    res.json(orders);
  }
}

Eventual Consistency#

// Handle eventual consistency in UI
class OrderService {
  async createOrder(items: OrderItem[]): Promise<{ orderId: string }> {
    // Send command
    const result = await api.post('/orders', { items });
    const orderId = result.orderId;

    // Poll for read model update
    const order = await this.waitForOrder(orderId);
    return order;
  }

  private async waitForOrder(orderId: string, maxAttempts = 10): Promise<Order> {
    for (let i = 0; i < maxAttempts; i++) {
      try {
        const order = await api.get(`/orders/${orderId}`);
        return order;
      } catch (error) {
        if (i === maxAttempts - 1) throw error;
        await sleep(100 * Math.pow(2, i));
      }
    }
    throw new Error('Order not found after creation');
  }
}

// Or use WebSocket for real-time updates
socket.on('OrderCreated', (order) => {
  addOrderToUI(order);
});

Best Practices#

When to Use CQRS:
✓ Complex domain logic
✓ Different read/write scaling needs
✓ Multiple read model representations
✓ Audit/history requirements

When to Avoid:
✗ Simple CRUD applications
✗ Small teams/projects
✗ Real-time consistency required

Implementation:
✓ Start simple, add complexity as needed
✓ Handle eventual consistency in UI
✓ Monitor projection lag
✓ Plan for replay scenarios

CQRS separates concerns and enables independent optimization of reads and writes. Start with simple command/query separation, add event sourcing if you need audit trails, and carefully handle eventual consistency. The complexity is justified when you have different scaling needs or complex domain logic.