event-sourcing-cqrs（事件溯源 cqrs）

Event Sourcing & CQRS

Core Concepts

Command ──▶ Command Handler ──▶ Aggregate ──▶ Events ──▶ Event Store
                                                           │
                                                    ┌──────┘
                                                    ▼
                                              Projections ──▶ Read Models ──▶ Queries

Event Sourcing (TypeScript)

Event Store

interface DomainEvent {
  eventId: string;
  aggregateId: string;
  type: string;
  data: unknown;
  version: number;
  timestamp: Date;
}

class EventStore {
  async append(aggregateId: string, events: DomainEvent[], expectedVersion: number): Promise<void> {
    // Optimistic concurrency: check current version matches expected
    const current = await this.getCurrentVersion(aggregateId);
    if (current !== expectedVersion) throw new ConcurrencyError(aggregateId);

    await db.events.createMany({
      data: events.map((e, i) => ({ ...e, version: expectedVersion + i + 1 })),
    });
  }

  async getEvents(aggregateId: string): Promise<DomainEvent[]> {
    return db.events.findMany({
      where: { aggregateId },
      orderBy: { version: 'asc' },
    });
  }
}

Event-Sourced Aggregate

class OrderAggregate {
  private id!: string;
  private status!: string;
  private items: OrderItem[] = [];
  private version = 0;
  private uncommittedEvents: DomainEvent[] = [];

  static fromHistory(events: DomainEvent[]): OrderAggregate {
    const aggregate = new OrderAggregate();
    events.forEach((e) => aggregate.apply(e, false));
    return aggregate;
  }

  createOrder(id: string, items: OrderItem[]) {
    this.apply({ type: 'OrderCreated', data: { id, items } }, true);
  }

  confirmOrder() {
    if (this.status !== 'pending') throw new DomainError('Cannot confirm');
    this.apply({ type: 'OrderConfirmed', data: { id: this.id } }, true);
  }

  private apply(event: Partial<DomainEvent>, isNew: boolean) {
    switch (event.type) {
      case 'OrderCreated':
        this.id = event.data.id;
        this.items = event.data.items;
        this.status = 'pending';
        break;
      case 'OrderConfirmed':
        this.status = 'confirmed';
        break;
    }
    this.version++;
    if (isNew) this.uncommittedEvents.push(event as DomainEvent);
  }
}

CQRS

Command Side

class ConfirmOrderHandler {
  constructor(private eventStore: EventStore) {}

  async handle(cmd: ConfirmOrderCommand): Promise<void> {
    const events = await this.eventStore.getEvents(cmd.orderId);
    const aggregate = OrderAggregate.fromHistory(events);

    aggregate.confirmOrder();

    await this.eventStore.append(
      cmd.orderId,
      aggregate.uncommittedEvents,
      aggregate.version - aggregate.uncommittedEvents.length,
    );
  }
}

Query Side (Projection)

class OrderProjection {
  async handle(event: DomainEvent): Promise<void> {
    switch (event.type) {
      case 'OrderCreated':
        await readDb.orders.create({
          data: { id: event.data.id, status: 'pending', total: calcTotal(event.data.items) },
        });
        break;
      case 'OrderConfirmed':
        await readDb.orders.update({
          where: { id: event.aggregateId },
          data: { status: 'confirmed', confirmedAt: event.timestamp },
        });
        break;
    }
  }
}

Snapshots (for long event streams)

async function loadAggregate(id: string): Promise<OrderAggregate> {
  const snapshot = await snapshotStore.getLatest(id);
  const events = await eventStore.getEvents(id, snapshot?.version ?? 0);

  const aggregate = snapshot
    ? OrderAggregate.fromSnapshot(snapshot)
    : new OrderAggregate();

  events.forEach((e) => aggregate.apply(e, false));
  return aggregate;
}

Anti-Patterns

Production Checklist

• Event store with optimistic concurrency
• Projections for each read model
• Snapshots for aggregates with many events
• Idempotent projection handlers
• Event versioning strategy (upcasting)
• Monitoring: projection lag, event throughput