From 4738cd8432a6fb4a7de37d164dc313678f38ea2e Mon Sep 17 00:00:00 2001 From: wehub-skill-sync Date: Mon, 13 Jul 2026 21:36:58 +0800 Subject: [PATCH] chore: import zh skill event-store-design --- README.wehub.md | 9 ++ SKILL.md | 80 ++++++++++ references/details.md | 356 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 445 insertions(+) create mode 100644 README.wehub.md create mode 100644 SKILL.md create mode 100644 references/details.md diff --git a/README.wehub.md b/README.wehub.md new file mode 100644 index 0000000..c6cbf3d --- /dev/null +++ b/README.wehub.md @@ -0,0 +1,9 @@ +# WeHub 来源说明 + +- Skill 名称：`event-store-design` +- 中文类目：事件存储设计 +- 上游仓库：`wshobson__agents` +- 上游路径：`plugins/backend-development/skills/event-store-design/SKILL.md` +- 上游链接：https://github.com/wshobson/agents/blob/HEAD/plugins/backend-development/skills/event-store-design/SKILL.md +- 本仓库为 WeHub 中文 Skill 汉化包，基于 skill 市场筛选 Top200 清单整理 +- 原作者、版权和许可证信息以上游仓库为准 diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..9cc8195 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,80 @@ +--- +name: event-store-design +description: 为事件溯源系统设计并实现事件存储。适用于构建事件溯源基础设施、选择事件存储技术或实现事件持久化模式。 +--- + +# 事件存储设计 + +面向事件溯源应用的事件存储设计综合指南。 + +## 何时使用本技能 + +- 设计事件溯源基础设施 +- 在事件存储技术之间进行选择 +- 实现自定义事件存储 +- 优化事件存储与检索 +- 设置事件存储模式 +- 规划事件存储扩展 + +## 核心概念 + +### 1. 事件存储架构 + +``` +┌─────────────────────────────────────────────────────┐ +│ 事件存储 │ +├─────────────────────────────────────────────────────┤ +│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ +│ │ 流 1 │ │ 流 2 │ │ 流 3 │ │ +│ │（聚合根） │ │（聚合根） │ │（聚合根） │ │ +│ ├─────────────┤ ├─────────────┤ ├─────────────┤ │ +│ │ 事件 1 │ │ 事件 1 │ │ 事件 1 │ │ +│ │ 事件 2 │ │ 事件 2 │ │ 事件 2 │ │ +│ │ 事件 3 │ │ ... │ │ 事件 3 │ │ +│ │ ... │ │ │ │ 事件 4 │ │ +│ └─────────────┘ └─────────────┘ └─────────────┘ │ +├─────────────────────────────────────────────────────┤ +│ 全局位置：1 → 2 → 3 → 4 → 5 → 6 → ... │ +└─────────────────────────────────────────────────────┘ +``` + +### 2. 事件存储要求 + +| 要求 | 说明 | +| -------------- | ---------------------------- | +| **仅追加** | 事件不可变，只允许追加 | +| **有序** | 按流排序以及全局排序 | +| **版本化** | 乐观并发控制 | +| **订阅** | 实时事件通知 | +| **幂等性** | 安全处理重复写入 | + +## 技术对比 + +| 技术 | 最佳适用场景 | 局限性 | +| ---------------- | ---------------------- | -------------------------- | +| **EventStoreDB** | 纯事件溯源 | 单一用途 | +| **PostgreSQL** | 现有 Postgres 技术栈 | 需手动实现 | +| **Kafka** | 高吞吐量流处理 | 不适合按流查询 | +| **DynamoDB** | 无服务器、AWS 原生 | 查询能力有限 | +| **Marten** | .NET 生态系统 | 仅限 .NET | + +## 模板及详细工作示例 + +完整模板库和详细工作示例位于 `references/details.md`。当你需要具体模板时，请阅读该文件。 + +## 最佳实践 + +### 应做的 + +- **使用包含聚合类型的流 ID** —— `Order-{uuid}` +- **包含关联/因果 ID** —— 用于追踪 +- **从第一天起就对事件进行版本管理** —— 为模式演化做好规划 +- **实现幂等性** —— 使用事件 ID 进行去重 +- **适当建立索引** —— 根据查询模式而定 + +### 不应做的 + +- **不要更新或删除事件** —— 它们是不可变的事实记录 +- **不要存储较大的负载** —— 保持事件内容小巧 +- **不要跳过乐观并发控制** —— 防止数据损坏 +- **不要忽略背压** —— 处理慢速消费者 diff --git a/references/details.md b/references/details.md new file mode 100644 index 0000000..e9e6b86 --- /dev/null +++ b/references/details.md @@ -0,0 +1,356 @@ +# 事件存储设计——模板与实操示例 + +## 模板 + +### 模板 1：PostgreSQL 事件存储 Schema + +```sql +-- 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 事件存储实现 + +```python +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Optional, List +from uuid import UUID, uuid4 +import json +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=datetime.utcnow) + + +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(): + # 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 = checkpoint or from_position + + 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 用法 + +```python +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 +async def subscribe_to_all(handler, from_position: int = 0): + subscription = client.subscribe_to_all(commit_position=from_position) + async for event in subscription: + await 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 事件存储 + +```python +import boto3 +from boto3.dynamodb.conditions import Key +from datetime import datetime +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.""" + with self.table.batch_writer() as batch: + for i, event in enumerate(events): + version = (expected_version or 0) + i + 1 + item = { + 'PK': f"STREAM#{stream_id}", + 'SK': f"VERSION#{version:020d}", + 'GSI1PK': 'EVENTS', + 'GSI1SK': datetime.utcnow().isoformat(), + 'event_id': str(uuid.uuid4()), + 'stream_id': stream_id, + 'event_type': event['type'], + 'event_data': json.dumps(event['data']), + 'version': version, + 'created_at': datetime.utcnow().isoformat() + } + batch.put_item(Item=item) + return events + + def read_stream(self, stream_id: str, from_version: int = 0): + """Read events from a stream.""" + response = self.table.query( + 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 + - GSI1PK, GSI1SK for global ordering + +Capacity: On-demand or provisioned based on throughput needs +""" +```