设计and Implementing Event-Driven Architecture：Claude Code 实战指南

事件駆動架构とは

事件駆動架构（EDA）は、事件の発行と購読通过システムを疎結合に设计するパターンです。借助 Claude Code，EDAの设计から实现まで高效地進められます。

类型安全な事件バス

type EventMap = {
  "user.created": { userId: string; email: string };
  "user.updated": { userId: string; changes: Record<string, unknown> };
  "order.created": { orderId: string; userId: string; total: number };
  "order.completed": { orderId: string; completedAt: Date };
  "order.cancelled": { orderId: string; reason: string };
  "payment.processed": { paymentId: string; amount: number };
};

type EventName = keyof EventMap;
type EventHandler<T extends EventName> = (payload: EventMap[T]) => Promise<void>;

class EventBus {
  private handlers = new Map<string, Set<Function>>();

  on<T extends EventName>(event: T, handler: EventHandler<T>) {
    if (!this.handlers.has(event)) {
      this.handlers.set(event, new Set());
    }
    this.handlers.get(event)!.add(handler);

    // unsubscribe函数を返す
    return () => {
      this.handlers.get(event)?.delete(handler);
    };
  }

  async emit<T extends EventName>(event: T, payload: EventMap[T]) {
    const handlers = this.handlers.get(event);
    if (!handlers) return;

    const results = await Promise.allSettled(
      Array.from(handlers).map((handler) => handler(payload))
    );

    const failures = results.filter((r) => r.status === "rejected");
    if (failures.length > 0) {
      console.error(
        `${failures.length} handlers failed for ${event}:`,
        failures
      );
    }
  }
}

const eventBus = new EventBus();

事件处理程序の注册

// 用户创建時の処理
eventBus.on("user.created", async ({ userId, email }) => {
  // ウェルカムメール发送
  await emailQueue.add("send", {
    to: email,
    subject: "ようこそ！",
    template: "welcome",
    data: { userId },
  });
});

eventBus.on("user.created", async ({ userId }) => {
  // デフォルト配置の创建
  await prisma.userSettings.create({
    data: {
      userId,
      theme: "light",
      language: "ja",
      notifications: true,
    },
  });
});

// 注文完成時の処理
eventBus.on("order.completed", async ({ orderId }) => {
  const order = await prisma.order.findUnique({
    where: { id: orderId },
    include: { user: true, items: true },
  });

  if (!order) return;

  // 在庫の更新
  for (const item of order.items) {
    await prisma.product.update({
      where: { id: item.productId },
      data: { stock: { decrement: item.quantity } },
    });
  }
});

ドメイン事件パターン

abstract class DomainEvent {
  readonly occurredAt: Date;
  readonly eventId: string;

  constructor() {
    this.occurredAt = new Date();
    this.eventId = crypto.randomUUID();
  }
}

class OrderCreatedEvent extends DomainEvent {
  constructor(
    public readonly orderId: string,
    public readonly userId: string,
    public readonly items: Array<{ productId: string; quantity: number }>,
    public readonly total: number
  ) {
    super();
  }
}

// Aggregate Root
class Order {
  private domainEvents: DomainEvent[] = [];

  static create(params: {
    userId: string;
    items: Array<{ productId: string; quantity: number; price: number }>;
  }): Order {
    const order = new Order();
    const orderId = crypto.randomUUID();
    const total = params.items.reduce(
      (sum, item) => sum + item.price * item.quantity,
      0
    );

    order.domainEvents.push(
      new OrderCreatedEvent(orderId, params.userId, params.items, total)
    );

    return order;
  }

  pullDomainEvents(): DomainEvent[] {
    const events = [...this.domainEvents];
    this.domainEvents = [];
    return events;
  }
}

CQRS パターン

// Command側（写入）
interface Command<T = void> {
  execute(): Promise<T>;
}

class CreateOrderCommand implements Command<string> {
  constructor(
    private userId: string,
    private items: Array<{ productId: string; quantity: number }>
  ) {}

  async execute(): Promise<string> {
    // 验证
    for (const item of this.items) {
      const product = await prisma.product.findUnique({
        where: { id: item.productId },
      });
      if (!product || product.stock < item.quantity) {
        throw new Error(`Insufficient stock: ${item.productId}`);
      }
    }

    // 注文创建
    const order = await prisma.order.create({
      data: {
        userId: this.userId,
        status: "pending",
        items: {
          create: this.items.map((item) => ({
            productId: item.productId,
            quantity: item.quantity,
          })),
        },
      },
    });

    // 事件発行
    await eventBus.emit("order.created", {
      orderId: order.id,
      userId: this.userId,
      total: 0,
    });

    return order.id;
  }
}

// Query側（読み取り）
interface Query<T> {
  execute(): Promise<T>;
}

class GetOrdersQuery implements Query<OrderSummary[]> {
  constructor(
    private userId: string,
    private page: number = 1
  ) {}

  async execute(): Promise<OrderSummary[]> {
    // 読み取り専用のビューから获取
    return prisma.orderView.findMany({
      where: { userId: this.userId },
      orderBy: { createdAt: "desc" },
      take: 20,
      skip: (this.page - 1) * 20,
    });
  }
}

事件ストア

interface StoredEvent {
  id: string;
  aggregateId: string;
  aggregateType: string;
  eventType: string;
  payload: Record<string, unknown>;
  version: number;
  occurredAt: Date;
}

class EventStore {
  async append(
    aggregateId: string,
    aggregateType: string,
    events: DomainEvent[],
    expectedVersion: number
  ) {
    // 楽観的ロック
    const currentVersion = await this.getVersion(aggregateId);
    if (currentVersion !== expectedVersion) {
      throw new Error("Concurrency conflict");
    }

    const storedEvents = events.map((event, i) => ({
      id: event.eventId,
      aggregateId,
      aggregateType,
      eventType: event.constructor.name,
      payload: event as any,
      version: expectedVersion + i + 1,
      occurredAt: event.occurredAt,
    }));

    await prisma.event.createMany({ data: storedEvents });

    // 事件を発行
    for (const event of events) {
      await eventBus.emit(
        event.constructor.name as any,
        event as any
      );
    }
  }

  async getEvents(aggregateId: string): Promise<StoredEvent[]> {
    return prisma.event.findMany({
      where: { aggregateId },
      orderBy: { version: "asc" },
    });
  }

  private async getVersion(aggregateId: string): Promise<number> {
    const last = await prisma.event.findFirst({
      where: { aggregateId },
      orderBy: { version: "desc" },
    });
    return last?.version ?? 0;
  }
}

通过 Claude Codeの活用

事件駆動架构の实现を让 Claude Code依頼する例です。异步処理相关内容请参阅作业队列・异步処理、外部联动はWebhook实现パターン也请参阅。

イベント駆動アーキテクチャを導入して。
- 型安全なイベントバスの実装
- ドメインイベントパターン
- CQRS: コマンドとクエリの分離
- イベントストアの実装
- 既存のサービス層をリファクタリングして統合

事件駆動设计的详细信息请参阅Martin Fowler - Event-Driven Architecture。Claude Codeの使い方は官方文档中可以查看。

总结

事件駆動架构はシステムの疎結合性と可扩展性を高めます。借助 Claude Code，类型安全な事件バスからCQRS、事件ソーシングまで、段階的にEDAを引入可以。