CQRS at Scale

┌────────────────────────────────────────────────────────────────────────┐
│                         COMMAND HANDLER                                 │
│                                                                         │
│   1. Validate command                                                  │
│   2. Execute business logic                                            │
│   3. Persist to write store (transaction)                              │
│   4. Publish events to message broker     ←─── Can fail after step 3!  │
│                                                                         │
└─────────────────────────────────┬──────────────────────────────────────┘
                                  │
                                  ▼ Events
                    ┌─────────────────────────────┐
                    │      MESSAGE BROKER         │
                    │   (Kafka, RabbitMQ, etc.)   │
                    └─────────────────────────────┘
                                  │
           ┌──────────────────────┼──────────────────────┐
           │                      │                      │
           ▼                      ▼                      ▼
   ┌───────────────┐     ┌───────────────┐     ┌───────────────┐
   │  Projection 1 │     │  Projection 2 │     │  Other        │
   │  (Product     │     │  (Analytics   │     │  Consumers    │
   │   Read Model) │     │   Service)    │     │               │
   └───────────────┘     └───────────────┘     └───────────────┘

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class OrderCommandHandler {
  constructor(
    private orderRepository: OrderRepository,
    private eventPublisher: EventPublisher,
    private transactionManager: TransactionManager
  ) {}
 
  async placeOrder(command: PlaceOrderCommand): Promise<Order> {
    // Execute within transaction
    return await this.transactionManager.transaction(async (tx) => {
      // 1. Create domain entity
      const order = Order.create(command);
      
      // 2. Persist to write store
      await this.orderRepository.save(order, tx);
      
      // 3. Collect domain events
      const events = order.getUncommittedEvents();
      
      // PROBLEM: If we commit the transaction but event publish fails,
      // the order exists but no events are published
      
      return order;
    });
    
    // 4. Publish events AFTER transaction commits
    // But what if this fails? Data is saved, events are lost!
    const events = order.getUncommittedEvents();
    await this.eventPublisher.publishAll(events);
    order.markEventsAsCommitted();
    
    return order;
  }
}

┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│    PostgreSQL    │────▶│     Debezium     │────▶│      Kafka       │
│                  │ WAL │    Connector     │     │                  │
│ orders           │     │                  │     │ dbserver.public. │
│ customers        │     │  • Reads WAL     │     │ orders           │
│ products         │     │  • Transforms    │     │                  │
│                  │     │  • Publishes     │     │ dbserver.public. │
└──────────────────┘     └──────────────────┘     │ customers        │
                                                  └────────┬─────────┘
                                                           │
              ┌────────────────────────────────────────────┼─────┐
              │                                            │     │
              ▼                                            ▼     ▼
    ┌──────────────────┐                        ┌──────────────────┐
    │ Order Projection │                        │ Search Indexer   │
    │                  │                        │                  │
    │ Consumes events  │                        │ Updates          │
    │ Updates read DB  │                        │ Elasticsearch    │
    └──────────────────┘                        └──────────────────┘

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

{
  "schema": { /* Kafka Connect schema */ },
  "payload": {
    "before": null,  // Previous state (null for INSERT)
    "after": {       // New state
      "id": 12345,
      "customer_id": 789,
      "total_amount": 99.99,
      "status": "placed",
      "created_at": 1704897600000
    },
    "source": {
      "version": "2.4.0",
      "connector": "postgresql",
      "name": "dbserver",
      "ts_ms": 1704897600123,      // Transaction timestamp
      "snapshot": "false",
      "db": "ecommerce",
      "schema": "public",
      "table": "orders",
      "txId": 987654,              // Transaction ID
      "lsn": 23456789,             // Log sequence number
      "xmin": null
    },
    "op": "c",        // Operation: c=create, u=update, d=delete, r=read (snapshot)
    "ts_ms": 1704897600200,
    "transaction": {
      "id": "987654:23456789",
      "total_order": 1,
      "data_collection_order": 1
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    
    -- Event metadata
    aggregate_type VARCHAR(255) NOT NULL,    -- e.g., 'Order'
    aggregate_id VARCHAR(255) NOT NULL,      -- e.g., order ID
    event_type VARCHAR(255) NOT NULL,        -- e.g., 'OrderPlaced'
    
    -- Event payload
    payload JSONB NOT NULL,
    
    -- Tracking
    created_at TIMESTAMPTZ DEFAULT NOW(),
    published_at TIMESTAMPTZ,                -- NULL until published
    
    -- For ordering and deduplication
    sequence_number BIGSERIAL,
    
    -- Indexes for relay processing
    INDEX idx_unpublished (published_at) WHERE published_at IS NULL,
    INDEX idx_aggregate (aggregate_type, aggregate_id)
);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

class OrderCommandHandler {
  async placeOrder(command: PlaceOrderCommand): Promise<Order> {
    return await this.transactionManager.transaction(async (tx) => {
      // 1. Create and save order
      const order = Order.create(command);
      await this.orderRepository.save(order, tx);
      
      // 2. Write events to outbox in SAME transaction
      const events = order.getUncommittedEvents();
      await this.outboxRepository.insertAll(events.map(event => ({
        aggregateType: 'Order',
        aggregateId: order.id,
        eventType: event.type,
        payload: event.toJSON()
      })), tx);
      
      // 3. Transaction commits both atomically
      return order;
    });
    // Events are now safely in outbox; relay will publish them
  }
}
 
// Outbox relay process (runs separately)
class OutboxRelay {
  async processOutbox(): Promise<void> {
    while (true) {
      const events = await this.outboxRepository.getUnpublished({ limit: 100 });
      
      if (events.length === 0) {
        await sleep(100); // Polling interval
        continue;
      }
      
      for (const event of events) {
        try {
          // Publish to message broker
          await this.messageBroker.publish(
            `${event.aggregateType}.${event.eventType}`,
            event.payload,
            {
              messageId: event.id, // For deduplication
              headers: {
                'aggregate-id': event.aggregateId,
                'sequence': event.sequenceNumber.toString()
              }
            }
          );
          
          // Mark as published
          await this.outboxRepository.markPublished(event.id);
        } catch (error) {
          // Log error; will retry on next poll
          console.error(`Failed to publish event ${event.id}: ${error}`);
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

// Producer configuration for reliable publishing
const producer = kafka.producer({
  idempotent: true,                  // Prevent duplicate messages on retry
  maxInFlightRequests: 5,            // Balance throughput and ordering
  transactionTimeout: 60000,         // For exactly-once semantics
});
 
// Topic configuration for CQRS events
const topicConfig = {
  topic: 'order-events',
  numPartitions: 12,                 // Parallelism for consumers
  replicationFactor: 3,              // Durability across broker failures
  
  // Compaction for event sourcing (keep latest per key)
  // OR deletion for pure event streaming
  cleanupPolicy: 'delete',           // or 'compact' for event store pattern
  
  retentionMs: 7 * 24 * 60 * 60 * 1000,  // 7 days retention
  minInsyncReplicas: 2,              // Require 2 replicas for durability
};
 
// Consumer configuration for projections
const consumer = kafka.consumer({
  groupId: 'order-projection-v1',    // Change for projection rebuild
  
  // Start from earliest to rebuild, or latest for new projections
  fromBeginning: true,
  
  // Manual commits for at-least-once
  autoCommit: false,
  
  // Tune for latency vs throughput
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
  maxWaitTimeInMs: 500,
});
 
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value.toString());
    
    // Process the event
    await projection.apply(event);
    
    // Commit offset after successful processing
    await consumer.commitOffsets([{
      topic,
      partition,
      offset: (Number(message.offset) + 1).toString()
    }]);
  }
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

// Pattern 1: Event ID tracking
class IdempotentProjection {
  async apply(event: DomainEvent): Promise<void> {
    // Check if already processed
    const processed = await this.eventTracker.isProcessed(event.id);
    if (processed) {
      console.log(`Skipping duplicate event: ${event.id}`);
      return;
    }
    
    // Process event
    await this.doApply(event);
    
    // Mark as processed (atomically with the update if possible)
    await this.eventTracker.markProcessed(event.id);
  }
}
 
// Pattern 2: Version/sequence-based idempotency
class VersionedProjection {
  async apply(event: OrderEvent): Promise<void> {
    const current = await this.readStore.getOrder(event.orderId);
    
    // Version check: only apply if event is newer
    if (current && event.version <= current.lastAppliedVersion) {
      console.log(`Skipping outdated event: v${event.version} <= v${current.lastAppliedVersion}`);
      return;
    }
    
    // Apply and update version
    const updated = this.applyEvent(current, event);
    updated.lastAppliedVersion = event.version;
    await this.readStore.save(updated);
  }
}
 
// Pattern 3: Conditional/upsert operations
class ConditionalProjection {
  async handleOrderPlaced(event: OrderPlacedEvent): Promise<void> {
    // Use database upsert - naturally idempotent
    await this.readStore.upsertOrder({
      id: event.orderId,
      customerId: event.customerId,
      items: event.items,
      status: 'placed',
      placedAt: event.timestamp,
    });
  }
  
  async handleInventoryDecremented(event: InventoryDecrementedEvent): Promise<void> {
    // Use conditional update - only apply once
    const result = await this.readStore.updateInventoryIfEventNotApplied(
      event.productId,
      event.eventId,
      -event.quantity
    );
    
    if (result.rowsAffected === 0) {
      console.log(`Event already applied: ${event.eventId}`);
    }
  }
}
 
// SQL implementation of conditional update
```sql
-- Table has 'applied_events' JSONB column tracking processed event IDs
UPDATE inventory_read_model
SET 
  quantity = quantity - $quantity,
  applied_events = applied_events || jsonb_build_array($eventId)
WHERE 
  product_id = $productId
  AND NOT applied_events ? $eventId;  -- Only if not already applied
```

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

interface RetryPolicy {
  maxRetries: number;
  backoffMs: number;
  backoffMultiplier: number;
  maxBackoffMs: number;
}
 
class ResilientProjectionConsumer {
  private retryPolicy: RetryPolicy = {
    maxRetries: 5,
    backoffMs: 100,
    backoffMultiplier: 2,
    maxBackoffMs: 30000
  };
  
  async processEvent(event: DomainEvent): Promise<void> {
    let attempt = 0;
    let lastError: Error;
    
    while (attempt < this.retryPolicy.maxRetries) {
      try {
        await this.projection.apply(event);
        return; // Success
      } catch (error) {
        lastError = error;
        attempt++;
        
        // Classify error
        if (this.isPermanentFailure(error)) {
          // Don't retry permanent failures (e.g., validation errors)
          break;
        }
        
        if (attempt < this.retryPolicy.maxRetries) {
          // Exponential backoff
          const backoffMs = Math.min(
            this.retryPolicy.backoffMs * Math.pow(this.retryPolicy.backoffMultiplier, attempt - 1),
            this.retryPolicy.maxBackoffMs
          );
          
          console.log(`Retry ${attempt}/${this.retryPolicy.maxRetries} in ${backoffMs}ms for event ${event.id}`);
          await sleep(backoffMs);
        }
      }
    }
    
    // All retries exhausted - send to DLQ
    await this.sendToDeadLetterQueue(event, lastError);
  }
  
  private isPermanentFailure(error: Error): boolean {
    // Errors that won't succeed on retry
    return (
      error instanceof ValidationError ||
      error instanceof MalformedEventError ||
      error instanceof BusinessRuleViolation
    );
  }
  
  private async sendToDeadLetterQueue(event: DomainEvent, error: Error): Promise<void> {
    const dlqMessage = {
      originalEvent: event,
      error: {
        message: error.message,
        stack: error.stack,
        name: error.name
      },
      failedAt: new Date().toISOString(),
      projectionName: this.projection.name,
      attempts: this.retryPolicy.maxRetries
    };
    
    // Publish to DLQ topic
    await this.dlqProducer.publish('projection-dlq', dlqMessage);
    
    // Increment DLQ metric for alerting
    this.metrics.increment('projection.dlq.count', {
      projection: this.projection.name,
      eventType: event.type
    });
    
    console.error(`Event sent to DLQ: ${event.id}`, error);
  }
}
 
// DLQ processing service (manual or automated)
class DLQProcessor {
  async replayEvent(dlqMessageId: string): Promise<void> {
    const dlqMessage = await this.dlqStore.getMessage(dlqMessageId);
    
    // Attempt to reprocess (after fixing the underlying issue)
    await this.projectionConsumer.processEvent(dlqMessage.originalEvent);
    
    // If successful, remove from DLQ
    await this.dlqStore.markResolved(dlqMessageId);
  }
  
  async skipEvent(dlqMessageId: string, reason: string): Promise<void> {
    // Mark as intentionally skipped (e.g., corrupted event)
    await this.dlqStore.markSkipped(dlqMessageId, reason);
  }
}