event-store-design

イベントストアの設計

イベントソーシングアプリケーション向けのイベントストア設計に関する包括的なガイドです。

このスキルを使用しない場合

• タスクがイベントストアの設計に関係しない場合
• この範囲外の別のドメインやツールが必要な場合

手順

• 目標、制約、必要な入力を明確にします。
• 関連するベストプラクティスを適用し、結果を検証します。
• 実行可能な手順と検証方法を提示します。
• 詳細な例が必要な場合は、resources/implementation-playbook.md を開きます。

このスキルを使用する場合

• イベントソーシングのインフラを設計する場合
• イベントストアのテクノロジーを選択する場合
• カスタムイベントストアを実装する場合
• イベントの保存と取得を最適化する場合
• イベントストアのスキーマを設定する場合
• イベントストアのスケーリングを計画する場合

主な概念

1. イベントストアのアーキテクチャ

┌─────────────────────────────────────────────────────┐
│                    Event Store                       │
├─────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │
│  │   Stream 1   │  │   Stream 2   │  │   Stream 3   │ │
│  │ (Aggregate)  │  │ (Aggregate)  │  │ (Aggregate)  │ │
│  ├─────────────┤  ├─────────────┤  ├─────────────┤ │
│  │ Event 1     │  │ Event 1     │  │ Event 1     │ │
│  │ Event 2     │  │ Event 2     │  │ Event 2     │ │
│  │ Event 3     │  │ ...         │  │ Event 3     │ │
│  │ ...         │  │             │  │ Event 4     │ │
│  └─────────────┘  └─────────────┘  └─────────────┘ │
├─────────────────────────────────────────────────────┤
│  Global Position: 1 → 2 → 3 → 4 → 5 → 6 → ...     │
└─────────────────────────────────────────────────────┘

2. イベントストアの要件

要件	説明
追記専用	イベントは不変で、追記のみ可能です
順序付け	ストリーム内および全体で順序が保たれます
バージョン管理	楽観的同時実行制御を行います
サブスクリプション	リアルタイムでイベントを通知します
冪等性	重複する書き込みを安全に処理します

テクノロジーの比較

テクノロジー	最適な用途	制限事項
EventStoreDB	純粋なイベントソーシング	用途が限定されています
PostgreSQL	既存の Postgres スタック	手動での実装が必要です
Kafka	高スループットのストリーミング	ストリーム単位のクエリには不向きです
DynamoDB	サーバーレス、AWS ネイティブ	クエリに制限があります
Marten	.NET エコシステム	.NET 専用です

テンプレート

テンプレート 1：PostgreSQL イベントストアのスキーマ

-- Events table
CREATE TABLE events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    stream_id VARCHAR(255) NOT NULL,
    stream_type VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    event_data JSONB NOT NULL,
    metadata JSONB DEFAULT '{}',
    version BIGINT NOT NULL,
    global_position BIGSERIAL,
    created_at TIMESTAMPTZ DEFAULT NOW(),

    CONSTRAINT unique_stream_version UNIQUE (stream_id, version)
);

-- Index for stream queries
CREATE INDEX idx_events_stream_id ON events(stream_id, version);

-- Index for global subscription
CREATE INDEX idx_events_global_position ON events(global_position);

-- Index for event type queries
CREATE INDEX idx_events_event_type ON events(event_type);

-- Index for time-based queries
CREATE INDEX idx_events_created_at ON events(created_at);

-- Snapshots table
CREATE TABLE snapshots (
    stream_id VARCHAR(255) PRIMARY KEY,
    stream_type VARCHAR(255) NOT NULL,
    snapshot_data JSONB NOT NULL,
    version BIGINT NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Subscriptions checkpoint table
CREATE TABLE subscription_checkpoints (
    subscription_id VARCHAR(255) PRIMARY KEY,
    last_position BIGINT NOT NULL DEFAULT 0,
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

テンプレート 2：Python イベントストアの実装

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional, List
from uuid import UUID, uuid4
import json
import asyncio
import asyncpg

@dataclass
class Event:
    stream_id: str
    event_type: str
    data: dict
    metadata: dict = field(default_factory=dict)
    event_id: UUID = field(default_factory=uuid4)
    version: Optional[int] = None
    global_position: Optional[int] = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


class EventStore:
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    async def append_events(
        self,
        stream_id: str,
        stream_type: str,
        events: List[Event],
        expected_version: Optional[int] = None
    ) -> List[Event]:
        """Append events to a stream with optimistic concurrency."""
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                # Serialize ALL appenders before allocating positions. This simple global
                # lock trades throughput for commit-consistent polling; every writer must use it.
                await conn.execute("SELECT pg_advisory_xact_lock(784291)")
                # Check expected version
                if expected_version is not None:
                    current = await conn.fetchval(
                        "SELECT MAX(version) FROM events WHERE stream_id = $1",
                        stream_id
                    )
                    current = current or 0
                    if current != expected_version:
                        raise ConcurrencyError(
                            f"Expected version {expected_version}, got {current}"
                        )

                # Get starting version
                start_version = await conn.fetchval(
                    "SELECT COALESCE(MAX(version), 0) + 1 FROM events WHERE stream_id = $1",
                    stream_id
                )

                # Insert events
                saved_events = []
                for i, event in enumerate(events):
                    event.version = start_version + i
                    row = await conn.fetchrow(
                        """
                        INSERT INTO events (id, stream_id, stream_type, event_type,
                                          event_data, metadata, version, created_at)
                        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                        RETURNING global_position
                        """,
                        event.event_id,
                        stream_id,
                        stream_type,
                        event.event_type,
                        json.dumps(event.data),
                        json.dumps(event.metadata),
                        event.version,
                        event.created_at
                    )
                    event.global_position = row['global_position']
                    saved_events.append(event)

                return saved_events

    async def read_stream(
        self,
        stream_id: str,
        from_version: int = 0,
        limit: int = 1000
    ) -> List[Event]:
        """Read events from a stream."""
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT id, stream_id, event_type, event_data, metadata,
                       version, global_position, created_at
                FROM events
                WHERE stream_id = $1 AND version >= $2
                ORDER BY version
                LIMIT $3
                """,
                stream_id, from_version, limit
            )
            return [self._row_to_event(row) for row in rows]

    async def read_all(
        self,
        from_position: int = 0,
        limit: int = 1000
    ) -> List[Event]:
        """Read all events globally."""
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT id, stream_id, event_type, event_data, metadata,
                       version, global_position, created_at
                FROM events
                WHERE global_position > $1
                ORDER BY global_position
                LIMIT $2
                """,
                from_position, limit
            )
            return [self._row_to_event(row) for row in rows]

    async def subscribe(
        self,
        subscription_id: str,
        handler,
        from_position: int = 0,
        batch_size: int = 100
    ):
        """Subscribe to all events from a position."""
        # Get checkpoint
        async with self.pool.acquire() as conn:
            checkpoint = await conn.fetchval(
                """
                SELECT last_position FROM subscription_checkpoints
                WHERE subscription_id = $1
                """,
                subscription_id
            )
            position = from_position if checkpoint is None else checkpoint

        while True:
            events = await self.read_all(position, batch_size)
            if not events:
                await asyncio.sleep(1)  # Poll interval
                continue

            for event in events:
                await handler(event)
                position = event.global_position

            # Save checkpoint
            async with self.pool.acquire() as conn:
                await conn.execute(
                    """
                    INSERT INTO subscription_checkpoints (subscription_id, last_position)
                    VALUES ($1, $2)
                    ON CONFLICT (subscription_id)
                    DO UPDATE SET last_position = $2, updated_at = NOW()
                    """,
                    subscription_id, position
                )

    def _row_to_event(self, row) -> Event:
        return Event(
            event_id=row['id'],
            stream_id=row['stream_id'],
            event_type=row['event_type'],
            data=json.loads(row['event_data']),
            metadata=json.loads(row['metadata']),
            version=row['version'],
            global_position=row['global_position'],
            created_at=row['created_at']
        )


class ConcurrencyError(Exception):
    """Raised when optimistic concurrency check fails."""
    pass

テンプレート 3：EventStoreDB の使用方法

from esdbclient import EventStoreDBClient, NewEvent, StreamState
import json

# Connect
client = EventStoreDBClient(uri="esdb://localhost:2113?tls=false")

# Append events
def append_events(stream_name: str, events: list, expected_revision=None):
    new_events = [
        NewEvent(
            type=event['type'],
            data=json.dumps(event['data']).encode(),
            metadata=json.dumps(event.get('metadata', {})).encode()
        )
        for event in events
    ]

    if expected_revision is None:
        state = StreamState.ANY
    elif expected_revision == -1:
        state = StreamState.NO_STREAM
    else:
        state = expected_revision

    return client.append_to_stream(
        stream_name=stream_name,
        events=new_events,
        current_version=state
    )

# Read stream
def read_stream(stream_name: str, from_revision: int = 0):
    events = client.get_stream(
        stream_name=stream_name,
        stream_position=from_revision
    )
    return [
        {
            'type': event.type,
            'data': json.loads(event.data),
            'metadata': json.loads(event.metadata) if event.metadata else {},
            'stream_position': event.stream_position,
            'commit_position': event.commit_position
        }
        for event in events
    ]

# Subscribe to all
def subscribe_to_all(handler, from_position: int = 0):
    subscription = client.subscribe_to_all(commit_position=from_position)
    for event in subscription:
        handler({
            'type': event.type,
            'data': json.loads(event.data),
            'stream_id': event.stream_name,
            'position': event.commit_position
        })

# Category projection ($ce-Category)
def read_category(category: str):
    """Read all events for a category using system projection."""
    return read_stream(f"$ce-{category}")

テンプレート 4：DynamoDB イベントストア

import boto3
from boto3.dynamodb.conditions import Key
from datetime import datetime, timezone
import json
import uuid

class DynamoEventStore:
    def __init__(self, table_name: str):
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table(table_name)

    def append_events(self, stream_id: str, events: list, expected_version: int = None):
        """Append events with conditional write for concurrency."""
        # Single-event teaching example; no batch atomicity or automatic retry.
        if len(events) != 1 or expected_version is None or expected_version < 0:
            raise ValueError("Supply exactly one event and a nonnegative expected version")
        event = events[0]
        version = expected_version + 1
        item = {
            'PK': f"STREAM#{stream_id}",
            'SK': f"VERSION#{version:020d}",
            'event_id': str(uuid.uuid4()),
            'event_type': event['type'],
            'event_data': json.dumps(event['data']),
            'version': version,
        }
        self.table.put_item(Item=item, ConditionExpression='attribute_not_exists(PK)')
        return events

    def read_stream(self, stream_id: str, from_version: int = 0):
        """Read events from a stream."""
        response = self.table.query(
            ConsistentRead=True,
            KeyConditionExpression=Key('PK').eq(f"STREAM#{stream_id}") &
                                  Key('SK').gte(f"VERSION#{from_version:020d}")
        )
        return [
            {
                'event_type': item['event_type'],
                'data': json.loads(item['event_data']),
                'version': item['version']
            }
            for item in response['Items']
        ]

# Table definition (CloudFormation/Terraform)
"""
DynamoDB Table:
  - PK (Partition Key): String
  - SK (Sort Key): String
  - No global subscription ordering implemented here

Capacity: On-demand or provisioned based on throughput needs
"""

ベストプラクティス

推奨事項

• 集約の種類を含むストリーム ID を使用します - Order-{uuid}
• 相関 ID／原因 ID を含めます - トレースに使用します
• 最初からイベントのバージョンを管理します - スキーマの進化に備えます
• 冪等性を実装します - イベント ID を使用して重複を排除します
• 適切にインデックスを設定します - クエリパターンに合わせます

避けるべき事項

• イベントを更新または削除しないでください - イベントは不変の事実です
• 大きなペイロードを保存しないでください - イベントは小さく保ちます
• 楽観的同時実行制御を省略しないでください - データの破損を防ぎます
• バックプレッシャーを無視しないでください - 処理の遅いコンシューマーに対応します

リソース

• EventStoreDB
• Marten のイベント
• イベントソーシングパターン

制限事項

• このスキルは、タスクが上記の説明された範囲に明確に該当する場合にのみ使用してください。
• この出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認を求めてください。

実装上の制限と検証

PostgreSQL のシーケンスだけではコミット順序は保証されません。並行トランザクションが位置を割り当てた後、順不同でコミットすることがあり、その結果、ポーリングのチェックポイントが後からコミットされたイベントを飛ばす可能性があります。このサンプル実装では、すべての追記処理を単一のトランザクションアドバイザリロックで直列化しています。このロックを迂回すると、その保証は無効になり、スループットを測定する必要があります。一意なストリーム／バージョン制約によって、ストリームインデックスはすでに提供されています。追記専用の権限、安定したリクエスト ID／イベント ID、重複リクエストの調整は、引き続き実装が必要です。UUID の主キーは重複挿入を防ぎますが、再試行を冪等な成功応答に変えるものではありません。変更されたインメモリイベントのバージョンを、トランザクションがコミットされた証拠として再利用しないでください。

DynamoDB の例は条件付きイベント書き込みを 1 回行うものであり、完全なイベントストアではありません。そのキーでの上書きは防ぎますが、ストリームヘッドのバージョンをアトミックにチェックしたり、欠落を補完したり、リクエストを重複排除したり、全体の順序や複数イベントのバッチを実装したりすることはありません。これらを保証するには、条件付きヘッドレコードを使ったトランザクションを使用してください。read_stream はクエリの 1 ページしか返さないため、最後まで読み取るには LastEvaluatedKey をたどる必要があります。結果整合性のタイムスタンプ GSI を全体順序のチェックポイントとして使用しないでください。

esdbclient の例では、同期クライアント／イテレーターと同期ハンドラーを使用しています。非同期サービスでは、バージョンに対応した非同期クライアントを使用してください。平文の localhost URI は隔離された開発サーバー専用です。リモート接続には認証付き TLS を使用してください。カテゴリストリームでは、システムプロジェクションを有効化して設定する必要があります。読み取り件数に上限を設け、シャットダウン／バックプレッシャーを実装し、サブスクリプションごとにチェックポイントの所有者を 1 つにしてください。障害が発生するとハンドラーは再実行される可能性があります。外部への影響を冪等にするか、トランザクションで処理してください。スナップショットは最適化であり、信頼できる情報源ではありません。デプロイ前に、並行追記、中断されたコミット、重複 ID、プロジェクションの再生、復元をリハーサルしてください。変更不可能なイベントに生の個人データを含めないでください。履歴データは決して削除できないという絶対的な主張をするのではなく、保持ポリシーと、法令に準拠したマスキング／暗号鍵破棄のポリシーを定義してください。