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	已有 PostgreSQL 技术栈	需要手动实现
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 序列不能确定提交顺序：并发事务可能先分配位置、后乱序提交，导致轮询检查点跳过较晚提交的记录。教学实现使用一个事务级 advisory lock 串行化所有追加操作；绕过此锁会破坏该保证，且必须测量其吞吐量。唯一的流/版本约束已经提供流索引。追加权限限制、稳定的请求 ID/事件 ID 以及重复请求协调仍需实现：UUID 主键可以阻止重复插入，但不会让重试变成成功的幂等响应。切勿将已变更的内存中事件版本用作事务已提交的依据。

DynamoDB 示例只执行一次有条件的事件写入，并非完整的事件存储：它可以防止覆盖该键上的记录，但无法原子地检查流头版本、填补间隙、对请求去重、实现全局排序或处理多事件批次。若要实现这些保证，请使用带有条件流头记录的事务；read_stream 只返回一个查询页，必须继续读取 LastEvaluatedKey 才能获取完整结果。不要将最终一致性时间戳 GSI 用作全序检查点。

esdbclient 示例使用同步客户端/迭代器和同步处理程序；异步服务应使用与版本匹配的异步客户端。其明文 localhost URI 仅适用于隔离的开发服务器；远程连接应使用经过身份验证的 TLS。类别流需要启用/配置系统投影。限制读取量，实现关闭/背压机制，并为每个订阅指定一个检查点所有者。处理程序可能在失败后重复执行：应确保其外部副作用具备幂等性或采用事务处理。快照只是优化手段，并非事实来源。部署前，应演练并发追加、中断的提交、重复 ID、投影重放和恢复。避免在不可变事件中存储原始个人数据；应制定保留策略以及符合法律要求的删改/加密擦除政策，而不是绝对声称历史数据永远无法移除。