distributed-systems-event-sourcing

Event sourcing architecture, event design, CQRS, aggregates, projections, snapshots, and production implementations

Event Sourcing

Scope: Event sourcing fundamentals, event design, CQRS patterns, aggregates, projections, snapshots, saga patterns, temporal queries Lines: ~450 Last Updated: 2025-10-27 Format Version: 1.0 (Atomic)

When to Use This Skill

Activate this skill when:

Building event-driven systems
Implementing audit trails
Designing CQRS architectures
Creating temporal query capabilities
Working with distributed transactions
Building saga patterns
Implementing domain-driven design
Creating analytics from historical data
Debugging complex state transitions
Building collaborative systems

Core Concepts

Event Sourcing Overview

Principle: Store all changes as immutable events instead of current state

Traditional:
  users = {id: 1, name: "Alice", status: "active"}

Event Sourced:
  events = [
    UserCreated(id=1, name="Alice"),
    UserActivated(id=1)
  ]
  Current state = reduce(events, initial_state)

Key benefits:

Complete audit trail
Temporal queries (state at any point in time)
Event replay for debugging
Multiple projections from same events
Natural fit for event-driven architectures

Trade-offs:

Higher complexity
Storage growth (mitigated by snapshots)
Eventual consistency
Learning curve
Event versioning challenges

Event Design

Event naming: Past tense, specific facts

# Good: Past tense
UserRegistered
OrderPlaced
PaymentProcessed
EmailSent

# Bad: Commands or present tense
RegisterUser  # Command, not event
PlaceOrder    # Command
UpdateUser    # Too vague

Event structure:

{
  "eventId": "uuid",
  "eventType": "OrderPlaced",
  "aggregateId": "order-123",
  "aggregateType": "Order",
  "version": 1,
  "timestamp": "2025-10-27T10:30:00Z",
  "data": {
    "orderId": "order-123",
    "customerId": "customer-456",
    "items": [...],
    "totalAmount": 99.99
  },
  "metadata": {
    "causationId": "command-789",
    "correlationId": "request-012",
    "userId": "admin-1"
  }
}

Required fields: eventId, eventType, aggregateId, version, timestamp, data

Recommended fields: aggregateType, metadata (causationId, correlationId, userId)

Aggregates

Aggregate Pattern

Aggregate: Consistency boundary that enforces invariants and produces events

class BankAccount:
    """Aggregate enforcing account invariants"""

    def __init__(self, account_id):
        self.account_id = account_id
        self.balance = 0
        self.status = "pending"
        self.version = 0
        self._uncommitted_events = []

    def deposit(self, amount):
        """Command: Deposit money"""
        if self.status != "active":
            raise AccountNotActiveError()
        if amount <= 0:
            raise InvalidAmountError()

        # Create and apply event
        event = MoneyDeposited(
            accountId=self.account_id,
            amount=amount,
            balanceAfter=self.balance + amount
        )
        self._apply_event(event)
        self._uncommitted_events.append(event)

    def _apply_event(self, event):
        """Event handler: Update state (NO business logic)"""
        if isinstance(event, AccountOpened):
            self.balance = event.initial_deposit
            self.status = "active"
        elif isinstance(event, MoneyDeposited):
            self.balance = event.balance_after

        self.version += 1

Aggregate boundaries:

Small, focused aggregates (< 100 events typical)
One aggregate per consistency boundary
Reference other aggregates by ID only
Use eventual consistency between aggregates

Event Store

Event Store Responsibilities

Core operations:

Append events (with optimistic concurrency)
Read stream (all events for aggregate)
Stream events (for projections)
Save/load snapshots

Schema (PostgreSQL):

CREATE TABLE events (
    id BIGSERIAL PRIMARY KEY,
    event_id UUID NOT NULL UNIQUE,
    event_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    aggregate_type VARCHAR(100) NOT NULL,
    version INTEGER NOT NULL,
    data JSONB NOT NULL,
    metadata JSONB,
    timestamp TIMESTAMPTZ NOT NULL,
    UNIQUE (aggregate_id, version)
);

CREATE TABLE snapshots (
    aggregate_id VARCHAR(255) NOT NULL,
    version INTEGER NOT NULL,
    data JSONB NOT NULL,
    timestamp TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (aggregate_id, version)
);

Optimistic concurrency:

def append_events(aggregate_id, expected_version, events):
    """Append with concurrency control"""
    # Lock and check version
    current_version = get_stream_version(aggregate_id)

    if current_version != expected_version:
        raise ConcurrencyException(
            f"Expected {expected_version}, got {current_version}"
        )

    # Append events
    for i, event in enumerate(events):
        version = expected_version + i + 1
        insert_event(aggregate_id, version, event)

    update_stream_version(aggregate_id, expected_version + len(events))

Projections (Read Models)

Projection Pattern

Projection: Read model built by processing events

Event Stream → Projection Handler → Read Model

Example:

class UserProjection:
    """Build user view from events"""

    def handle(self, event):
        if event.event_type == "UserRegistered":
            db.execute(
                "INSERT INTO user_view (user_id, email, status) "
                "VALUES (%s, %s, %s)",
                (event.data['userId'], event.data['email'], 'pending')
            )

        elif event.event_type == "UserActivated":
            db.execute(
                "UPDATE user_view SET status = 'active' WHERE user_id = %s",
                (event.data['userId'],)
            )

Key properties:

Idempotent (processing same event twice = same result)
Eventually consistent with event store
Can be rebuilt from events
Optimized for queries

Checkpoint management:

# Track last processed event
checkpoint = get_checkpoint("user_view")

# Process new events
for event in stream_events(from_position=checkpoint):
    projection.handle(event)
    update_checkpoint("user_view", event.position)

CQRS Integration

Command Query Responsibility Segregation

Principle: Separate write model (commands) from read model (queries)

Commands → Aggregate → Events → Event Store
                         ↓
                    Event Bus
                         ↓
            ┌────────────┼────────────┐
            ▼            ▼            ▼
      Projection 1  Projection 2  Projection 3
            ↓            ↓            ↓
      Read Model 1  Read Model 2  Read Model 3
            ↑            ↑            ↑
         Queries      Queries      Queries

Benefits:

Scale reads and writes independently
Multiple read models from same events
Optimize each read model for its queries
Flexible schema evolution

Snapshots

Snapshot Strategy

Purpose: Optimize aggregate loading by saving state periodically

# Without snapshot: Load 10,000 events
aggregate = load_aggregate("account-123")  # Slow!

# With snapshot: Load snapshot + recent events
snapshot = load_snapshot("account-123")
recent_events = load_events_after(snapshot.version)
aggregate = apply_events(snapshot.state, recent_events)  # Fast!

Snapshot frequency:

Every N events: Snapshot every 100 events
Time-based: Snapshot aggregates not snapshotted in 7 days
On-demand: Create snapshot if loading takes > 100ms

Snapshot storage:

def save_snapshot(aggregate, event_store):
    snapshot = {
        'aggregateId': aggregate.id,
        'version': aggregate.version,
        'data': {
            'balance': aggregate.balance,
            'status': aggregate.status
        }
    }
    event_store.save_snapshot(snapshot)

Saga Pattern

Long-Running Processes

Saga: Coordinate process across multiple aggregates/services

Example: Order fulfillment

1. Order submitted
2. Reserve inventory → If fails: Cancel order
3. Process payment   → If fails: Release inventory, cancel order
4. Ship order        → If fails: Refund payment, release inventory, cancel
5. Complete order

Implementation:

class OrderFulfillmentSaga:
    """Coordinate order fulfillment"""

    def handle_event(self, event):
        if event.type == "OrderSubmitted":
            # Step 1: Reserve inventory
            send_command(ReserveInventory(order_id, items))

        elif event.type == "InventoryReserved":
            # Step 2: Process payment
            send_command(ProcessPayment(order_id, amount))

        elif event.type == "PaymentFailed":
            # Compensate: Release inventory
            send_command(ReleaseInventory(order_id, items))
            send_command(CancelOrder(order_id))

Saga state persistence required for reliability

Event Versioning

Schema Evolution

Strategy 1: Weak schema (additive only)

# V1
{"eventType": "UserRegistered", "userId": "123", "email": "..."}

# V2: Add optional field
{"eventType": "UserRegistered", "userId": "123", "email": "...", "source": "mobile"}

# Handler handles both
def handle(event):
    source = event.get('source', 'unknown')  # Default for old events

Strategy 2: Explicit versioning

# Event includes version
{"eventType": "UserRegistered", "version": 2, ...}

# Handler routes by version
if event['version'] == 1:
    handle_v1(event)
elif event['version'] == 2:
    handle_v2(event)

Strategy 3: Upcasting

def upcast(event):
    """Convert old events to latest schema"""
    if event['version'] == 1:
        event['data']['source'] = 'unknown'
        event['version'] = 2
    return event

Temporal Queries

Query Historical State

Temporal query: State at specific point in time

def get_aggregate_at_time(aggregate_id, timestamp, event_store):
    """Rebuild aggregate as it was at timestamp"""
    events = event_store.get_events_until(aggregate_id, timestamp)

    aggregate = BankAccount(aggregate_id)
    for event in events:
        aggregate._apply_event(event)

    return aggregate

# Query yesterday's balance
account_yesterday = get_aggregate_at_time("account-123", yesterday, event_store)
print(f"Balance yesterday: {account_yesterday.balance}")

Use cases:

Audit investigations
Regulatory reporting
"What if" analysis
Debugging production issues

Production Implementations

EventStoreDB

Purpose-built event store:

docker run -d -p 2113:2113 eventstore/eventstore:latest

from esdbclient import EventStoreDBClient

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

# Append events
client.append_to_stream(
    stream_name="account-123",
    events=[NewEvent(type="MoneyDeposited", data={...})]
)

# Read stream
events = client.get_stream("account-123")

PostgreSQL Event Store

Advantages: No new infrastructure, familiar tooling, ACID

class PostgresEventStore:
    def append_events(self, aggregate_id, expected_version, events):
        # Optimistic concurrency control
        # See REFERENCE.md for full implementation
        pass

Kafka as Event Store

For distributed systems:

producer.send(
    'user-events',
    key=aggregate_id,
    value=event_data
)

Anti-Patterns

Common Mistakes

1. Large events: Keep events focused

# Bad: Kitchen sink
OrderUpdated(changes={...100 fields...})

# Good: Specific events
ItemAdded(productId, quantity, price)

2. Missing idempotency:

# Bad
def handle(event):
    balance += event.amount  # Applying twice doubles amount!

# Good
def handle(event):
    if event.id not in processed:
        balance += event.amount
        processed.add(event.id)

3. Business logic in projections:

# Bad: Logic in projection
def handle(event):
    if event.total > 1000:
        discount = calculate_discount(event.total)  # Business logic!

# Good: Logic in aggregate, event contains result
# Event: OrderPlaced(total=1000, discount=100)

4. No versioning strategy: Plan for schema evolution from day one

5. Synchronous projections: Projections should be asynchronous

Performance Optimization

Key Optimizations

1. Snapshots: Massive speedup for large aggregates

2. Indexing:

CREATE INDEX idx_events_aggregate ON events(aggregate_id, version);
CREATE INDEX idx_events_id ON events(id);
CREATE INDEX idx_events_type ON events(event_type);

3. Projection batching:

# Batch commit every 100 events
batch = []
for event in events:
    projection.handle(event)
    batch.append(event)
    if len(batch) >= 100:
        db.commit()
        batch = []

4. Read model denormalization: Optimize for queries

5. Caching: Cache frequently-accessed aggregates

Testing Event-Sourced Systems

Test Strategies

1. Aggregate behavior:

def test_withdraw_reduces_balance():
    account = BankAccount.open("account-123", "Alice", 1000)
    account.withdraw(200)
    assert account.balance == 800

2. Event production:

def test_withdraw_produces_event():
    account.withdraw(200)
    events = account.get_uncommitted_events()
    assert events[-1].event_type == "MoneyWithdrawn"

3. Event replay:

def test_rebuild_from_events():
    events = [
        AccountOpened(accountId="123", initialDeposit=1000),
        MoneyDeposited(accountId="123", amount=500)
    ]
    account = BankAccount.from_events("123", events)
    assert account.balance == 1500

4. Projection correctness:

def test_projection_handles_event():
    projection.handle(UserRegistered(userId="123", email="alice@example.com"))
    user = db.query("SELECT * FROM user_view WHERE user_id = '123'")
    assert user['email'] == "alice@example.com"

Level 3: Resources

Reference Documentation

REFERENCE.md (2,378 lines): Comprehensive guide covering:

Event sourcing fundamentals
Event design patterns (naming, structure, granularity)
Aggregate design and lifecycle
Event store architecture (PostgreSQL schema, operations)
Projections and read models
Snapshot strategies (frequency, storage, loading)
CQRS integration patterns
Saga patterns for distributed transactions
Event versioning and schema evolution
Temporal queries and event replay
Production implementations (EventStoreDB, PostgreSQL, Kafka, Axon, Marten)
Common anti-patterns and solutions
Performance optimization techniques
Testing strategies (aggregates, projections, idempotency)

Location: skills/distributed-systems/event-sourcing/resources/REFERENCE.md

Executable Scripts

1. validate_events.py (16K, 575 lines)

Validates event schemas and design patterns.

# Validate event file
./validate_events.py --file events.json

# Validate directory
./validate_events.py --directory ./events --json

# Validate from event store
./validate_events.py --stream postgres://localhost/events --aggregate-id account-123

Features:

Event naming validation (past tense)
Schema validation (required fields)
Idempotency checks
Payload size limits
Anti-pattern detection
JSON output support

2. replay_events.py (17K, 600 lines)

Replay events for debugging, temporal queries, and projection rebuilds.

# Debug aggregate
./replay_events.py --source postgres://localhost/events --aggregate account-123 --debug

# Rebuild projection
./replay_events.py --source postgres://localhost/events --projection user_view

# Temporal query
./replay_events.py --source postgres://localhost/events --temporal account-123 --until "2025-10-01"

# Event store statistics
./replay_events.py --source postgres://localhost/events --stats

Features:

Step-by-step aggregate debugging
Projection rebuilding
Temporal queries
Event comparison
Statistics analysis
File and database sources

3. benchmark_eventstore.py (22K, 820 lines)

Benchmark event store performance.

# Write benchmark
./benchmark_eventstore.py --store postgres://localhost/events --write --events 10000

# Read benchmark
./benchmark_eventstore.py --store postgres://localhost/events --read --aggregates 100

# Concurrent writes
./benchmark_eventstore.py --store postgres://localhost/events --concurrent 10

# Full benchmark suite
./benchmark_eventstore.py --store postgres://localhost/events --full --json

Features:

Single and batch writes
Aggregate reads
Concurrent operations
Event streaming
Latency percentiles (p95, p99)
Throughput measurement
JSON output

Location: skills/distributed-systems/event-sourcing/resources/scripts/

Production-Ready Examples

1. Bank Account Aggregate (Python) - Complete aggregate implementation with command handling, event production, and state reconstruction

2. PostgreSQL Event Store (Python) - Full event store with optimistic concurrency, snapshots, and connection pooling

3. User View Projection (Python) - Projection with checkpoint management and continuous processing

4. Order Aggregate (TypeScript) - Type-safe aggregate with strong typing and invariant enforcement

5. Event Schemas (JSON Schema) - Comprehensive event schema definitions with validation

6. Docker Compose Setup - Complete event sourcing stack (PostgreSQL, EventStoreDB, Kafka, Redis)

7. SQL Initialization - Event store database schema with indexes and triggers

8. Order Fulfillment Saga (Python) - Saga pattern with compensating transactions and state management

Location: skills/distributed-systems/event-sourcing/resources/examples/

All examples are:

Production-ready with error handling
Fully documented with docstrings
Executable with example usage
Following best practices
Demonstrating key patterns

Quick Decision Tree

Need to track state changes?
  ├─ Audit trail critical? → Event sourcing
  ├─ Temporal queries needed? → Event sourcing
  ├─ Multiple read models? → Event sourcing + CQRS
  ├─ Simple CRUD? → Traditional persistence
  └─ Team unfamiliar? → Start simple, migrate later

Implementing event sourcing?
  ├─ Design events (past tense, focused)
  ├─ Create aggregates (enforce invariants)
  ├─ Set up event store (PostgreSQL/EventStoreDB)
  ├─ Build projections (asynchronous)
  ├─ Add snapshots (for performance)
  ├─ Plan versioning (additive changes)
  └─ Test thoroughly (aggregates, replay, projections)

References

Books:

"Domain-Driven Design" by Eric Evans
"Implementing Domain-Driven Design" by Vaughn Vernon
"Versioning in an Event Sourced System" by Greg Young

Online:

Martin Fowler: Event Sourcing pattern
Greg Young: CQRS and Event Sourcing talks
EventStoreDB documentation
Microsoft CQRS Journey

Tools:

EventStoreDB: https://www.eventstore.com/
Marten (PostgreSQL): https://martendb.io/
Axon Framework (Java): https://axoniq.io/

Last Updated: 2025-10-27