| name | event-store-design |
| description | Design and implement event stores for event-sourced systems. Use when building event sourcing infrastructure, choosing event store technologies, or implementing event persistence patterns. |
Event Store Design
Comprehensive guide to designing event stores for event-sourced applications.
When to Use This Skill
- Designing event sourcing infrastructure
- Choosing between event store technologies
- Implementing custom event stores
- Optimizing event storage and retrieval
- Setting up event store schemas
- Planning for event store scaling
Core Concepts
1. Event Store Architecture
┌─────────────────────────────────────────────────────┐
│ 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. Event Store Requirements
| Requirement | Description |
|---|
| Append-only | Events are immutable, only appends |
| Ordered | Per-stream and global ordering |
| Versioned | Optimistic concurrency control |
| Subscriptions | Real-time event notifications |
| Idempotent | Handle duplicate writes safely |
Technology Comparison
| Technology | Best For | Limitations |
|---|
| EventStoreDB | Pure event sourcing | Single-purpose |
| PostgreSQL | Existing Postgres stack | Manual implementation |
| Kafka | High-throughput streaming | Not ideal for per-stream queries |
| DynamoDB | Serverless, AWS-native | Query limitations |
| Marten | .NET ecosystems | .NET specific |
Templates
Template 1: PostgreSQL Event Store Schema
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)
);
CREATE INDEX idx_events_stream_id ON events(stream_id, version);
CREATE INDEX idx_events_global_position ON events(global_position);
CREATE INDEX idx_events_event_type ON events(event_type);
CREATE INDEX idx_events_created_at ON events(created_at);
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()
);
subscription_checkpoints (
subscription_id () ,
last_position ,
updated_at TIMESTAMPTZ NOW()
);
Template 2: Python Event Store Implementation
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() conn:
conn.transaction():
expected_version :
current = conn.fetchval(
,
stream_id
)
current = current
current != expected_version:
ConcurrencyError(
)
start_version = conn.fetchval(
,
stream_id
)
saved_events = []
i, event (events):
event.version = start_version + i
row = conn.fetchrow(
,
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[]
saved_events.append(event)
saved_events
() -> [Event]:
.pool.acquire() conn:
rows = conn.fetch(
,
stream_id, from_version, limit
)
[._row_to_event(row) row rows]
() -> [Event]:
.pool.acquire() conn:
rows = conn.fetch(
,
from_position, limit
)
[._row_to_event(row) row rows]
():
.pool.acquire() conn:
checkpoint = conn.fetchval(
,
subscription_id
)
position = checkpoint from_position
:
events = .read_all(position, batch_size)
events:
asyncio.sleep()
event events:
handler(event)
position = event.global_position
.pool.acquire() conn:
conn.execute(
,
subscription_id, position
)
() -> Event:
Event(
event_id=row[],
stream_id=row[],
event_type=row[],
data=json.loads(row[]),
metadata=json.loads(row[]),
version=row[],
global_position=row[],
created_at=row[]
)
():
Template 3: EventStoreDB Usage
from esdbclient import EventStoreDBClient, NewEvent, StreamState
import json
client = EventStoreDBClient(uri="esdb://localhost:2113?tls=false")
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
)
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
}
event events
]
():
subscription = client.subscribe_to_all(commit_position=from_position)
event subscription:
handler({
: event.,
: json.loads(event.data),
: event.stream_name,
: event.commit_position
})
():
read_stream()
Template 4: DynamoDB Event Store
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[],
: json.dumps(event[]),
: version,
: datetime.utcnow().isoformat()
}
batch.put_item(Item=item)
events
():
response = .table.query(
KeyConditionExpression=Key().eq() &
Key().gte()
)
[
{
: item[],
: json.loads(item[]),
: item[]
}
item response[]
]
Best Practices
Do's
- Use stream IDs that include aggregate type -
Order-{uuid}
- Include correlation/causation IDs - For tracing
- Version events from day one - Plan for schema evolution
- Implement idempotency - Use event IDs for deduplication
- Index appropriately - For your query patterns
Don'ts
- Don't update or delete events - They're immutable facts
- Don't store large payloads - Keep events small
- Don't skip optimistic concurrency - Prevents data corruption
- Don't ignore backpressure - Handle slow consumers
Resources
