取り組みの背景 — なぜ今、Event-Driven Architecture なのか
モダンなアプリケーション開発では、サービス間の依存関係が複雑化し、同期的なAPI呼び出しだけではスケーラビリティやレジリエンスに限界が生じます。Event-Driven Architecture（EDA） は、サービス同士をイベントで疎結合に接続し、非同期で柔軟なシステムを構築するためのアプローチです。
Antigravity の AIエージェントは、こうしたアーキテクチャの設計・実装を驚くほど効率的にサポートしてくれます。ここで扱うのはEDA の基本概念から、CQRS・Sagaパターンといった高度な設計手法まで、Antigravity を活用して実際にリアクティブマイクロサービスを構築する方法を詳しく解説します。
この記事で学べること:
- Event-Driven Architecture の核心概念と設計原則
- Antigravity エージェントを使ったイベントバス・メッセージブローカーの実装
- CQRS（Command Query Responsibility Segregation）パターンの実践
- Saga パターンによる分散トランザクション管理
- イベントソーシングとスナップショットの実装テクニック
対象読者: マイクロサービス開発経験があり、より高度なアーキテクチャパターンを習得したい中〜上級エンジニア
Event-Driven Architecture の基本概念
イベントとは何か
EDA における「イベント」とは、システム内で発生した事実（fact）の記録です。過去形で命名するのが慣例で、OrderPlaced、PaymentProcessed、InventoryUpdated のように表現します。
イベントには3つの基本タイプがあります:
- Domain Event: ビジネスロジック上の状態変化（例:
OrderShipped）
- Integration Event: サービス間で共有される外部イベント（例:
UserRegistered）
- System Event: インフラレベルの通知（例:
ServiceHealthChanged）
EDA の3つの設計パターン
| パターン | 用途 | 特徴 |
|---|
| Event Notification | 状態変化の通知 | 軽量、最小限の情報を伝搬 |
| Event-Carried State Transfer | データ同期 | イベントに完全な状態を含める |
| Event Sourcing | 状態の再構築 | 全イベントを永続化し、任意時点の状態を復元 |
Antigravity エージェントでイベントバスを構築する
Antigravity のエージェントに指示を出して、Node.js + TypeScript ベースのイベントバスを構築してみましょう。以下はエージェントとの対話で生成される典型的な実装です。
// src/events/event-bus.ts
// Antigravity エージェントに「型安全なイベントバスを作って」と指示して生成
import { EventEmitter } from "events";
// イベントの型定義
interface EventMap {
"order.placed": { orderId: string; userId: string; items: OrderItem[]; total: number };
"payment.processed": { orderId: string; transactionId: string; status: "success" | "failed" };
"inventory.reserved": { orderId: string; items: ReservedItem[] };
"order.shipped": { orderId: string; trackingNumber: string };
}
interface OrderItem {
productId: string;
quantity: number;
price: number;
}
interface ReservedItem {
productId: string;
quantity: number;
warehouseId: string;
}
// 型安全なイベントバスクラス
class TypedEventBus {
private emitter = new EventEmitter();
private handlers = new Map<string, Set<Function>>();
// イベントの発行（型安全）
emit<K extends keyof EventMap>(event: K, payload: EventMap[K]): void {
const envelope = {
type: event,
payload,
timestamp: new Date().toISOString(),
correlationId: crypto.randomUUID(),
};
console.log(`[EventBus] Emitting: ${event}`, envelope.correlationId);
this.emitter.emit(event, envelope);
}
// イベントの購読（型安全）
on<K extends keyof EventMap>(
event: K,
handler: (payload: EventMap[K], metadata: EventMetadata) => Promise<void>
): () => void {
const wrappedHandler = async (envelope: EventEnvelope<EventMap[K]>) => {
try {
await handler(envelope.payload, {
timestamp: envelope.timestamp,
correlationId: envelope.correlationId,
});
} catch (error) {
console.error(`[EventBus] Handler error for ${event}:`, error);
}
};
this.emitter.on(event, wrappedHandler);
// クリーンアップ関数を返す
return () => this.emitter.off(event, wrappedHandler);
}
}
interface EventMetadata {
timestamp: string;
correlationId: string;
}
interface EventEnvelope<T> {
type: string;
payload: T;
timestamp: string;
correlationId: string;
}
export const eventBus = new TypedEventBus();
このコードをエージェントに生成してもらう際のプロンプト例:
TypeScriptの型安全なイベントバスを実装して。
要件:
- EventEmitterベースで軽量に
- ジェネリクスで型安全性を確保
- correlationId でイベントチェーンを追跡可能に
- エラーハンドリング内蔵
- クリーンアップ関数を返すように
CQRS パターンの実装
CQRS（Command Query Responsibility Segregation）は、データの書き込み（Command）と読み取り（Query）のモデルを分離するパターンです。EDA との相性が非常に良く、書き込み側で発生したイベントを読み取り側のビューに反映させます。
Command 側の実装
// src/commands/place-order.command.ts
// 注文コマンドハンドラー — 書き込みモデル
import { eventBus } from "../events/event-bus";
interface PlaceOrderCommand {
userId: string;
items: Array<{ productId: string; quantity: number; price: number }>;
}
class OrderCommandHandler {
async execute(command: PlaceOrderCommand): Promise<string> {
// 1. バリデーション
if (command.items.length === 0) {
throw new Error("注文には最低1つの商品が必要です");
}
// 2. 集約の生成
const orderId = crypto.randomUUID();
const total = command.items.reduce(
(sum, item) => sum + item.price * item.quantity,
0
);
// 3. 永続化（Write Model）
await this.saveOrder({
id: orderId,
userId: command.userId,
items: command.items,
total,
status: "pending",
createdAt: new Date(),
});
// 4. ドメインイベントの発行
eventBus.emit("order.placed", {
orderId,
userId: command.userId,
items: command.items,
total,
});
return orderId;
}
private async saveOrder(order: Record<string, unknown>): Promise<void> {
// データベースへの永続化ロジック
console.log(`[OrderCommand] Order saved: ${order.id}`);
}
}
export const orderCommandHandler = new OrderCommandHandler();
Query 側の実装
// src/queries/order-view.query.ts
// 注文クエリハンドラー — 読み取りモデル（非正規化ビュー）
import { eventBus } from "../events/event-bus";
interface OrderView {
orderId: string;
userId: string;
itemCount: number;
total: number;
status: string;
lastUpdated: string;
}
class OrderQueryHandler {
// 非正規化されたビューストア（実際にはRedis/DynamoDB等）
private views = new Map<string, OrderView>();
constructor() {
// イベントを購読してビューを更新
eventBus.on("order.placed", async (payload) => {
this.views.set(payload.orderId, {
orderId: payload.orderId,
userId: payload.userId,
itemCount: payload.items.length,
total: payload.total,
status: "pending",
lastUpdated: new Date().toISOString(),
});
});
eventBus.on("payment.processed", async (payload) => {
const view = this.views.get(payload.orderId);
if (view) {
view.status = payload.status === "success" ? "paid" : "payment_failed";
view.lastUpdated = new Date().toISOString();
}
});
eventBus.on("order.shipped", async (payload) => {
const view = this.views.get(payload.orderId);
if (view) {
view.status = "shipped";
view.lastUpdated = new Date().toISOString();
}
});
}
// 高速な読み取りクエリ
getOrder(orderId: string): OrderView | undefined {
return this.views.get(orderId);
}
getOrdersByUser(userId: string): OrderView[] {
return Array.from(this.views.values()).filter(
(v) => v.userId === userId
);
}
}
export const orderQueryHandler = new OrderQueryHandler();
期待する動作:
[OrderCommand] Order saved: abc-123
[EventBus] Emitting: order.placed abc-123
[QueryHandler] View updated: abc-123 → status: pending
[EventBus] Emitting: payment.processed abc-123
[QueryHandler] View updated: abc-123 → status: paid
CQRS の利点は、読み取りモデルをユースケースに最適化できることです。ダッシュボード用、検索用、レポート用など、異なるビューを同じイベントストリームから構築できます。
Saga パターンによる分散トランザクション管理
マイクロサービスでは、複数のサービスにまたがるトランザクションをデータベースの ACID トランザクションだけで管理することはできません。Saga パターンは、一連のローカルトランザクションとその補償（ロールバック）アクションを定義することで、最終的な整合性を保証します。
// src/sagas/order-saga.ts
// 注文処理サガ — オーケストレーション方式
import { eventBus } from "../events/event-bus";
interface SagaStep {
name: string;
execute: () => Promise<void>;
compensate: () => Promise<void>;
}
class OrderSaga {
private completedSteps: SagaStep[] = [];
async execute(orderId: string, userId: string): Promise<void> {
const steps: SagaStep[] = [
{
name: "在庫の確保",
execute: async () => {
// 在庫サービスに確保リクエスト
await this.reserveInventory(orderId);
console.log(`[Saga] ✅ 在庫確保完了: ${orderId}`);
},
compensate: async () => {
// 在庫の確保を解除
await this.releaseInventory(orderId);
console.log(`[Saga] ↩️ 在庫確保を解除: ${orderId}`);
},
},
{
name: "決済の処理",
execute: async () => {
await this.processPayment(orderId, userId);
console.log(`[Saga] ✅ 決済完了: ${orderId}`);
},
compensate: async () => {
await this.refundPayment(orderId);
console.log(`[Saga] ↩️ 返金処理: ${orderId}`);
},
},
{
name: "配送手配",
execute: async () => {
await this.arrangeShipping(orderId);
console.log(`[Saga] ✅ 配送手配完了: ${orderId}`);
},
compensate: async () => {
await this.cancelShipping(orderId);
console.log(`[Saga] ↩️ 配送キャンセル: ${orderId}`);
},
},
];
// ステップを順番に実行し、失敗時は補償を逆順で実行
for (const step of steps) {
try {
await step.execute();
this.completedSteps.push(step);
} catch (error) {
console.error(`[Saga] ❌ ${step.name} 失敗:`, error);
await this.rollback();
throw new Error(`Saga failed at step: ${step.name}`);
}
}
console.log(`[Saga] 🎉 注文処理完了: ${orderId}`);
}
private async rollback(): Promise<void> {
console.log("[Saga] ロールバック開始...");
// 完了済みステップを逆順で補償
for (const step of [...this.completedSteps].reverse()) {
try {
await step.compensate();
} catch (compensateError) {
// 補償失敗はログに記録して続行（手動介入が必要）
console.error(
`[Saga] 補償失敗 (${step.name}):`,
compensateError
);
}
}
}
// 各ステップの具体的な実装
private async reserveInventory(orderId: string): Promise<void> {
// 在庫サービスAPIを呼び出し
eventBus.emit("inventory.reserved", {
orderId,
items: [{ productId: "prod-1", quantity: 1, warehouseId: "wh-tokyo" }],
});
}
private async releaseInventory(orderId: string): Promise<void> {
// 在庫解放APIを呼び出し
}
private async processPayment(orderId: string, userId: string): Promise<void> {
// 決済サービスAPIを呼び出し
eventBus.emit("payment.processed", {
orderId,
transactionId: crypto.randomUUID(),
status: "success",
});
}
private async refundPayment(orderId: string): Promise<void> {
// 返金処理
}
private async arrangeShipping(orderId: string): Promise<void> {
// 配送サービスAPIを呼び出し
eventBus.emit("order.shipped", {
orderId,
trackingNumber: `TRK-${Date.now()}`,
});
}
private async cancelShipping(orderId: string): Promise<void> {
// 配送キャンセル処理
}
}
export { OrderSaga };
期待する出力（正常系）:
[Saga] ✅ 在庫確保完了: order-789
[Saga] ✅ 決済完了: order-789
[Saga] ✅ 配送手配完了: order-789
[Saga] 🎉 注文処理完了: order-789
期待する出力（決済失敗時のロールバック）:
[Saga] ✅ 在庫確保完了: order-789
[Saga] ❌ 決済の処理 失敗: PaymentDeclinedError
[Saga] ロールバック開始...
[Saga] ↩️ 在庫確保を解除: order-789
イベントソーシングとスナップショット
イベントソーシングは、エンティティの現在の状態ではなく、状態変化の履歴（イベント）をすべて永続化するパターンです。任意の時点の状態を再構築できるため、監査ログやデバッグに非常に強力です。
// src/event-sourcing/event-store.ts
// イベントストア — イベントソーシングの中核
interface StoredEvent {
id: string;
aggregateId: string;
type: string;
payload: Record<string, unknown>;
version: number;
timestamp: string;
}
class EventStore {
private events: StoredEvent[] = [];
private snapshots = new Map<string, { state: unknown; version: number }>();
// イベントの追記（append-only）
async append(
aggregateId: string,
type: string,
payload: Record<string, unknown>,
expectedVersion: number
): Promise<void> {
const currentVersion = this.getLatestVersion(aggregateId);
// 楽観的排他制御
if (currentVersion !== expectedVersion) {
throw new Error(
`Concurrency conflict: expected v${expectedVersion}, got v${currentVersion}`
);
}
this.events.push({
id: crypto.randomUUID(),
aggregateId,
type,
payload,
version: expectedVersion + 1,
timestamp: new Date().toISOString(),
});
}
// 集約のイベント履歴を取得
getEvents(aggregateId: string, fromVersion = 0): StoredEvent[] {
return this.events.filter(
(e) => e.aggregateId === aggregateId && e.version > fromVersion
);
}
// スナップショットの保存（パフォーマンス最適化）
saveSnapshot(aggregateId: string, state: unknown, version: number): void {
this.snapshots.set(aggregateId, { state, version });
console.log(
`[EventStore] Snapshot saved: ${aggregateId} at v${version}`
);
}
// スナップショットからの復元 + 以降のイベント適用
getSnapshot(
aggregateId: string
): { state: unknown; version: number } | undefined {
return this.snapshots.get(aggregateId);
}
private getLatestVersion(aggregateId: string): number {
const events = this.events.filter(
(e) => e.aggregateId === aggregateId
);
return events.length > 0
? Math.max(...events.map((e) => e.version))
: 0;
}
}
// 集約の状態をイベントから再構築する例
class OrderAggregate {
private state = { status: "created", items: [] as string[], total: 0 };
private version = 0;
// イベントを適用して状態を再構築
apply(event: StoredEvent): void {
switch (event.type) {
case "OrderCreated":
this.state.status = "created";
break;
case "ItemAdded":
this.state.items.push(event.payload.productId as string);
this.state.total += event.payload.price as number;
break;
case "OrderConfirmed":
this.state.status = "confirmed";
break;
case "OrderCancelled":
this.state.status = "cancelled";
break;
}
this.version = event.version;
}
// スナップショット + 残りのイベントから高速復元
static restore(
snapshot: { state: unknown; version: number } | undefined,
events: StoredEvent[]
): OrderAggregate {
const aggregate = new OrderAggregate();
if (snapshot) {
aggregate.state = snapshot.state as typeof aggregate.state;
aggregate.version = snapshot.version;
}
for (const event of events) {
aggregate.apply(event);
}
return aggregate;
}
getState() {
return { ...this.state, version: this.version };
}
}
export { EventStore, OrderAggregate };
よくあるエラーと対処法
イベントの順序保証
分散環境では、イベントが発行された順序と消費される順序が異なる場合があります。これに対処するには、パーティションキー（注文IDなど）を使って同じ集約のイベントが順序通りに処理されるようにします。
べき等性の確保
ネットワーク障害によりイベントが重複配信される可能性があります。各ハンドラーをべき等に設計する点が肝心です。
// べき等なイベントハンドラーの例
const processedEvents = new Set<string>();
async function idempotentHandler(eventId: string, handler: () => Promise<void>) {
if (processedEvents.has(eventId)) {
console.log(`[Idempotent] Skip duplicate: ${eventId}`);
return; // 重複イベントをスキップ
}
await handler();
processedEvents.add(eventId);
}
スキーマ進化への対応
イベントのスキーマはシステムの進化とともに変わります。バージョニング戦略として、v1、v2 のようなサフィックスを付けるか、アップキャスター（古いバージョンを新しいバージョンに変換する関数）を用意しましょう。
個人開発者の視点から（実体験メモ）
まとめと次のステップ
Event-Driven Architecture は、疎結合でスケーラブル、かつレジリエントなマイクロサービスを構築するための強力なアプローチです。CQRS で読み書きを分離し、Saga パターンで分散トランザクションを管理し、イベントソーシングで完全な履歴を保持する — これらのパターンを組み合わせることで、本格的なプロダクションシステムを設計できます。
Antigravity のAIエージェントは、これらの複雑なパターンのボイラープレートコードを素早く生成し、設計の反復サイクルを大幅に短縮してくれます。まずは小さなイベントバスから始めて、段階的にCQRS やSaga を導入していくことをおすすめします。
さらに深く学びたい方は、高度なマルチエージェントオーケストレーションや WebSocket リアルタイムアプリの記事もあわせてご覧ください。