Event Sourcing and CQRS

# Event Sourcing and CQRS

You are an expert in Event Sourcing and CQRS (Command Query Responsibility Segregation) patterns for building scalable, auditable systems.

## Key Principles
- Store events as the source of truth, not current state
- Separate read and write models for optimization
- Use projections to build read-optimized views
- Handle event versioning and schema evolution
- Implement snapshots for performance

## Event Store Design
```typescript
// Event base types
interface DomainEvent {
  eventId: string;
  aggregateId: string;
  aggregateType: string;
  eventType: string;
  version: number;
  timestamp: Date;
  data: Record<string, unknown>;
  metadata: EventMetadata;
}

interface EventMetadata {
  correlationId: string;
  causationId: string;
  userId?: string;
  schemaVersion: number;
}

// Event Store implementation
class EventStore {
  constructor(private db: Database) {}
  
  async append(
    aggregateId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<void> {
    return this.db.transaction(async (tx) => {
      // Optimistic concurrency check
      const currentVersion = await tx.query(
        `SELECT MAX(version) as version 
         FROM events 
         WHERE aggregate_id = $1`,
        [aggregateId]
      );
      
      if (currentVersion !== expectedVersion) {
        throw new ConcurrencyError(
          `Expected version ${expectedVersion}, got ${currentVersion}`
        );
      }
      
      // Append events
      for (const event of events) {
        await tx.query(
          `INSERT INTO events 
           (event_id, aggregate_id, aggregate_type, event_type, 
            version, timestamp, data, metadata)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
          [
            event.eventId,
            event.aggregateId,
            event.aggregateType,
            event.eventType,
            event.version,
            event.timestamp,
            JSON.stringify(event.data),
            JSON.stringify(event.metadata)
          ]
        );
      }
    });
  }
  
  async getEvents(
    aggregateId: string,
    fromVersion?: number
  ): Promise<DomainEvent[]> {
    const result = await this.db.query(
      `SELECT * FROM events 
       WHERE aggregate_id = $1 
       AND version >= $2
       ORDER BY version ASC`,
      [aggregateId, fromVersion ?? 0]
    );
    return result.rows;
  }
}
```

## Aggregate Implementation
```typescript
// Order aggregate
class Order extends AggregateRoot {
  private status: OrderStatus = "pending";
  private items: OrderItem[] = [];
  private total: number = 0;
  
  // Command handler
  static create(command: CreateOrderCommand): Order {
    const order = new Order();
    order.apply(new OrderCreated({
      orderId: command.orderId,
      customerId: command.customerId,
      items: command.items,
      createdAt: new Date()
    }));
    return order;
  }
  
  ship(command: ShipOrderCommand): void {
    if (this.status !== "confirmed") {
      throw new InvalidOperationError("Order must be confirmed to ship");
    }
    
    this.apply(new OrderShipped({
      orderId: this.id,
      trackingNumber: command.trackingNumber,
      carrier: command.carrier,
      shippedAt: new Date()
    }));
  }
  
  // Event handlers (state reconstruction)
  protected onOrderCreated(event: OrderCreated): void {
    this.id = event.orderId;
    this.items = event.items;
    this.total = event.items.reduce((sum, item) => 
      sum + item.price * item.quantity, 0
    );
    this.status = "pending";
  }
  
  protected onOrderConfirmed(event: OrderConfirmed): void {
    this.status = "confirmed";
  }
  
  protected onOrderShipped(event: OrderShipped): void {
    this.status = "shipped";
  }
}

// Base aggregate
abstract class AggregateRoot {
  protected id: string;
  private version: number = 0;
  private uncommittedEvents: DomainEvent[] = [];
  
  protected apply(event: DomainEvent): void {
    this.version++;
    event.version = this.version;
    this.uncommittedEvents.push(event);
    this.mutate(event);
  }
  
  private mutate(event: DomainEvent): void {
    const handler = `on${event.eventType}`;
    if (typeof this[handler] === "function") {
      this[handler](event);
    }
  }
  
  loadFromHistory(events: DomainEvent[]): void {
    for (const event of events) {
      this.mutate(event);
      this.version = event.version;
    }
  }
  
  getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents];
  }
  
  clearUncommittedEvents(): void {
    this.uncommittedEvents = [];
  }
}
```

## Command and Query Handlers
```typescript
// Command handler
class OrderCommandHandler {
  constructor(
    private eventStore: EventStore,
    private repository: OrderRepository
  ) {}
  
  async handle(command: CreateOrderCommand): Promise<void> {
    const order = Order.create(command);
    
    await this.eventStore.append(
      order.id,
      order.getUncommittedEvents(),
      0  // New aggregate
    );
    
    order.clearUncommittedEvents();
  }
  
  async handleShip(command: ShipOrderCommand): Promise<void> {
    const order = await this.repository.getById(command.orderId);
    
    order.ship(command);
    
    await this.eventStore.append(
      order.id,
      order.getUncommittedEvents(),
      order.version - order.getUncommittedEvents().length
    );
  }
}

// Query side (read model)
class OrderReadModel {
  async getOrderSummary(orderId: string): Promise<OrderSummary> {
    return this.db.query(
      `SELECT * FROM order_summaries WHERE id = $1`,
      [orderId]
    );
  }
  
  async getCustomerOrders(customerId: string): Promise<OrderSummary[]> {
    return this.db.query(
      `SELECT * FROM order_summaries 
       WHERE customer_id = $1 
       ORDER BY created_at DESC`,
      [customerId]
    );
  }
}
```

## Projections
```typescript
// Event projection for read model
class OrderProjection {
  constructor(
    private db: Database,
    private eventStore: EventStore
  ) {}
  
  async project(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case "OrderCreated":
        await this.onOrderCreated(event as OrderCreated);
        break;
      case "OrderShipped":
        await this.onOrderShipped(event as OrderShipped);
        break;
      // ... other events
    }
    
    // Track projection position
    await this.db.query(
      `UPDATE projection_positions 
       SET position = $1 
       WHERE projection_name = $2`,
      [event.version, "order_summary"]
    );
  }
  
  private async onOrderCreated(event: OrderCreated): Promise<void> {
    await this.db.query(
      `INSERT INTO order_summaries 
       (id, customer_id, status, total, item_count, created_at)
       VALUES ($1, $2, $3, $4, $5, $6)`,
      [
        event.orderId,
        event.customerId,
        "pending",
        event.items.reduce((s, i) => s + i.price * i.quantity, 0),
        event.items.length,
        event.createdAt
      ]
    );
  }
  
  private async onOrderShipped(event: OrderShipped): Promise<void> {
    await this.db.query(
      `UPDATE order_summaries 
       SET status = $1, shipped_at = $2, tracking_number = $3
       WHERE id = $4`,
      ["shipped", event.shippedAt, event.trackingNumber, event.orderId]
    );
  }
}
```

## Snapshots
```typescript
class SnapshotRepository {
  async saveSnapshot(aggregate: AggregateRoot): Promise<void> {
    await this.db.query(
      `INSERT INTO snapshots (aggregate_id, version, state, created_at)
       VALUES ($1, $2, $3, $4)
       ON CONFLICT (aggregate_id) 
       DO UPDATE SET version = $2, state = $3, created_at = $4`,
      [aggregate.id, aggregate.version, JSON.stringify(aggregate), new Date()]
    );
  }
  
  async loadAggregate<T extends AggregateRoot>(
    aggregateId: string,
    factory: () => T
  ): Promise<T> {
    const snapshot = await this.db.query(
      `SELECT * FROM snapshots WHERE aggregate_id = $1`,
      [aggregateId]
    );
    
    const aggregate = factory();
    
    if (snapshot) {
      Object.assign(aggregate, JSON.parse(snapshot.state));
      const events = await this.eventStore.getEvents(
        aggregateId, 
        snapshot.version + 1
      );
      aggregate.loadFromHistory(events);
    } else {
      const events = await this.eventStore.getEvents(aggregateId);
      aggregate.loadFromHistory(events);
    }
    
    return aggregate;
  }
}
```

## Best Practices
- Use immutable events with clear naming (past tense)
- Implement idempotent event handlers
- Version events for schema evolution
- Create snapshots for aggregates with many events
- Use eventual consistency for read models
- Test with event replay scenarios