event-store-design

name	event-store-design
description	ALWAYS use this when the request matches Event Store Design: Design and implement event stores for event-sourced systems.

Event Store Design

Selective Reading Rule

Start with:

• references/senior-master-standard.md
• references/usage-routing.md
• references/quality-checklist.md

Then load only the inherited docs, scripts, assets, or examples that match the user's actual task.

Comprehensive guide to designing event stores for event-sourced applications.

Do not use this skill when

• The task is unrelated to event store design
• You need a different domain or tool outside this scope

Instructions

• Clarify goals, constraints, and required inputs.
• Apply relevant best practices and validate outcomes.
• Provide actionable steps and verification.
• If detailed examples are required, open resources/implementation-playbook.md.

Use this skill when

• 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

-- 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()
);


 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

# 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
        }
         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

• EventStoreDB
• Marten Events
• Event Sourcing Pattern

Limitations

• Use this skill only when the task clearly matches the scope described above.
• Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
• Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.