Claude Code Research

Event Sourcing and CQRS Patterns with Claude Code

13 min read

Event Sourcing and CQRS Patterns with Claude Code

This guide explores how to implement Event Sourcing and Command Query Responsibility Segregation (CQRS) patterns with Claude Code, enabling scalable, auditable, and maintainable event-driven architectures.

Overview

Event Sourcing and CQRS are complementary architectural patterns that revolutionize how we handle state and data operations:

  • Event Sourcing: Stores all changes to application state as a sequence of immutable events
  • CQRS: Separates read and write operations into different models for optimization

Why Event Sourcing and CQRS with Claude Code?

  1. Complete Audit Trail: Every state change is recorded as an event
  2. Time Travel: Reconstruct application state at any point in time
  3. Scalability: Read and write models can scale independently
  4. Flexibility: Easy to add new projections and views
  5. AI Integration: Claude can analyze event streams for insights

Core Concepts

1. Commands

Commands represent an intent to change the system state:

// commands/CreateTodoList.ts
export class CreateTodoListCommand {
  constructor(
    public readonly aggregateId: string,
    public readonly name: string,
    public readonly description: string,
    public readonly expectedVersion: number = -1
  ) {}
}
 
// commands/AddTodoItem.ts
export class AddTodoItemCommand {
  constructor(
    public readonly aggregateId: string,
    public readonly itemId: string,
    public readonly title: string,
    public readonly dueDate?: Date,
    public readonly expectedVersion?: number
  ) {}
}

2. Events

Events are immutable facts that have occurred:

// events/TodoListCreated.ts
export class TodoListCreatedEvent {
  constructor(
    public readonly aggregateId: string,
    public readonly name: string,
    public readonly description: string,
    public readonly createdAt: Date,
    public readonly version: number
  ) {}
}
 
// events/TodoItemAdded.ts
export class TodoItemAddedEvent {
  constructor(
    public readonly aggregateId: string,
    public readonly itemId: string,
    public readonly title: string,
    public readonly dueDate?: Date,
    public readonly addedAt: Date = new Date(),
    public readonly version: number
  ) {}
}

3. Aggregates

Aggregates encapsulate business logic and generate events:

// aggregates/TodoListAggregate.ts
import { AggregateRoot } from './AggregateRoot';
 
export class TodoListAggregate extends AggregateRoot {
  private name: string;
  private description: string;
  private items: Map<string, TodoItem> = new Map();
  
  constructor() {
    super();
  }
  
  // Command handlers
  createTodoList(name: string, description: string): void {
    if (!name || name.trim().length === 0) {
      throw new Error('Todo list name is required');
    }
    
    this.applyChange(new TodoListCreatedEvent(
      this.id,
      name,
      description,
      new Date(),
      this.getVersion() + 1
    ));
  }
  
  addTodoItem(itemId: string, title: string, dueDate?: Date): void {
    if (this.items.has(itemId)) {
      throw new Error(`Item ${itemId} already exists`);
    }
    
    this.applyChange(new TodoItemAddedEvent(
      this.id,
      itemId,
      title,
      dueDate,
      new Date(),
      this.getVersion() + 1
    ));
  }
  
  // Event appliers
  applyTodoListCreatedEvent(event: TodoListCreatedEvent): void {
    this.name = event.name;
    this.description = event.description;
  }
  
  applyTodoItemAddedEvent(event: TodoItemAddedEvent): void {
    this.items.set(event.itemId, {
      id: event.itemId,
      title: event.title,
      dueDate: event.dueDate,
      completed: false
    });
  }
}

4. Event Store

The event store persists and retrieves events:

// infrastructure/EventStore.ts
export interface EventDescriptor {
  aggregateId: string;
  eventType: string;
  eventData: any;
  version: number;
  timestamp: Date;
}
 
export class EventStore {
  private events: EventDescriptor[] = [];
  private snapshots: Map<string, any> = new Map();
  
  async saveEvents(
    aggregateId: string,
    events: any[],
    expectedVersion: number
  ): Promise<void> {
    // Check for concurrency conflicts
    const currentVersion = await this.getCurrentVersion(aggregateId);
    if (expectedVersion !== -1 && currentVersion !== expectedVersion) {
      throw new ConcurrencyError(
        `Expected version ${expectedVersion} but current version is ${currentVersion}`
      );
    }
    
    // Save events
    let version = currentVersion;
    for (const event of events) {
      version++;
      const descriptor: EventDescriptor = {
        aggregateId,
        eventType: event.constructor.name,
        eventData: event,
        version,
        timestamp: new Date()
      };
      this.events.push(descriptor);
      
      // Publish event
      await this.publishEvent(descriptor);
    }
  }
  
  async getEventsForAggregate(
    aggregateId: string,
    fromVersion?: number
  ): Promise<EventDescriptor[]> {
    return this.events.filter(e => 
      e.aggregateId === aggregateId && 
      (!fromVersion || e.version > fromVersion)
    );
  }
  
  async saveSnapshot(aggregateId: string, snapshot: any, version: number): Promise<void> {
    this.snapshots.set(`${aggregateId}:${version}`, snapshot);
  }
  
  async getLatestSnapshot(aggregateId: string): Promise<{ snapshot: any, version: number } | null> {
    // Implementation to get latest snapshot
    const keys = Array.from(this.snapshots.keys())
      .filter(k => k.startsWith(`${aggregateId}:`))
      .sort((a, b) => {
        const versionA = parseInt(a.split(':')[1]);
        const versionB = parseInt(b.split(':')[1]);
        return versionB - versionA;
      });
    
    if (keys.length === 0) return null;
    
    const latestKey = keys[0];
    const version = parseInt(latestKey.split(':')[1]);
    return { snapshot: this.snapshots.get(latestKey), version };
  }
  
  private async getCurrentVersion(aggregateId: string): Promise<number> {
    const events = await this.getEventsForAggregate(aggregateId);
    return events.length > 0 ? events[events.length - 1].version : 0;
  }
  
  private async publishEvent(event: EventDescriptor): Promise<void> {
    // Publish to message bus
    await MessageBus.publish(event.eventType, event);
  }
}

MCP Integration

1. Event Sourcing MCP Server

Create an MCP server that exposes event sourcing capabilities:

// mcp-event-sourcing-server.ts
import { MCPServer } from '@modelcontextprotocol/sdk';
import { CommandBus } from './infrastructure/CommandBus';
import { EventStore } from './infrastructure/EventStore';
import { ProjectionEngine } from './infrastructure/ProjectionEngine';
 
class EventSourcingMCPServer extends MCPServer {
  private commandBus: CommandBus;
  private eventStore: EventStore;
  private projectionEngine: ProjectionEngine;
  
  async initialize() {
    this.eventStore = new EventStore();
    this.commandBus = new CommandBus(this.eventStore);
    this.projectionEngine = new ProjectionEngine(this.eventStore);
    
    // Register command handler tool
    this.registerTool({
      name: 'execute-command',
      description: 'Execute a command in the event-sourced system',
      parameters: {
        type: 'object',
        properties: {
          commandType: { type: 'string' },
          aggregateId: { type: 'string' },
          payload: { type: 'object' },
          expectedVersion: { type: 'number' }
        }
      },
      execute: async ({ commandType, aggregateId, payload, expectedVersion }) => {
        const command = this.createCommand(commandType, aggregateId, payload, expectedVersion);
        const result = await this.commandBus.send(command);
        return { success: true, result };
      }
    });
    
    // Register query tool
    this.registerTool({
      name: 'query-projection',
      description: 'Query a read model projection',
      parameters: {
        type: 'object',
        properties: {
          projectionName: { type: 'string' },
          query: { type: 'object' }
        }
      },
      execute: async ({ projectionName, query }) => {
        const projection = this.projectionEngine.getProjection(projectionName);
        const result = await projection.query(query);
        return { result };
      }
    });
    
    // Register event replay tool
    this.registerTool({
      name: 'replay-events',
      description: 'Replay events to rebuild projections',
      parameters: {
        type: 'object',
        properties: {
          aggregateId: { type: 'string', optional: true },
          fromVersion: { type: 'number', optional: true },
          toVersion: { type: 'number', optional: true }
        }
      },
      execute: async ({ aggregateId, fromVersion, toVersion }) => {
        const events = await this.eventStore.getEventsForAggregate(
          aggregateId,
          fromVersion
        );
        
        let replayedCount = 0;
        for (const event of events) {
          if (toVersion && event.version > toVersion) break;
          await this.projectionEngine.handleEvent(event);
          replayedCount++;
        }
        
        return { replayedCount };
      }
    });
  }
  
  private createCommand(type: string, aggregateId: string, payload: any, expectedVersion?: number): any {
    // Command factory logic
    switch (type) {
      case 'CreateTodoList':
        return new CreateTodoListCommand(
          aggregateId,
          payload.name,
          payload.description,
          expectedVersion || -1
        );
      case 'AddTodoItem':
        return new AddTodoItemCommand(
          aggregateId,
          payload.itemId,
          payload.title,
          payload.dueDate,
          expectedVersion
        );
      default:
        throw new Error(`Unknown command type: ${type}`);
    }
  }
}

2. Claude Code Configuration

{
  "mcpServers": {
    "event-sourcing": {
      "command": "node",
      "args": ["mcp-event-sourcing-server.js"],
      "env": {
        "EVENT_STORE_PATH": "./data/events",
        "SNAPSHOT_INTERVAL": "100"
      }
    }
  }
}

Advanced Patterns

1. Saga Pattern for Long-Running Processes

// sagas/OrderFulfillmentSaga.ts
export class OrderFulfillmentSaga {
  private state: 'started' | 'payment_pending' | 'inventory_reserved' | 'completed' | 'failed';
  private orderId: string;
  private compensations: Array<() => Promise<void>> = [];
  
  async handle(event: DomainEvent): Promise<void> {
    switch (event.constructor.name) {
      case 'OrderPlacedEvent':
        await this.handleOrderPlaced(event as OrderPlacedEvent);
        break;
      case 'PaymentCompletedEvent':
        await this.handlePaymentCompleted(event as PaymentCompletedEvent);
        break;
      case 'InventoryReservedEvent':
        await this.handleInventoryReserved(event as InventoryReservedEvent);
        break;
      case 'PaymentFailedEvent':
      case 'InventoryReservationFailedEvent':
        await this.compensate();
        break;
    }
  }
  
  private async handleOrderPlaced(event: OrderPlacedEvent): Promise<void> {
    this.orderId = event.orderId;
    this.state = 'started';
    
    // Request payment
    await this.commandBus.send(new ProcessPaymentCommand(
      event.orderId,
      event.totalAmount,
      event.customerId
    ));
    
    this.state = 'payment_pending';
  }
  
  private async handlePaymentCompleted(event: PaymentCompletedEvent): Promise<void> {
    if (event.orderId !== this.orderId) return;
    
    // Add compensation
    this.compensations.push(async () => {
      await this.commandBus.send(new RefundPaymentCommand(event.paymentId));
    });
    
    // Reserve inventory
    await this.commandBus.send(new ReserveInventoryCommand(
      this.orderId,
      event.items
    ));
    
    this.state = 'inventory_reserved';
  }
  
  private async compensate(): Promise<void> {
    // Run compensations in reverse order
    for (const compensation of this.compensations.reverse()) {
      await compensation();
    }
    
    this.state = 'failed';
    await this.commandBus.send(new CancelOrderCommand(this.orderId));
  }
}

2. Event Sourcing with Claude Analysis

// analysis/EventStreamAnalyzer.ts
export class EventStreamAnalyzer {
  private claudeClient: ClaudeClient;
  
  async analyzeEventPatterns(
    aggregateId: string,
    timeRange: { start: Date; end: Date }
  ): Promise<AnalysisResult> {
    // Get events for analysis
    const events = await this.eventStore.getEventsInRange(
      aggregateId,
      timeRange
    );
    
    // Prepare event data for Claude
    const eventSummary = events.map(e => ({
      type: e.eventType,
      timestamp: e.timestamp,
      data: this.sanitizeEventData(e.eventData)
    }));
    
    // Ask Claude to analyze patterns
    const analysis = await this.claudeClient.analyze({
      prompt: `Analyze these event patterns and identify:
        1. Common sequences of events
        2. Unusual patterns or anomalies
        3. Performance bottlenecks
        4. Optimization opportunities
        
        Events: ${JSON.stringify(eventSummary)}`,
      model: 'claude-3-opus-20240229'
    });
    
    return {
      patterns: analysis.patterns,
      anomalies: analysis.anomalies,
      recommendations: analysis.recommendations
    };
  }
  
  private sanitizeEventData(data: any): any {
    // Remove sensitive information before sending to Claude
    const sanitized = { ...data };
    delete sanitized.password;
    delete sanitized.creditCard;
    delete sanitized.ssn;
    return sanitized;
  }
}

3. Projection Patterns

// projections/TodoListProjection.ts
export class TodoListProjection {
  private todoLists: Map<string, TodoListReadModel> = new Map();
  private eventHandlers: Map<string, Function>;
  
  constructor() {
    this.eventHandlers = new Map([
      ['TodoListCreatedEvent', this.handleTodoListCreated.bind(this)],
      ['TodoItemAddedEvent', this.handleTodoItemAdded.bind(this)],
      ['TodoItemCompletedEvent', this.handleTodoItemCompleted.bind(this)]
    ]);
  }
  
  async handleEvent(event: EventDescriptor): Promise<void> {
    const handler = this.eventHandlers.get(event.eventType);
    if (handler) {
      await handler(event.eventData);
    }
  }
  
  private handleTodoListCreated(event: TodoListCreatedEvent): void {
    this.todoLists.set(event.aggregateId, {
      id: event.aggregateId,
      name: event.name,
      description: event.description,
      items: [],
      createdAt: event.createdAt,
      completedCount: 0,
      totalCount: 0
    });
  }
  
  private handleTodoItemAdded(event: TodoItemAddedEvent): void {
    const list = this.todoLists.get(event.aggregateId);
    if (list) {
      list.items.push({
        id: event.itemId,
        title: event.title,
        dueDate: event.dueDate,
        completed: false,
        addedAt: event.addedAt
      });
      list.totalCount++;
    }
  }
  
  // Query methods
  async getById(id: string): Promise<TodoListReadModel | null> {
    return this.todoLists.get(id) || null;
  }
  
  async getAll(): Promise<TodoListReadModel[]> {
    return Array.from(this.todoLists.values());
  }
  
  async getByDueDate(date: Date): Promise<TodoListReadModel[]> {
    return Array.from(this.todoLists.values()).filter(list =>
      list.items.some(item => 
        item.dueDate && 
        item.dueDate.toDateString() === date.toDateString()
      )
    );
  }
}

4. Snapshot Strategy

// infrastructure/SnapshotStrategy.ts
export class SnapshotStrategy {
  private snapshotInterval: number;
  
  constructor(snapshotInterval: number = 100) {
    this.snapshotInterval = snapshotInterval;
  }
  
  async loadAggregate<T extends AggregateRoot>(
    aggregateId: string,
    aggregateClass: new() => T,
    eventStore: EventStore
  ): Promise<T> {
    // Try to load from snapshot
    const snapshotData = await eventStore.getLatestSnapshot(aggregateId);
    
    let aggregate: T;
    let fromVersion = 0;
    
    if (snapshotData) {
      // Restore from snapshot
      aggregate = Object.assign(new aggregateClass(), snapshotData.snapshot);
      aggregate.restoreSnapshot(snapshotData.snapshot);
      fromVersion = snapshotData.version;
    } else {
      // Create new aggregate
      aggregate = new aggregateClass();
    }
    
    // Apply events after snapshot
    const events = await eventStore.getEventsForAggregate(
      aggregateId,
      fromVersion
    );
    
    for (const event of events) {
      aggregate.applyEvent(event.eventData);
    }
    
    // Create new snapshot if needed
    if (events.length >= this.snapshotInterval) {
      await this.createSnapshot(aggregate, eventStore);
    }
    
    return aggregate;
  }
  
  private async createSnapshot(
    aggregate: AggregateRoot,
    eventStore: EventStore
  ): Promise<void> {
    const snapshot = aggregate.getSnapshot();
    await eventStore.saveSnapshot(
      aggregate.getId(),
      snapshot,
      aggregate.getVersion()
    );
  }
}

Testing Event-Sourced Systems

1. Given-When-Then Testing

// tests/TodoListAggregate.test.ts
import { GivenWhenThen } from '../testing/GivenWhenThen';
 
describe('TodoListAggregate', () => {
  let gwt: GivenWhenThen;
  
  beforeEach(() => {
    gwt = new GivenWhenThen();
  });
  
  it('should create todo list', async () => {
    await gwt
      .given([])
      .when(new CreateTodoListCommand(
        'list-1',
        'Shopping List',
        'Weekly groceries'
      ))
      .then([
        new TodoListCreatedEvent(
          'list-1',
          'Shopping List',
          'Weekly groceries',
          expect.any(Date),
          1
        )
      ]);
  });
  
  it('should add item to existing list', async () => {
    await gwt
      .given([
        new TodoListCreatedEvent(
          'list-1',
          'Shopping List',
          'Weekly groceries',
          new Date(),
          1
        )
      ])
      .when(new AddTodoItemCommand(
        'list-1',
        'item-1',
        'Buy milk'
      ))
      .then([
        new TodoItemAddedEvent(
          'list-1',
          'item-1',
          'Buy milk',
          undefined,
          expect.any(Date),
          2
        )
      ]);
  });
  
  it('should reject duplicate items', async () => {
    await gwt
      .given([
        new TodoListCreatedEvent(
          'list-1',
          'Shopping List',
          'Weekly groceries',
          new Date(),
          1
        ),
        new TodoItemAddedEvent(
          'list-1',
          'item-1',
          'Buy milk',
          undefined,
          new Date(),
          2
        )
      ])
      .when(new AddTodoItemCommand(
        'list-1',
        'item-1',
        'Buy milk again'
      ))
      .thenThrows(new Error('Item item-1 already exists'));
  });
});

2. Projection Testing

// tests/TodoListProjection.test.ts
describe('TodoListProjection', () => {
  let projection: TodoListProjection;
  
  beforeEach(() => {
    projection = new TodoListProjection();
  });
  
  it('should build read model from events', async () => {
    // Apply events
    await projection.handleEvent({
      aggregateId: 'list-1',
      eventType: 'TodoListCreatedEvent',
      eventData: new TodoListCreatedEvent(
        'list-1',
        'Shopping List',
        'Weekly groceries',
        new Date('2024-01-01'),
        1
      ),
      version: 1,
      timestamp: new Date()
    });
    
    await projection.handleEvent({
      aggregateId: 'list-1',
      eventType: 'TodoItemAddedEvent',
      eventData: new TodoItemAddedEvent(
        'list-1',
        'item-1',
        'Buy milk',
        new Date('2024-01-05'),
        new Date('2024-01-01'),
        2
      ),
      version: 2,
      timestamp: new Date()
    });
    
    // Query projection
    const list = await projection.getById('list-1');
    expect(list).toEqual({
      id: 'list-1',
      name: 'Shopping List',
      description: 'Weekly groceries',
      items: [{
        id: 'item-1',
        title: 'Buy milk',
        dueDate: new Date('2024-01-05'),
        completed: false,
        addedAt: new Date('2024-01-01')
      }],
      createdAt: new Date('2024-01-01'),
      completedCount: 0,
      totalCount: 1
    });
  });
});

Performance Optimization

1. Event Store Optimization

// infrastructure/OptimizedEventStore.ts
export class OptimizedEventStore extends EventStore {
  private indexByType: Map<string, EventDescriptor[]> = new Map();
  private indexByAggregate: Map<string, EventDescriptor[]> = new Map();
  private eventCache: LRUCache<string, EventDescriptor>;
  
  constructor() {
    super();
    this.eventCache = new LRUCache({ max: 10000 });
  }
  
  async saveEvents(
    aggregateId: string,
    events: any[],
    expectedVersion: number
  ): Promise<void> {
    await super.saveEvents(aggregateId, events, expectedVersion);
    
    // Update indexes
    for (const event of events) {
      const descriptor = this.createDescriptor(aggregateId, event);
      
      // Type index
      if (!this.indexByType.has(event.constructor.name)) {
        this.indexByType.set(event.constructor.name, []);
      }
      this.indexByType.get(event.constructor.name)!.push(descriptor);
      
      // Aggregate index
      if (!this.indexByAggregate.has(aggregateId)) {
        this.indexByAggregate.set(aggregateId, []);
      }
      this.indexByAggregate.get(aggregateId)!.push(descriptor);
      
      // Cache
      this.eventCache.set(`${aggregateId}:${descriptor.version}`, descriptor);
    }
  }
  
  async getEventsByType(
    eventType: string,
    options?: { limit?: number; offset?: number }
  ): Promise<EventDescriptor[]> {
    const events = this.indexByType.get(eventType) || [];
    const offset = options?.offset || 0;
    const limit = options?.limit || events.length;
    
    return events.slice(offset, offset + limit);
  }
  
  async getAggregateIds(): Promise<string[]> {
    return Array.from(this.indexByAggregate.keys());
  }
}

2. Parallel Projection Processing

// infrastructure/ParallelProjectionEngine.ts
export class ParallelProjectionEngine {
  private projections: Map<string, Projection> = new Map();
  private workerPool: WorkerPool;
  
  constructor(workerCount: number = 4) {
    this.workerPool = new WorkerPool(workerCount);
  }
  
  async processEvents(events: EventDescriptor[]): Promise<void> {
    // Group events by aggregate for parallel processing
    const eventsByAggregate = this.groupByAggregate(events);
    
    // Process each aggregate's events in parallel
    const promises = Array.from(eventsByAggregate.entries()).map(
      ([aggregateId, aggregateEvents]) => 
        this.workerPool.execute(async () => {
          for (const event of aggregateEvents) {
            await this.processEvent(event);
          }
        })
    );
    
    await Promise.all(promises);
  }
  
  private groupByAggregate(
    events: EventDescriptor[]
  ): Map<string, EventDescriptor[]> {
    const grouped = new Map<string, EventDescriptor[]>();
    
    for (const event of events) {
      if (!grouped.has(event.aggregateId)) {
        grouped.set(event.aggregateId, []);
      }
      grouped.get(event.aggregateId)!.push(event);
    }
    
    return grouped;
  }
  
  private async processEvent(event: EventDescriptor): Promise<void> {
    // Process event in all projections
    const promises = Array.from(this.projections.values()).map(
      projection => projection.handleEvent(event)
    );
    
    await Promise.all(promises);
  }
}

Monitoring and Observability

1. Event Stream Monitoring

// monitoring/EventStreamMonitor.ts
export class EventStreamMonitor {
  private metrics: MetricsCollector;
  
  async monitorEventProcessing(): Promise<void> {
    // Track event processing rate
    this.metrics.gauge('events.processing.rate', async () => {
      const count = await this.eventStore.getEventCount(
        new Date(Date.now() - 60000), // Last minute
        new Date()
      );
      return count;
    });
    
    // Track event lag
    this.metrics.gauge('events.processing.lag', async () => {
      const lastProcessed = await this.getLastProcessedEventTime();
      const lastStored = await this.getLastStoredEventTime();
      return lastStored - lastProcessed;
    });
    
    // Track projection rebuild time
    this.metrics.histogram('projections.rebuild.duration');
    
    // Alert on high lag
    setInterval(async () => {
      const lag = await this.metrics.getGauge('events.processing.lag');
      if (lag > 5000) { // 5 seconds
        await this.alerting.send({
          severity: 'warning',
          message: `Event processing lag is ${lag}ms`,
          context: { lag, threshold: 5000 }
        });
      }
    }, 10000);
  }
}

2. Claude-Powered Anomaly Detection

// monitoring/AnomalyDetector.ts
export class EventAnomalyDetector {
  private claudeClient: ClaudeClient;
  private baseline: EventPatternBaseline;
  
  async detectAnomalies(timeWindow: number = 3600000): Promise<Anomaly[]> {
    const now = new Date();
    const events = await this.eventStore.getEventsInRange({
      start: new Date(now.getTime() - timeWindow),
      end: now
    });
    
    // Get baseline patterns
    const baseline = await this.baseline.getPatterns();
    
    // Ask Claude to identify anomalies
    const analysis = await this.claudeClient.analyze({
      prompt: `Analyze these recent events for anomalies compared to baseline patterns:
        
        Baseline patterns:
        ${JSON.stringify(baseline)}
        
        Recent events:
        ${JSON.stringify(events.map(e => ({
          type: e.eventType,
          timestamp: e.timestamp,
          aggregateId: e.aggregateId
        })))}
        
        Look for:
        1. Unusual event sequences
        2. Abnormal frequency patterns
        3. Missing expected events
        4. Suspicious timing patterns`,
      model: 'claude-3-opus-20240229'
    });
    
    return this.parseAnomalies(analysis);
  }
}

Best Practices

  1. Event Design

    • Make events immutable
    • Include all necessary data
    • Use past tense naming
    • Version events for evolution

  2. Aggregate Design

    • Keep aggregates small
    • One aggregate per transaction
    • Define clear boundaries
    • Use domain language

  3. Projection Design

    • Design for queries
    • Accept eventual consistency
    • Handle idempotency
    • Consider denormalization

  4. Performance

    • Use snapshots for long event streams
    • Implement event archiving
    • Cache frequently accessed data
    • Monitor processing lag

  5. Testing

    • Use Given-When-Then pattern
    • Test event sequences
    • Verify projection consistency
    • Test error scenarios

Common Pitfalls

  1. Large Aggregates: Keep aggregates focused on a single concept
  2. Synchronous Projections: Use async processing for scalability
  3. Missing Events: Always handle unknown event types gracefully
  4. Ignored Versioning: Implement proper concurrency control
  5. Poor Event Design: Include sufficient context in events

References

0 items under this folder.