CQRS and Event Sourcing patterns

// events/order-events.ts
export interface BaseEvent {
  eventId: string;
  aggregateId: string;
  aggregateType: string;
  timestamp: Date;
  version: number;
}

export interface OrderCreated extends BaseEvent {
  type: 'OrderCreated';
  payload: {
    customerId: string;
    items: Array<{ productId: string; quantity: number; price: number }>;
    totalAmount: number;
  };
}

export interface OrderItemAdded extends BaseEvent {
  type: 'OrderItemAdded';
  payload: {
    productId: string;
    quantity: number;
    price: number;
  };
}

export type OrderEvent = OrderCreated | OrderItemAdded;

// domain/order-aggregate.ts
import { OrderEvent, OrderCreated, OrderItemAdded } from '../events/order-events';

export class OrderAggregate {
  public id: string;
  public customerId: string = '';
  public items: Array<{ productId: string; quantity: number; price: number }> = [];
  public totalAmount: number = 0;
  public version: number = 0;

  constructor(id: string) {
    this.id = id;
  }

  // Replay events to rebuild state
  public loadFromHistory(events: OrderEvent[]): void {
    for (const event of events) {
      this.applyEvent(event);
    }
  }

  // Apply single event to mutate aggregate
  private applyEvent(event: OrderEvent): void {
    switch (event.type) {
      case 'OrderCreated':
        this.customerId = event.payload.customerId;
        this.items = [...event.payload.items];
        this.totalAmount = event.payload.totalAmount;
        break;
      case 'OrderItemAdded':
        this.items.push(event.payload);
        this.totalAmount += event.payload.price * event.payload.quantity;
        break;
    }
    this.version = event.version;
  }

  // Command handler: creates order
  public static create(id: string, customerId: string, items: Array<{ productId: string; quantity: number; price: number }>): OrderEvent[] {
    const total = items.reduce((sum, i) => sum + i.price * i.quantity, 0);
    return [{
      eventId: crypto.randomUUID(),
      aggregateId: id,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: 1,
      type: 'OrderCreated',
      payload: { customerId, items, totalAmount: total }
    }];
  }

  // Command handler: adds item
  public addItem(productId: string, quantity: number, price: number): OrderEvent[] {
    if (this.version === 0) throw new Error('Aggregate not initialized');
    return [{
      eventId: crypto.randomUUID(),
      aggregateId: this.id,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: this.version + 1,
      type: 'OrderItemAdded',
      payload: { productId, quantity, price }
    }];
  }
}

// infrastructure/event-store.ts
import { OrderEvent } from '../events/order-events';

export interface IEventStore {
  append(aggregateId: string, expectedVersion: number, events: OrderEvent[]): Promise<void>;
  load(aggregate

Id: string): Promise<OrderEvent[]>; }

// Example using PostgreSQL with JSONB (production would use dedicated ES like EventStoreDB or Kafka) export class PostgresEventStore implements IEventStore { constructor(private pool: any) {}

async load(aggregateId: string): Promise<OrderEvent[]> { const res = await this.pool.query( 'SELECT payload, version FROM events WHERE aggregate_id = $1 ORDER BY version ASC', [aggregateId] ); return res.rows.map((r: any) => ({ ...JSON.parse(r.payload), version: r.version })); }

async append(aggregateId: string, expectedVersion: number, events: OrderEvent[]): Promise<void> { // Optimistic concurrency: version must match exactly const client = await this.pool.connect(); try { await client.query('BEGIN'); for (const event of events) { if (event.version !== expectedVersion + 1) { throw new Error(Concurrency conflict: expected version ${expectedVersion + 1}, got ${event.version}); } await client.query( 'INSERT INTO events (aggregate_id, version, payload, created_at) VALUES ($1, $2, $3, NOW())', [aggregateId, event.version, JSON.stringify(event)] ); expectedVersion = event.version; } await client.query('COMMIT'); } catch (e) { await client.query('ROLLBACK'); throw e; } finally { client.release(); } } }

### Step 4: Query Side Projection
Projections consume events asynchronously and maintain denormalized read models optimized for specific queries.

```typescript
// projections/order-read-model.ts
import { OrderEvent } from '../events/order-events';

export class OrderReadModelProjection {
  constructor(private db: any) {}

  async handle(event: OrderEvent): Promise<void> {
    switch (event.type) {
      case 'OrderCreated':
        await this.db.query(
          `INSERT INTO order_read (order_id, customer_id, total_amount, status, created_at)
           VALUES ($1, $2, $3, 'CREATED', NOW())`,
          [event.aggregateId, event.payload.customerId, event.payload.totalAmount]
        );
        break;
      case 'OrderItemAdded':
        await this.db.query(
          `UPDATE order_read 
           SET total_amount = total_amount + $1, item_count = COALESCE(item_count, 0) + 1
           WHERE order_id = $2`,
          [event.payload.price * event.payload.quantity, event.aggregateId]
        );
        break;
    }
  }
}

Step 5: Command Handler & Message Routing

Commands enter through the command side, mutate aggregates, persist events, and publish them to a message broker for projection consumption.

// application/order-command-handler.ts
import { OrderAggregate } from '../domain/order-aggregate';
import { IEventStore } from '../infrastructure/event-store';
import { EventBus } from '../infrastructure/event-bus';

export class OrderCommandHandler {
  constructor(
    private eventStore: IEventStore,
    private eventBus: EventBus
  ) {}

  async handleCreateOrder(command: { orderId: string; customerId: string; items: any[] }): Promise<void> {
    const aggregate = OrderAggregate.create(command.orderId, command.customerId, command.items);
    await this.eventStore.append(command.orderId, 0, aggregate);
    await this.eventBus.publish(aggregate);
  }

  async handleAddItem(command: { orderId: string; productId: string; quantity: number; price: number }): Promise<void> {
    const history = await this.eventStore.load(command.orderId);
    const aggregate = new OrderAggregate(command.orderId);
    aggregate.loadFromHistory(history);
    
    const newEvents = aggregate.addItem(command.productId, command.quantity, command.price);
    await this.eventStore.append(command.orderId, aggregate.version, newEvents);
    await this.eventBus.publish(newEvents);
  }
}

Architecture Decisions & Rationale

• Append-only event store: Guarantees immutability. Enables temporal queries, debugging, and replay without snapshot corruption.
• Optimistic concurrency via versioning: Prevents lost updates in distributed environments. Fails fast on version mismatch rather than overwriting state.
• Asynchronous projections: Decouples write latency from read model complexity. Allows independent scaling and technology selection (e.g., Elasticsearch for search, Redis for caching, PostgreSQL for relational queries).
• Aggregate as consistency boundary: Commands validate against loaded state. Events are emitted only after successful validation. This eliminates partial updates and enforces domain invariants.
• Idempotency at command/event level: Critical for exactly-once processing in distributed systems. Handled via unique command IDs or event deduplication tables.

Pitfall Guide

1. Treating events as database rows Events represent business facts, not column updates. Storing UPDATE orders SET status = 'SHIPPED' as an event violates the pattern. Events should be domain-language statements: OrderShipped, PaymentReceived. Database deltas create tight coupling to schema changes and break replay capability.

2. Ignoring event schema evolution Events are immutable, but business requirements change. Teams that freeze event schemas cannot adapt. Teams that mutate them break replay. Solution: implement explicit versioning (event.version) and backward-compatible deserialization. Use upcasters to transform legacy events during projection replay.

3. Synchronous projections blocking commands Running projections in the same transaction as command execution kills write throughput and couples read/write models. Projections must be asynchronous. Use a message broker or stream processor (Kafka, RabbitMQ, AWS EventBridge) with retry and dead-letter queue handling.

4. Missing idempotency controls Distributed systems retry. Without idempotency, duplicate commands create duplicate events or corrupt read models. Implement command deduplication using a command_id table with unique constraints. Event handlers should check processed event IDs before applying projections.

5. Over-partitioning aggregates Creating an aggregate per UI form or database table leads to chatty systems with high latency and complex consistency rules. Aggregates must represent true consistency boundaries. If two entities must never be in an invalid state together, they belong in the same aggregate. Otherwise, use eventual consistency and sagas.

6. Assuming strong consistency across models CQRS inherently uses eventual consistency between command and query sides. Teams that demand strong consistency defeat the pattern's purpose. Design UI/UX to handle pending states, use optimistic updates, and implement conflict resolution strategies at the domain level.

7. Storing derived state instead of events Caching totalAmount in the event payload without storing the raw line items breaks auditability. Events should contain the minimal data required to reconstruct state. Derived values belong in projections, not events.

Best practices from production:

• Implement a dedicated projection worker with consumer lag monitoring.
• Use structured logging with correlation IDs across command → event → projection.
• Run periodic replay tests in staging to validate projection correctness.
• Enforce event naming conventions: EntityActioned (past tense, domain language).
• Separate infrastructure concerns: event store, message broker, and read DB should be independently scalable.

Production Bundle

Action Checklist

• Define aggregate boundaries based on consistency requirements, not UI layouts
• Design event schema with explicit versioning and backward-compatible fields
• Implement optimistic concurrency control using aggregate version tracking
• Build asynchronous projection pipeline with retry and dead-letter handling
• Add command and event idempotency checks using unique identifiers
• Set up consumer lag monitoring and projection replay automation
• Document consistency guarantees and UI handling for eventual states

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
High read asymmetry (10:1 read:write)	CQRS + Denormalized Read DB	Read models optimized for specific queries reduce latency and offload primary DB	Moderate infra cost, lower query compute
Strict audit/compliance requirements	Event Sourcing + Append-Only Store	Immutable event log enables deterministic replay and regulatory reporting	Higher storage cost, lower audit engineering cost
Simple CRUD domain (user profiles, settings)	Traditional RDBMS with soft deletes	Overhead of ES/CQRS outweighs benefits for low-complexity state	Lowest dev/infra cost
Real-time sync across multiple clients	CQRS + WebSocket/SSE projections	Read models push updates asynchronously without blocking commands	Moderate infra, high UX value
Multi-tenant SaaS with complex billing	Event Sourcing + Saga orchestration	Financial state requires auditability and distributed transaction compensation	High initial dev cost, low compliance risk

Configuration Template

// infrastructure/event-bridge.config.ts
export const EventBridgeConfig = {
  eventStore: {
    type: 'postgresql',
    connectionString: process.env.EVENT_STORE_DB_URL,
    maxRetries: 3,
    concurrencyCheck: 'version_based'
  },
  messageBroker: {
    type: 'rabbitmq',
    url: process.env.RABBITMQ_URL,
    exchange: 'domain_events',
    queuePrefix: 'projection_',
    prefetchCount: 10,
    deadLetterQueue: 'dlq_failed_events'
  },
  projections: {
    orderReadModel: {
      targetDb: 'read_replica',
      batchSize: 50,
      checkpointTable: 'projection_checkpoints',
      retryDelayMs: 1000,
      maxRetryAttempts: 5
    }
  },
  idempotency: {
    commandTable: 'processed_commands',
    eventTable: 'processed_events',
    ttlHours: 72
  }
};

Quick Start Guide

1. Initialize local infrastructure: Run docker compose up -d with PostgreSQL (event store), PostgreSQL (read model), and RabbitMQ. Use the provided EventBridgeConfig with local connection strings.
2. Scaffold the domain: Create the aggregate root, event types, and command handler using the TypeScript examples. Ensure version tracking and loadFromHistory are implemented.
3. Deploy the projection worker: Start a separate Node.js process that subscribes to the domain_events exchange, reads events in batches, applies the OrderReadModelProjection, and updates projection_checkpoints.
4. Validate the pipeline: Send a CreateOrder command. Verify the event is appended to the event store, the projection worker processes it asynchronously, and the read model contains the denormalized order. Check processed_commands for idempotency enforcement.
5. Test replay capability: Drop the read model table. Restart the projection worker with checkpoint reset. Verify it rebuilds state deterministically from the event log without manual intervention.