Claude Code事件溯源全实战：CQRS六步交付

graph TB
    subgraph "Write Side 命令端"
        CMD[Command<br/>命令] --> CH[Command Handler<br/>命令处理器]
        CH --> AR[Aggregate Root<br/>聚合根]
        AR --> |产生| EVT[Domain Event<br/>领域事件]
        EVT --> ES[Event Store<br/>事件存储]
    end

    subgraph "Read Side 查询端"
        ES --> |发布| PB[Event Bus<br/>事件总线]
        PB --> PJ[Projection<br/>投影]
        PJ --> RM[Read Model<br/>读模型]
        QRY[Query<br/>查询] --> QH[Query Handler<br/>查询处理器]
        QH --> RM
    end

    subgraph "Cross-Cutting 横切关注点"
        PB --> SG[Saga/Process Manager<br/>流程管理器]
        SG --> CMD
    end

    style ES fill:#e74c3c,color:#fff
    style AR fill:#3498db,color:#fff
    style RM fill:#2ecc71,color:#fff
    style SG fill:#f39c12,color:#fff

mkdir event-sourcing-cqrs && cd event-sourcing-cqrs
npm init -y
npm pkg set type="module"
npm install typescript @types/node pg @types/pg uuid
npm install -D vitest tsx @types/uuid
npx tsc --init

claude
> /init

# Event Sourcing + CQRS 订单系统

## 架构
- Event Sourcing 模式：所有状态变更以事件形式持久化
- CQRS：命令端（写入）和查询端（读取）完全分离
- PostgreSQL 作为 Event Store 和 Read Model 存储

## 项目结构
- src/domain/          - 领域模型（聚合根、值对象、领域事件）
- src/commands/        - 命令定义和命令处理器
- src/queries/         - 查询定义和查询处理器
- src/events/          - 事件定义和事件总线
- src/projections/     - 投影处理器（事件 → 读模型）
- src/sagas/           - Saga/流程管理器
- src/infrastructure/  - 基础设施（Event Store、数据库连接）
- src/api/             - HTTP API 层

## 约定
- 所有事件名使用过去时：OrderCreated, PaymentProcessed
- 命令使用祈使语气：CreateOrder, ProcessPayment
- 聚合根必须实现 apply() 方法处理事件
- 事件不可变，一旦写入不可修改
- 所有异步操作使用 async/await，禁止回调
- 测试使用 vitest

event-sourcing-cqrs/
├── CLAUDE.md
├── tsconfig.json
├── package.json
├── src/
│   ├── domain/
│   │   ├── aggregate-root.ts
│   │   ├── order.aggregate.ts
│   │   └── value-objects.ts
│   ├── events/
│   │   ├── domain-event.ts
│   │   ├── order-events.ts
│   │   └── event-bus.ts
│   ├── commands/
│   │   ├── command.ts
│   │   └── order-commands.handler.ts
│   ├── queries/
│   │   ├── query.ts
│   │   └── order-queries.handler.ts
│   ├── projections/
│   │   └── order-projection.ts
│   ├── sagas/
│   │   └── order-saga.ts
│   ├── infrastructure/
│   │   ├── event-store.ts
│   │   └── db.ts
│   └── api/
│       └── server.ts
└── tests/
    ├── order.aggregate.test.ts
    ├── event-store.test.ts
    └── projection.test.ts

// src/events/domain-event.ts

export interface DomainEvent {
  readonly eventId: string;
  readonly eventType: string;
  readonly aggregateId: string;
  readonly aggregateType: string;
  readonly version: number;
  readonly timestamp: Date;
  readonly payload: Record<string, unknown>;
}

// src/infrastructure/event-store.ts
import { Pool } from 'pg';
import { DomainEvent } from '../events/domain-event.js';

export class EventStore {
  constructor(private pool: Pool) {}

  async init(): Promise<void> {
    await this.pool.query(`
      CREATE TABLE IF NOT EXISTS events (
        event_id       UUID PRIMARY KEY,
        event_type     VARCHAR(255) NOT NULL,
        aggregate_id   UUID NOT NULL,
        aggregate_type VARCHAR(255) NOT NULL,
        version        INTEGER NOT NULL,
        timestamp      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
        payload        JSONB NOT NULL,
        UNIQUE(aggregate_id, version)
      );
      CREATE INDEX IF NOT EXISTS idx_events_aggregate
        ON events(aggregate_id, version);
      CREATE INDEX IF NOT EXISTS idx_events_type
        ON events(event_type);
    `);
  }

  async append(event: DomainEvent): Promise<void> {
    const query = `
      INSERT INTO events
        (event_id, event_type, aggregate_id, aggregate_type,
         version, timestamp, payload)
      VALUES ($1, $2, $3, $4, $5, $6, $7)
    `;
    await this.pool.query(query, [
      event.eventId,
      event.eventType,
      event.aggregateId,
      event.aggregateType,
      event.version,
      event.timestamp,
      JSON.stringify(event.payload),
    ]);
  }

  async appendBatch(events: DomainEvent[]): Promise<void> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');
      for (const event of events) {
        const query = `
          INSERT INTO events
            (event_id, event_type, aggregate_id, aggregate_type,
             version, timestamp, payload)
          VALUES ($1, $2, $3, $4, $5, $6, $7)
        `;
        await client.query(query, [
          event.eventId,
          event.eventType,
          event.aggregateId,
          event.aggregateType,
          event.version,
          event.timestamp,
          JSON.stringify(event.payload),
        ]);
      }
      await client.query('COMMIT');
    } catch (err) {
      await client.query('ROLLBACK');
      throw err;
    } finally {
      client.release();
    }
  }

  async loadEvents(
    aggregateId: string,
    fromVersion?: number
  ): Promise<DomainEvent[]> {
    const query = fromVersion
      ? `SELECT * FROM events
         WHERE aggregate_id = $1 AND version > $2
         ORDER BY version ASC`
      : `SELECT * FROM events
         WHERE aggregate_id = $1
         ORDER BY version ASC`;

    const params = fromVersion
      ? [aggregateId, fromVersion]
      : [aggregateId];

    const result = await this.pool.query(query, params);
    return result.rows.map(this.rowToEvent);
  }

  async loadEventsByType(
    eventType: string,
    afterTimestamp?: Date
  ): Promise<DomainEvent[]> {
    const query = afterTimestamp
      ? `SELECT * FROM events
         WHERE event_type = $1 AND timestamp > $2
         ORDER BY timestamp ASC`
      : `SELECT * FROM events
         WHERE event_type = $1
         ORDER BY timestamp ASC`;

    const params = afterTimestamp
      ? [eventType, afterTimestamp]
      : [eventType];

    const result = await this.pool.query(query, params);
    return result.rows.map(this.rowToEvent);
  }

  async getLatestVersion(aggregateId: string): Promise<number> {
    const result = await this.pool.query(
      `SELECT MAX(version) as version
       FROM events WHERE aggregate_id = $1`,
      [aggregateId]
    );
    return result.rows[0]?.version ?? 0;
  }

  private rowToEvent(row: Record<string, unknown>): DomainEvent {
    return {
      eventId: row.event_id as string,
      eventType: row.event_type as string,
      aggregateId: row.aggregate_id as string,
      aggregateType: row.aggregate_type as string,
      version: row.version as number,
      timestamp: row.timestamp as Date,
      payload: typeof row.payload === 'string'
        ? JSON.parse(row.payload)
        : (row.payload as Record<string, unknown>),
    };
  }
}

// src/events/event-bus.ts
import { DomainEvent } from './domain-event.js';

export type EventHandler = (event: DomainEvent) => Promise<void>;

export class EventBus {
  private handlers = new Map<string, EventHandler[]>();
  private wildcardHandlers: EventHandler[] = [];

  subscribe(eventType: string, handler: EventHandler): void {
    const handlers = this.handlers.get(eventType) ?? [];
    handlers.push(handler);
    this.handlers.set(eventType, handlers);
  }

  subscribeAll(handler: EventHandler): void {
    this.wildcardHandlers.push(handler);
  }

  async publish(event: DomainEvent): Promise<void> {
    const typedHandlers = this.handlers.get(event.eventType) ?? [];
    const allHandlers = [...typedHandlers, ...this.wildcardHandlers];

    await Promise.all(
      allHandlers.map(handler =>
        handler(event).catch(err =>
          console.error(
            `Event handler failed for ${event.eventType}:`,
            err
          )
        )
      )
    );
  }

  async publishBatch(events: DomainEvent[]): Promise<void> {
    for (const event of events) {
      await this.publish(event);
    }
  }
}

请检查 src/infrastructure/event-store.ts，确保：
1. 所有数据库操作都使用参数化查询（防SQL注入）
2. 事件写入使用事务保证原子性
3. 版本号使用乐观并发控制
4. 添加合适的索引

// src/domain/aggregate-root.ts
import { v4 as uuidv4 } from 'uuid';
import { DomainEvent } from '../events/domain-event.js';

export abstract class AggregateRoot {
  public readonly id: string;
  private uncommittedEvents: DomainEvent[] = [];
  private _version = 0;

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

  get version(): number {
    return this._version;
  }

  getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents];
  }

  clearUncommittedEvents(): void {
    this.uncommittedEvents = [];
  }

  protected applyEvent(event: DomainEvent): void {
    this._version++;
    this.uncommittedEvents.push(event);
    this.onEvent(event);
  }

  protected abstract onEvent(event: DomainEvent): void;

  protected createEvent(
    eventType: string,
    payload: Record<string, unknown>
  ): DomainEvent {
    return {
      eventId: uuidv4(),
      eventType,
      aggregateId: this.id,
      aggregateType: this.constructor.name,
      version: this._version + 1,
      timestamp: new Date(),
      payload,
    };
  }

  static rehydrate<T extends AggregateRoot>(
    this: new (id: string) => T,
    id: string,
    events: DomainEvent[]
  ): T {
    const aggregate = new this(id);
    for (const event of events) {
      aggregate._version = event.version;
      aggregate.onEvent(event);
    }
    return aggregate;
  }
}

// src/domain/order.aggregate.ts
import { AggregateRoot } from './aggregate-root.js';
import { DomainEvent } from '../events/domain-event.js';

export enum OrderStatus {
  PENDING = 'PENDING',
  CONFIRMED = 'CONFIRMED',
  PAID = 'PAID',
  SHIPPED = 'SHIPPED',
  DELIVERED = 'DELIVERED',
  CANCELLED = 'CANCELLED',
}

export interface OrderItem {
  productId: string;
  productName: string;
  quantity: number;
  unitPrice: number;
}

export class OrderAggregate extends AggregateRoot {
  private _customerId!: string;
  private _items: OrderItem[] = [];
  private _status: OrderStatus = OrderStatus.PENDING;
  private _totalAmount = 0;
  private _cancelReason?: string;

  get customerId(): string { return this._customerId; }
  get items(): OrderItem[] { return [...this._items]; }
  get status(): OrderStatus { return this._status; }
  get totalAmount(): number { return this._totalAmount; }
  get cancelReason(): string | undefined { return this._cancelReason; }

  // ─── 命令方法 ───

  create(customerId: string, items: OrderItem[]): void {
    if (this._status !== OrderStatus.PENDING) {
      throw new Error('Order already created');
    }
    if (items.length === 0) {
      throw new Error('Order must have at least one item');
    }
    const totalAmount = items.reduce(
      (sum, item) => sum + item.unitPrice * item.quantity, 0
    );
    const event = this.createEvent('OrderCreated', {
      customerId,
      items,
      totalAmount,
    });
    this.applyEvent(event);
  }

  confirm(): void {
    if (this._status !== OrderStatus.PENDING) {
      throw new Error(`Cannot confirm order in ${this._status} status`);
    }
    this.applyEvent(this.createEvent('OrderConfirmed', {}));
  }

  pay(paymentId: string): void {
    if (this._status !== OrderStatus.CONFIRMED) {
      throw new Error(`Cannot pay order in ${this._status} status`);
    }
    this.applyEvent(this.createEvent('OrderPaid', { paymentId }));
  }

  ship(trackingNumber: string): void {
    if (this._status !== OrderStatus.PAID) {
      throw new Error(`Cannot ship order in ${this._status} status`);
    }
    this.applyEvent(this.createEvent('OrderShipped', { trackingNumber }));
  }

  deliver(): void {
    if (this._status !== OrderStatus.SHIPPED) {
      throw new Error(`Cannot deliver order in ${this._status} status`);
    }
    this.applyEvent(this.createEvent('OrderDelivered', {}));
  }

  cancel(reason: string): void {
    if (this._status === OrderStatus.SHIPPED ||
        this._status === OrderStatus.DELIVERED) {
      throw new Error('Cannot cancel shipped/delivered order');
    }
    this.applyEvent(this.createEvent('OrderCancelled', { reason }));
  }

  // ─── 事件处理 ───

  protected onEvent(event: DomainEvent): void {
    switch (event.eventType) {
      case 'OrderCreated':
        this._customerId = event.payload.customerId as string;
        this._items = event.payload.items as OrderItem[];
        this._totalAmount = event.payload.totalAmount as number;
        this._status = OrderStatus.PENDING;
        break;
      case 'OrderConfirmed':
        this._status = OrderStatus.CONFIRMED;
        break;
      case 'OrderPaid':
        this._status = OrderStatus.PAID;
        break;
      case 'OrderShipped':
        this._status = OrderStatus.SHIPPED;
        break;
      case 'OrderDelivered':
        this._status = OrderStatus.DELIVERED;
        break;
      case 'OrderCancelled':
        this._status = OrderStatus.CANCELLED;
        this._cancelReason = event.payload.reason as string;
        break;
    }
  }
}

请为 src/domain/order.aggregate.ts 编写完整的单元测试。
覆盖以下场景：
1. 创建订单 - 正常流程
2. 创建订单 - 空商品列表应抛出错误
3. 订单状态流转：Created -> Confirmed -> Paid -> Shipped -> Delivered
4. 重复状态转换应抛出错误
5. 已发货订单不能取消
6. 事件溯源重建 - 从事件重建聚合根状态
7. 验证所有事件都有正确的版本号递增

// tests/order.aggregate.test.ts
import { describe, it, expect } from 'vitest';
import { OrderAggregate, OrderStatus } from '../src/domain/order.aggregate.js';
import { AggregateRoot } from '../src/domain/aggregate-root.js';

describe('OrderAggregate', () => {
  const sampleItems = [
    { productId: 'p1', productName: 'Widget', quantity: 2, unitPrice: 10 },
    { productId: 'p2', productName: 'Gadget', quantity: 1, unitPrice: 25 },
  ];

  describe('create', () => {
    it('should create order with correct properties', () => {
      const order = new OrderAggregate();
      order.create('customer-1', sampleItems);

      expect(order.customerId).toBe('customer-1');
      expect(order.items).toEqual(sampleItems);
      expect(order.totalAmount).toBe(45); // 2*10 + 1*25
      expect(order.status).toBe(OrderStatus.PENDING);
      expect(order.version).toBe(1);
    });

    it('should throw for empty items', () => {
      const order = new OrderAggregate();
      expect(() => order.create('customer-1', [])).toThrow(
        'Order must have at least one item'
      );
    });

    it('should throw when creating twice', () => {
      const order = new OrderAggregate();
      order.create('customer-1', sampleItems);
      expect(() => order.create('customer-1', sampleItems)).toThrow(
        'Order already created'
      );
    });
  });

  describe('state transitions', () => {
    it('should flow through full lifecycle', () => {
      const order = new OrderAggregate();
      order.create('customer-1', sampleItems);
      expect(order.status).toBe(OrderStatus.PENDING);

      order.confirm();
      expect(order.status).toBe(OrderStatus.CONFIRMED);

      order.pay('pay-123');
      expect(order.status).toBe(OrderStatus.PAID);

      order.ship('TRACK-456');
      expect(order.status).toBe(OrderStatus.SHIPPED);

      order.deliver();
      expect(order.status).toBe(OrderStatus.DELIVERED);
      expect(order.version).toBe(5);
    });

    it('should reject skipping states', () => {
      const order = new OrderAggregate();
      order.create('customer-1', sampleItems);

      expect(() => order.pay('pay-123')).toThrow(
        'Cannot pay order in PENDING status'
      );
    });
  });

  describe('cancel', () => {
    it('should cancel pending order', () => {
      const order = new OrderAggregate();
      order.create('customer-1', sampleItems);
      order.cancel('changed mind');

      expect(order.status).toBe(OrderStatus.CANCELLED);
      expect(order.cancelReason).toBe('changed mind');
    });

    it('should not cancel shipped order', () => {
      const order = new OrderAggregate();
      order.create('customer-1', sampleItems);
      order.confirm();
      order.pay('pay-123');
      order.ship('TRACK-456');

      expect(() => order.cancel('too late')).toThrow(
        'Cannot cancel shipped/delivered order'
      );
    });
  });

  describe('event sourcing rehydration', () => {
    it('should rebuild state from events', () => {
      const original = new OrderAggregate();
      original.create('customer-1', sampleItems);
      original.confirm();
      original.pay('pay-123');

      const events = original.getUncommittedEvents();

      const rebuilt = AggregateRoot.rehydrate.call(
        OrderAggregate,
        original.id,
        events
      );

      expect(rebuilt.id).toBe(original.id);
      expect(rebuilt.status).toBe(OrderStatus.PAID);
      expect(rebuilt.customerId).toBe('customer-1');
      expect(rebuilt.totalAmount).toBe(45);
      expect(rebuilt.version).toBe(3);
      expect(rebuilt.getUncommittedEvents()).toHaveLength(0);
    });
  });

  describe('event versioning', () => {
    it('should increment version for each event', () => {
      const order = new OrderAggregate();
      order.create('customer-1', sampleItems);
      order.confirm();
      order.pay('pay-123');

      const events = order.getUncommittedEvents();
      expect(events.map(e => e.version)).toEqual([1, 2, 3]);
    });
  });
});

npx vitest run tests/order.aggregate.test.ts

// src/commands/command.ts

export interface Command {
  readonly commandId: string;
  readonly commandType: string;
  readonly aggregateId?: string;
  readonly payload: Record<string, unknown>;
}

// src/commands/order-commands.ts
import { v4 as uuidv4 } from 'uuid';
import { Command } from './command.js';
import { OrderItem } from '../domain/order.aggregate.js';

export class CreateOrderCommand implements Command {
  readonly commandId = uuidv4();
  readonly commandType = 'CreateOrder';
  constructor(
    readonly aggregateId: string,
    readonly payload: {
      customerId: string;
      items: OrderItem[];
    }
  ) {}
}

export class ConfirmOrderCommand implements Command {
  readonly commandId = uuidv4();
  readonly commandType = 'ConfirmOrder';
  constructor(
    readonly aggregateId: string,
    readonly payload: Record<string, unknown> = {}
  ) {}
}

export class PayOrderCommand implements Command {
  readonly commandId = uuidv4();
  readonly commandType = 'PayOrder';
  constructor(
    readonly aggregateId: string,
    readonly payload: { paymentId: string }
  ) {}
}

export class ShipOrderCommand implements Command {
  readonly commandId = uuidv4();
  readonly commandType = 'ShipOrder';
  constructor(
    readonly aggregateId: string,
    readonly payload: { trackingNumber: string }
  ) {}
}

export class CancelOrderCommand implements Command {
  readonly commandId = uuidv4();
  readonly commandType = 'CancelOrder';
  constructor(
    readonly aggregateId: string,
    readonly payload: { reason: string }
  ) {}
}

// src/commands/order-commands.handler.ts
import { EventStore } from '../infrastructure/event-store.js';
import { EventBus } from '../events/event-bus.js';
import { OrderAggregate } from '../domain/order.aggregate.js';
import { AggregateRoot } from '../domain/aggregate-root.js';
import { Command } from './command.js';

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

  async handle(command: Command): Promise<void> {
    switch (command.commandType) {
      case 'CreateOrder':
        return this.handleCreate(command);
      case 'ConfirmOrder':
        return this.handleConfirm(command);
      case 'PayOrder':
        return this.handlePay(command);
      case 'ShipOrder':
        return this.handleShip(command);
      case 'CancelOrder':
        return this.handleCancel(command);
      default:
        throw new Error(`Unknown command: ${command.commandType}`);
    }
  }

  private async loadAggregate(id: string): Promise<OrderAggregate> {
    const events = await this.eventStore.loadEvents(id);
    if (events.length === 0) {
      return new OrderAggregate(id);
    }
    return AggregateRoot.rehydrate.call(
      OrderAggregate,
      id,
      events
    );
  }

  private async saveAndPublish(aggregate: OrderAggregate): Promise<void> {
    const events = aggregate.getUncommittedEvents();
    if (events.length === 0) return;

    await this.eventStore.appendBatch(events);
    await this.eventBus.publishBatch(events);
    aggregate.clearUncommittedEvents();
  }

  private async handleCreate(command: Command): Promise<void> {
    const order = new OrderAggregate(command.aggregateId);
    order.create(
      command.payload.customerId as string,
      command.payload.items as any[]
    );
    await this.saveAndPublish(order);
  }

  private async handleConfirm(command: Command): Promise<void> {
    const order = await this.loadAggregate(command.aggregateId!);
    order.confirm();
    await this.saveAndPublish(order);
  }

  private async handlePay(command: Command): Promise<void> {
    const order = await this.loadAggregate(command.aggregateId!);
    order.pay(command.payload.paymentId as string);
    await this.saveAndPublish(order);
  }

  private async handleShip(command: Command): Promise<void> {
    const order = await this.loadAggregate(command.aggregateId!);
    order.ship(command.payload.trackingNumber as string);
    await this.saveAndPublish(order);
  }

  private async handleCancel(command: Command): Promise<void> {
    const order = await this.loadAggregate(command.aggregateId!);
    order.cancel(command.payload.reason as string);
    await this.saveAndPublish(order);
  }
}

请审查 OrderCommandHandler，重点检查：
1. 并发冲突处理——两个命令同时修改同一聚合根怎么办？
2. loadAggregate 的性能——是否需要快照优化？
3. 事务边界——事件写入和发布是否在同一事务内？

// src/infrastructure/read-db.ts
import { Pool } from 'pg';

export async function initReadModel(pool: Pool): Promise<void> {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS order_summary (
      order_id        UUID PRIMARY KEY,
      customer_id     UUID NOT NULL,
      status          VARCHAR(50) NOT NULL DEFAULT 'PENDING',
      total_amount    DECIMAL(10, 2) NOT NULL DEFAULT 0,
      item_count      INTEGER NOT NULL DEFAULT 0,
      tracking_number VARCHAR(255),
      cancel_reason   TEXT,
      created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
      updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
    );

    CREATE TABLE IF NOT EXISTS order_items (
      id          SERIAL PRIMARY KEY,
      order_id    UUID NOT NULL REFERENCES order_summary(order_id),
      product_id  UUID NOT NULL,
      product_name VARCHAR(255) NOT NULL,
      quantity    INTEGER NOT NULL,
      unit_price  DECIMAL(10, 2) NOT NULL
    );

    CREATE INDEX IF NOT EXISTS idx_order_summary_customer
      ON order_summary(customer_id);
    CREATE INDEX IF NOT EXISTS idx_order_summary_status
      ON order_summary(status);
  `);
}

// src/projections/order-projection.ts
import { Pool } from 'pg';
import { DomainEvent } from '../events/domain-event.js';

export class OrderProjection {
  constructor(private pool: Pool) {}

  async handle(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderCreated':
        return this.onOrderCreated(event);
      case 'OrderConfirmed':
        return this.onStatusChange(event, 'CONFIRMED');
      case 'OrderPaid':
        return this.onStatusChange(event, 'PAID');
      case 'OrderShipped':
        return this.onOrderShipped(event);
      case 'OrderDelivered':
        return this.onStatusChange(event, 'DELIVERED');
      case 'OrderCancelled':
        return this.onOrderCancelled(event);
    }
  }

  private async onOrderCreated(event: DomainEvent): Promise<void> {
    const { customerId, items, totalAmount } = event.payload;
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');

      await client.query(
        `INSERT INTO order_summary
          (order_id, customer_id, status, total_amount, item_count)
         VALUES ($1, $2, 'PENDING', $3, $4)`,
        [
          event.aggregateId,
          customerId,
          totalAmount,
          (items as any[]).length,
        ]
      );

      for (const item of items as any[]) {
        await client.query(
          `INSERT INTO order_items
            (order_id, product_id, product_name, quantity, unit_price)
           VALUES ($1, $2, $3, $4, $5)`,
          [
            event.aggregateId,
            item.productId,
            item.productName,
            item.quantity,
            item.unitPrice,
          ]
        );
      }

      await client.query('COMMIT');
    } catch (err) {
      await client.query('ROLLBACK');
      throw err;
    } finally {
      client.release();
    }
  }

  private async onStatusChange(
    event: DomainEvent,
    status: string
  ): Promise<void> {
    await this.pool.query(
      `UPDATE order_summary
       SET status = $1, updated_at = NOW()
       WHERE order_id = $2`,
      [status, event.aggregateId]
    );
  }

  private async onOrderShipped(event: DomainEvent): Promise<void> {
    await this.pool.query(
      `UPDATE order_summary
       SET status = 'SHIPPED',
           tracking_number = $1,
           updated_at = NOW()
       WHERE order_id = $2`,
      [event.payload.trackingNumber, event.aggregateId]
    );
  }

  private async onOrderCancelled(event: DomainEvent): Promise<void> {
    await this.pool.query(
      `UPDATE order_summary
       SET status = 'CANCELLED',
           cancel_reason = $1,
           updated_at = NOW()
       WHERE order_id = $2`,
      [event.payload.reason, event.aggregateId]
    );
  }
}

// src/queries/order-queries.handler.ts
import { Pool } from 'pg';

export interface OrderSummaryView {
  orderId: string;
  customerId: string;
  status: string;
  totalAmount: number;
  itemCount: number;
  trackingNumber: string | null;
  cancelReason: string | null;
  createdAt: Date;
  updatedAt: Date;
}

export class OrderQueryHandler {
  constructor(private pool: Pool) {}

  async getOrderById(orderId: string): Promise<OrderSummaryView | null> {
    const result = await this.pool.query(
      `SELECT * FROM order_summary WHERE order_id = $1`,
      [orderId]
    );
    if (result.rows.length === 0) return null;
    return this.mapRow(result.rows[0]);
  }

  async getOrdersByCustomer(
    customerId: string,
    limit = 20,
    offset = 0
  ): Promise<OrderSummaryView[]> {
    const result = await this.pool.query(
      `SELECT * FROM order_summary
       WHERE customer_id = $1
       ORDER BY created_at DESC
       LIMIT $2 OFFSET $3`,
      [customerId, limit, offset]
    );
    return result.rows.map(this.mapRow);
  }

  async getOrdersByStatus(
    status: string,
    limit = 50
  ): Promise<OrderSummaryView[]> {
    const result = await this.pool.query(
      `SELECT * FROM order_summary
       WHERE status = $1
       ORDER BY updated_at DESC
       LIMIT $2`,
      [status, limit]
    );
    return result.rows.map(this.mapRow);
  }

  async getOrderHistory(orderId: string): Promise<{
    summary: OrderSummaryView | null;
    events: Array<{ eventType: string; timestamp: Date; version: number }>;
  }> {
    const summary = await this.getOrderById(orderId);
    if (!summary) return { summary: null, events: [] };

    const result = await this.pool.query(
      `SELECT event_type, timestamp, version
       FROM events
       WHERE aggregate_id = $1
       ORDER BY version ASC`,
      [orderId]
    );
    return {
      summary,
      events: result.rows.map(r => ({
        eventType: r.event_type,
        timestamp: r.timestamp,
        version: r.version,
      })),
    };
  }

  private mapRow(row: Record<string, unknown>): OrderSummaryView {
    return {
      orderId: row.order_id as string,
      customerId: row.customer_id as string,
      status: row.status as string,
      totalAmount: Number(row.total_amount),
      itemCount: row.item_count as number,
      trackingNumber: (row.tracking_number as string) ?? null,
      cancelReason: (row.cancel_reason as string) ?? null,
      createdAt: row.created_at as Date,
      updatedAt: row.updated_at as Date,
    };
  }
}

sequenceDiagram
    participant CMD as Command
    participant Saga as OrderSaga
    participant Order as Order Aggregate
    participant Payment as Payment Service
    participant Inventory as Inventory Service

    CMD->>Saga: CreateOrder
    Saga->>Order: OrderCreated
    Saga->>Payment: ProcessPayment
    alt Payment Success
        Payment->>Saga: PaymentSucceeded
        Saga->>Order: ConfirmOrder
        Saga->>Inventory: ReserveStock
    else Payment Failed
        Payment->>Saga: PaymentFailed
        Saga->>Order: CancelOrder
    end

// src/sagas/order-saga.ts
import { DomainEvent } from '../events/domain-event.js';
import { Command } from '../commands/command.js';
import { v4 as uuidv4 } from 'uuid';

export class OrderSaga {
  private sagaState = new Map<string, {
    orderId: string;
    customerId: string;
    totalAmount: number;
    paymentId?: string;
    step: 'AWAITING_PAYMENT' | 'AWAITING_INVENTORY' | 'COMPLETED';
  }>();

  handle(event: DomainEvent): Command[] {
    switch (event.eventType) {
      case 'OrderCreated':
        return this.onOrderCreated(event);
      case 'PaymentProcessed':
        return this.onPaymentProcessed(event);
      case 'InventoryReserved':
        return this.onInventoryReserved(event);
      case 'PaymentFailed':
        return this.onPaymentFailed(event);
      case 'InventoryReservationFailed':
        return this.onInventoryReservationFailed(event);
      default:
        return [];
    }
  }

  private onOrderCreated(event: DomainEvent): Command[] {
    this.sagaState.set(event.aggregateId, {
      orderId: event.aggregateId,
      customerId: event.payload.customerId as string,
      totalAmount: event.payload.totalAmount as number,
      step: 'AWAITING_PAYMENT',
    });

    // 发起支付命令
    return [{
      commandId: uuidv4(),
      commandType: 'ProcessPayment',
      aggregateId: event.aggregateId,
      payload: {
        orderId: event.aggregateId,
        amount: event.payload.totalAmount,
        customerId: event.payload.customerId,
      },
    }];
  }

  private onPaymentProcessed(event: DomainEvent): Command[] {
    const state = this.sagaState.get(event.aggregateId);
    if (!state || state.step !== 'AWAITING_PAYMENT') return [];

    state.paymentId = event.payload.paymentId as string;
    state.step = 'AWAITING_INVENTORY';

    // 支付成功，确认订单并预留库存
    return [
      {
        commandId: uuidv4(),
        commandType: 'ConfirmOrder',
        aggregateId: event.aggregateId,
        payload: {},
      },
      {
        commandId: uuidv4(),
        commandType: 'ReserveInventory',
        aggregateId: event.aggregateId,
        payload: {
          orderId: event.aggregateId,
          items: event.payload.items,
        },
      },
    ];
  }

  private onInventoryReserved(_event: DomainEvent): Command[] {
    // 库存预留成功，Saga 完成
    return [];
  }

  private onPaymentFailed(event: DomainEvent): Command[] {
    // 补偿事务：取消订单
    return [{
      commandId: uuidv4(),
      commandType: 'CancelOrder',
      aggregateId: event.aggregateId,
      payload: { reason: 'Payment failed' },
    }];
  }

  private onInventoryReservationFailed(event: DomainEvent): Command[] {
    // 补偿事务：取消订单 + 退款
    return [
      {
        commandId: uuidv4(),
        commandType: 'CancelOrder',
        aggregateId: event.aggregateId,
        payload: { reason: 'Out of stock' },
      },
      {
        commandId: uuidv4(),
        commandType: 'RefundPayment',
        aggregateId: event.aggregateId,
        payload: { orderId: event.aggregateId },
      },
    ];
  }
}

// src/api/server.ts
import express from 'express';
import { Pool } from 'pg';
import { EventStore } from '../infrastructure/event-store.js';
import { EventBus } from '../events/event-bus.js';
import { OrderCommandHandler } from '../commands/order-commands.handler.js';
import { OrderQueryHandler } from '../queries/order-queries.handler.js';
import { OrderProjection } from '../projections/order-projection.js';
import { OrderSaga } from '../sagas/order-saga.js';
import { initReadModel } from '../infrastructure/read-db.js';

async function bootstrap() {
  const pool = new Pool({
    connectionString: process.env.DATABASE_URL
      ?? 'postgresql://localhost:5432/eventstore',
  });

  // 初始化基础设施
  const eventStore = new EventStore(pool);
  await eventStore.init();
  await initReadModel(pool);

  // 初始化事件总线
  const eventBus = new EventBus();

  // 注册投影
  const projection = new OrderProjection(pool);
  eventBus.subscribeAll(event => projection.handle(event));

  // 注册 Saga
  const saga = new OrderSaga();
  const commandHandler = new OrderCommandHandler(eventStore, eventBus);
  eventBus.subscribeAll(async event => {
    const commands = saga.handle(event);
    for (const cmd of commands) {
      await commandHandler.handle(cmd);
    }
  });

  // 初始化处理器
  const queryHandler = new OrderQueryHandler(pool);

  // 创建 Express 应用
  const app = express();
  app.use(express.json());

  // ─── 命令端 API ───

  app.post('/orders', async (req, res) => {
    try {
      const { customerId, items } = req.body;
      const orderId = crypto.randomUUID();
      await commandHandler.handle({
        commandId: crypto.randomUUID(),
        commandType: 'CreateOrder',
        aggregateId: orderId,
        payload: { customerId, items },
      });
      res.status(201).json({ orderId });
    } catch (err: any) {
      res.status(400).json({ error: err.message });
    }
  });

  app.post('/orders/:id/confirm', async (req, res) => {
    try {
      await commandHandler.handle({
        commandId: crypto.randomUUID(),
        commandType: 'ConfirmOrder',
        aggregateId: req.params.id,
        payload: {},
      });
      res.json({ status: 'confirmed' });
    } catch (err: any) {
      res.status(400).json({ error: err.message });
    }
  });

  app.post('/orders/:id/pay', async (req, res) => {
    try {
      const { paymentId } = req.body;
      await commandHandler.handle({
        commandId: crypto.randomUUID(),
        commandType: 'PayOrder',
        aggregateId: req.params.id,
        payload: { paymentId },
      });
      res.json({ status: 'paid' });
    } catch (err: any) {
      res.status(400).json({ error: err.message });
    }
  });

  app.post('/orders/:id/ship', async (req, res) => {
    try {
      const { trackingNumber } = req.body;
      await commandHandler.handle({
        commandId: crypto.randomUUID(),
        commandType: 'ShipOrder',
        aggregateId: req.params.id,
        payload: { trackingNumber },
      });
      res.json({ status: 'shipped' });
    } catch (err: any) {
      res.status(400).json({ error: err.message });
    }
  });

  app.post('/orders/:id/cancel', async (req, res) => {
    try {
      const { reason } = req.body;
      await commandHandler.handle({
        commandId: crypto.randomUUID(),
        commandType: 'CancelOrder',
        aggregateId: req.params.id,
        payload: { reason },
      });
      res.json({ status: 'cancelled' });
    } catch (err: any) {
      res.status(400).json({ error: err.message });
    }
  });

  // ─── 查询端 API ───

  app.get('/orders/:id', async (req, res) => {
    const order = await queryHandler.getOrderById(req.params.id);
    if (!order) return res.status(404).json({ error: 'Not found' });
    res.json(order);
  });

  app.get('/orders/:id/history', async (req, res) => {
    const history = await queryHandler.getOrderHistory(req.params.id);
    if (!history.summary) {
      return res.status(404).json({ error: 'Not found' });
    }
    res.json(history);
  });

  app.get('/customers/:id/orders', async (req, res) => {
    const limit = Number(req.query.limit) || 20;
    const offset = Number(req.query.offset) || 0;
    const orders = await queryHandler.getOrdersByCustomer(
      req.params.id, limit, offset
    );
    res.json(orders);
  });

  app.get('/orders', async (req, res) => {
    const status = req.query.status as string;
    if (!status) {
      return res.status(400).json({ error: 'status required' });
    }
    const orders = await queryHandler.getOrdersByStatus(status);
    res.json(orders);
  });

  const port = Number(process.env.PORT) || 3000;
  app.listen(port, () => {
    console.log(`Order service running on :${port}`);
  });
}

bootstrap().catch(console.error);

# 启动服务
DATABASE_URL=postgresql://localhost:5432/eventstore npx tsx src/api/server.ts

# 创建订单
curl -X POST http://localhost:3000/orders \
  -H 'Content-Type: application/json' \
  -d '{
    "customerId": "cust-001",
    "items": [
      {"productId": "p1", "productName": "Widget", "quantity": 2, "unitPrice": 10},
      {"productId": "p2", "productName": "Gadget", "quantity": 1, "unitPrice": 25}
    ]
  }'

# 确认订单
curl -X POST http://localhost:3000/orders/{orderId}/confirm

# 支付
curl -X POST http://localhost:3000/orders/{orderId}/pay \
  -H 'Content-Type: application/json' \
  -d '{"paymentId": "pay-001"}'

# 查看订单
curl http://localhost:3000/orders/{orderId}

# 查看完整事件历史
curl http://localhost:3000/orders/{orderId}/history