イベントソーシング本番運用のアンチパターン：スキーマ進化、スナップショット、イベントストアスケーリング

// バージョン1: 元のイベント
interface OrderPlacedV1 {
  type: 'OrderPlaced'
  orderId: string
  customerId: string
  totalAmount: number
}

// バージョン2: shippingAddressを追加（後方互換のためオプショナル）
interface OrderPlacedV2 {
  type: 'OrderPlaced'
  orderId: string
  customerId: string
  totalAmount: number
  shippingAddress?: {
    street: string
    city: string
    zipCode: string
    country: string
  }
  currency?: string // デフォルト: 'USD'
}

// コンシューマーは寛容なリーダーパターンを使用
function handleOrderPlaced(event: OrderPlacedV1 | OrderPlacedV2) {
  const currency = 'currency' in event ? event.currency : 'USD'
  const address = 'shippingAddress' in event ? event.shippingAddress : null // フォールバックとして顧客プロファイルから取得
}

from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class EventEnvelope:
    stream_id: str
    event_type: str
    version: int
    data: dict
    metadata: dict

class UpcasterChain:
    """アップキャスターをチェーンしてイベントを任意のバージョンから最新に変換する。"""

    def __init__(self):
        self._upcasters: dict[tuple[str, int], Callable] = {}

    def register(self, event_type: str, from_version: int,
                 upcaster: Callable[[dict], dict]):
        self._upcasters[(event_type, from_version)] = upcaster

    def upcast(self, envelope: EventEnvelope,
               target_version: int) -> EventEnvelope:
        current = envelope
        while current.version < target_version:
            key = (current.event_type, current.version)
            if key not in self._upcasters:
                raise ValueError(
                    f"No upcaster for {current.event_type} "
                    f"v{current.version} -> v{current.version + 1}"
                )
            new_data = self._upcasters[key](current.data)
            current = EventEnvelope(
                stream_id=current.stream_id,
                event_type=current.event_type,
                version=current.version + 1,
                data=new_data,
                metadata=current.metadata,
            )
        return current

# アップキャスターの登録
chain = UpcasterChain()

# V1 -> V2: 'name'を'firstName'と'lastName'に分割
chain.register('CustomerRegistered', 1, lambda d: {
    **d,
    'firstName': d['name'].split(' ')[0],
    'lastName': ' '.join(d['name'].split(' ')[1:]),
})

# V2 -> V3: デフォルト値で'tier'フィールドを追加
chain.register('CustomerRegistered', 2, lambda d: {
    **d,
    'tier': 'standard',
})

// OrderPlacedV3で破壊的変更を加える代わりに、
// 完全に新しいイベントタイプを作成
interface OrderSubmitted {
  type: 'OrderSubmitted' // 新しい名前が新しいセマンティクスを示す
  orderId: string
  customer: {
    id: string
    tier: 'standard' | 'premium' | 'enterprise'
  }
  lineItems: Array<{
    productId: string
    quantity: number
    unitPrice: number
    discount: number
  }>
  pricing: {
    subtotal: number
    tax: number
    shipping: number
    total: number
    currency: string
  }
  submittedAt: string
}

-- 警告: これは破壊的操作。必ず完全バックアップを取得すること。
-- ステップ1: 新しいeventsテーブルを作成
CREATE TABLE events_v2 (
    global_position BIGSERIAL PRIMARY KEY,
    stream_id UUID NOT NULL,
    stream_position INTEGER NOT NULL,
    event_type VARCHAR(256) NOT NULL,
    data JSONB NOT NULL,
    metadata JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE (stream_id, stream_position)
);

-- ステップ2: 変換を伴う移行
INSERT INTO events_v2 (stream_id, stream_position, event_type, data, metadata, created_at)
SELECT
    stream_id,
    stream_position,
    event_type,
    CASE
        WHEN event_type = 'CustomerRegistered' AND data->>'version' = '1'
        THEN jsonb_build_object(
            'version', '3',
            'firstName', split_part(data->>'name', ' ', 1),
            'lastName', regexp_replace(data->>'name', '^[^ ]+ ', ''),
            'email', data->>'email',
            'tier', 'standard'
        )
        ELSE data
    END,
    metadata,
    created_at
FROM events
ORDER BY global_position;

-- ステップ3: テーブルのスワップ（メンテナンスウィンドウ中に実施）
ALTER TABLE events RENAME TO events_deprecated;
ALTER TABLE events_v2 RENAME TO events;

from dataclasses import dataclass, field
from typing import Optional
import json

@dataclass
class Snapshot:
    stream_id: str
    version: int  # スナップショット時のイベントバージョン
    state: dict
    created_at: str

class SnapshotStore:
    """PostgreSQLバックエンドのスナップショットストア。"""

    def __init__(self, conn):
        self._conn = conn

    async def save_snapshot(self, snapshot: Snapshot):
        await self._conn.execute("""
            INSERT INTO snapshots (stream_id, version, state, created_at)
            VALUES ($1, $2, $3, $4)
            ON CONFLICT (stream_id) DO UPDATE
            SET version = $4, state = $3, created_at = $4
        """, snapshot.stream_id, snapshot.version,
             json.dumps(snapshot.state), snapshot.created_at)

    async def load_snapshot(self, stream_id: str) -> Optional[Snapshot]:
        row = await self._conn.fetchrow("""
            SELECT stream_id, version, state, created_at
            FROM snapshots WHERE stream_id = $1
        """, stream_id)
        if row:
            return Snapshot(
                stream_id=row['stream_id'],
                version=row['version'],
                state=json.loads(row['state']),
                created_at=row['created_at'],
            )
        return None

class AggregateRepository:
    SNAPSHOT_INTERVAL = 100  # 100件ごとにスナップショット

    def __init__(self, event_store, snapshot_store):
        self._event_store = event_store
        self._snapshot_store = snapshot_store

    async def load(self, stream_id: str):
        snapshot = await self._snapshot_store.load_snapshot(stream_id)

        if snapshot:
            aggregate = self._rebuild_from_snapshot(snapshot)
            start_version = snapshot.version + 1
        else:
            aggregate = self._create_empty()
            start_version = 0

        events = await self._event_store.read_stream(
            stream_id, from_version=start_version
        )
        for event in events:
            aggregate.apply(event)

        return aggregate

    async def save(self, aggregate):
        new_events = aggregate.uncommitted_events
        current_version = aggregate.version

        await self._event_store.append(
            aggregate.stream_id, new_events, current_version
        )

        # 閾値に達したらスナップショットを作成
        if current_version % self.SNAPSHOT_INTERVAL == 0:
            await self._snapshot_store.save_snapshot(Snapshot(
                stream_id=aggregate.stream_id,
                version=current_version,
                state=aggregate.to_dict(),
                created_at=self._now(),
            ))

import asyncio
from dataclasses import dataclass

@dataclass
class ProjectionRebuildConfig:
    batch_size: int = 1000
    max_concurrent_batches: int = 5
    checkpoint_interval: int = 10_000
    progress_log_interval: int = 50_000

class ProjectionRebuilder:
    """バッチ処理とチェックポイントによる最適化されたプロジェクション再構築。"""

    def __init__(self, event_store, projection_handler, checkpoint_store,
                 config: ProjectionRebuildConfig = None):
        self._event_store = event_store
        self._handler = projection_handler
        self._checkpoint_store = checkpoint_store
        self._config = config or ProjectionRebuildConfig()

    async def rebuild(self, projection_name: str):
        # 再構築が中断された場合、最後のチェックポイントから再開
        last_position = await self._checkpoint_store.get(
            f"rebuild:{projection_name}"
        )
        start_position = last_position + 1 if last_position else 0

        total_processed = 0
        position = start_position

        while True:
            events = await self._event_store.read_all(
                from_position=position,
                count=self._config.batch_size,
            )
            if not events:
                break

            # バッチを処理
            await self._handler.handle_batch(events)
            position = events[-1].global_position + 1
            total_processed += len(events)

            # 定期的にチェックポイント
            if total_processed % self._config.checkpoint_interval == 0:
                await self._checkpoint_store.save(
                    f"rebuild:{projection_name}",
                    events[-1].global_position,
                )

            if total_processed % self._config.progress_log_interval == 0:
                print(
                    f"再構築進捗: {total_processed}件処理済み, "
                    f"ポジション: {position}"
                )

        print(f"再構築完了: 合計{total_processed}件処理済み")

// 悪い例: プロパティソーシング - ビジネス意図がない
const badEvents = [
  { type: 'FieldUpdated', field: 'status', value: 'suspended' },
  { type: 'FieldUpdated', field: 'reason', value: 'payment_failed' },
]

// 良い例: ビジネス意味を持つドメインイベント
const goodEvents = [
  {
    type: 'AccountSuspendedDueToPaymentFailure',
    accountId: 'ACC-123',
    failedPaymentId: 'PAY-456',
    suspendedAt: '2026-03-07T10:30:00Z',
    retryScheduledAt: '2026-03-10T10:30:00Z',
  },
]

-- 悪い例: 盲目的なINSERT（再配信時に重複）
INSERT INTO order_summary (order_id, status, total)
VALUES ('ORD-001', 'placed', 150.00);

-- 良い例: 冪等なUPSERT
INSERT INTO order_summary (order_id, status, total, last_event_position)
VALUES ('ORD-001', 'placed', 150.00, 42)
ON CONFLICT (order_id) DO UPDATE
SET status = EXCLUDED.status,
    total = EXCLUDED.total,
    last_event_position = EXCLUDED.last_event_position
WHERE order_summary.last_event_position < EXCLUDED.last_event_position;

+------------------+     +-----------------+     +------------------+
|   API Gateway    |---->| Command Handler |---->| Domain Aggregate |
+------------------+     +-----------------+     +--------+---------+
                                                          |
                              +---------------------------+
                              | Domain Events
                              v
                    +--------------------+
                    |    Event Store     |
                    | (append-only log)  |
                    +--------+-----------+
                             |
              +--------------+--------------+
              |              |              |
              v              v              v
     +--------+---+  +------+------+  +----+--------+
     | Projection |  | Projection  |  | Projection  |
     | Handler A  |  | Handler B   |  | Handler C   |
     +--------+---+  +------+------+  +----+--------+
              |              |              |
              v              v              v
     +--------+---+  +------+------+  +----+--------+
     | PostgreSQL |  |Elasticsearch|  |    Redis     |
     | Read Model |  |Search Index |  |   Cache      |
     +------------+  +-------------+  +-------------+

イベントスキーマ進化パイプライン:

  Event Store (生のバイト列)
       |
       v
  +----------+     +-----------+     +-----------+
  | Upcaster |---->| Upcaster  |---->| Upcaster  |
  |  V1->V2  |     |  V2->V3   |     |  V3->V4   |
  +----------+     +-----------+     +-----------+
                                          |
                                          v
                                  +---------------+
                                  | Current Event |
                                  | (Version 4)   |
                                  +---------------+
                                          |
                              +-----------+-----------+
                              |                       |
                              v                       v
                     +--------+------+      +---------+-----+
                     | Command Side  |      | Projection    |
                     | (Aggregate)   |      | (Read Model)  |
                     +---------------+      +---------------+