Event Sourcing Production Anti-Patterns: Schema Evolution, Snapshotting, and Event Store Scaling

// Version 1: Original event
interface OrderPlacedV1 {
  type: 'OrderPlaced'
  orderId: string
  customerId: string
  totalAmount: number
}

// Version 2: Added shippingAddress (optional for backward compat)
interface OrderPlacedV2 {
  type: 'OrderPlaced'
  orderId: string
  customerId: string
  totalAmount: number
  shippingAddress?: {
    street: string
    city: string
    zipCode: string
    country: string
  }
  currency?: string // default: 'USD'
}

// Consumer uses tolerant reader pattern
function handleOrderPlaced(event: OrderPlacedV1 | OrderPlacedV2) {
  const currency = 'currency' in event ? event.currency : 'USD'
  const address = 'shippingAddress' in event ? event.shippingAddress : null // fetch from customer profile as fallback
}

from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class EventEnvelope:
    stream_id: str
    event_type: str
    version: int
    data: dict
    metadata: dict

class UpcasterChain:
    """Chains upcasters to transform events from any version to the latest."""

    def __init__(self):
        self._upcasters: dict[tuple[str, int], Callable] = {}

    def register(self, event_type: str, from_version: int,
                 upcaster: Callable[[dict], dict]):
        self._upcasters[(event_type, from_version)] = upcaster

    def upcast(self, envelope: EventEnvelope,
               target_version: int) -> EventEnvelope:
        current = envelope
        while current.version < target_version:
            key = (current.event_type, current.version)
            if key not in self._upcasters:
                raise ValueError(
                    f"No upcaster for {current.event_type} "
                    f"v{current.version} -> v{current.version + 1}"
                )
            new_data = self._upcasters[key](current.data)
            current = EventEnvelope(
                stream_id=current.stream_id,
                event_type=current.event_type,
                version=current.version + 1,
                data=new_data,
                metadata=current.metadata,
            )
        return current

# Register upcasters
chain = UpcasterChain()

# V1 -> V2: split 'name' into 'firstName' and 'lastName'
chain.register('CustomerRegistered', 1, lambda d: {
    **d,
    'firstName': d['name'].split(' ')[0],
    'lastName': ' '.join(d['name'].split(' ')[1:]),
})

# V2 -> V3: add 'tier' field with default
chain.register('CustomerRegistered', 2, lambda d: {
    **d,
    'tier': 'standard',
})

// Instead of OrderPlacedV3 with breaking changes,
// create a completely new event type
interface OrderSubmitted {
  type: 'OrderSubmitted' // new name signals new semantics
  orderId: string
  customer: {
    id: string
    tier: 'standard' | 'premium' | 'enterprise'
  }
  lineItems: Array<{
    productId: string
    quantity: number
    unitPrice: number
    discount: number
  }>
  pricing: {
    subtotal: number
    tax: number
    shipping: number
    total: number
    currency: string
  }
  submittedAt: string
}

-- WARNING: This is destructive. Take full backups first.
-- Step 1: Create a new events table
CREATE TABLE events_v2 (
    global_position BIGSERIAL PRIMARY KEY,
    stream_id UUID NOT NULL,
    stream_position INTEGER NOT NULL,
    event_type VARCHAR(256) NOT NULL,
    data JSONB NOT NULL,
    metadata JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE (stream_id, stream_position)
);

-- Step 2: Migrate with transformation
INSERT INTO events_v2 (stream_id, stream_position, event_type, data, metadata, created_at)
SELECT
    stream_id,
    stream_position,
    event_type,
    CASE
        WHEN event_type = 'CustomerRegistered' AND data->>'version' = '1'
        THEN jsonb_build_object(
            'version', '3',
            'firstName', split_part(data->>'name', ' ', 1),
            'lastName', regexp_replace(data->>'name', '^[^ ]+ ', ''),
            'email', data->>'email',
            'tier', 'standard'
        )
        ELSE data
    END,
    metadata,
    created_at
FROM events
ORDER BY global_position;

-- Step 3: Swap tables (during maintenance window)
ALTER TABLE events RENAME TO events_deprecated;
ALTER TABLE events_v2 RENAME TO events;

from dataclasses import dataclass, field
from typing import Optional
import json

@dataclass
class Snapshot:
    stream_id: str
    version: int  # event version at snapshot time
    state: dict
    created_at: str

class SnapshotStore:
    """PostgreSQL-backed snapshot store."""

    def __init__(self, conn):
        self._conn = conn

    async def save_snapshot(self, snapshot: Snapshot):
        await self._conn.execute("""
            INSERT INTO snapshots (stream_id, version, state, created_at)
            VALUES ($1, $2, $3, $4)
            ON CONFLICT (stream_id) DO UPDATE
            SET version = $4, state = $3, created_at = $4
        """, snapshot.stream_id, snapshot.version,
             json.dumps(snapshot.state), snapshot.created_at)

    async def load_snapshot(self, stream_id: str) -> Optional[Snapshot]:
        row = await self._conn.fetchrow("""
            SELECT stream_id, version, state, created_at
            FROM snapshots WHERE stream_id = $1
        """, stream_id)
        if row:
            return Snapshot(
                stream_id=row['stream_id'],
                version=row['version'],
                state=json.loads(row['state']),
                created_at=row['created_at'],
            )
        return None

class AggregateRepository:
    SNAPSHOT_INTERVAL = 100  # snapshot every 100 events

    def __init__(self, event_store, snapshot_store):
        self._event_store = event_store
        self._snapshot_store = snapshot_store

    async def load(self, stream_id: str):
        snapshot = await self._snapshot_store.load_snapshot(stream_id)

        if snapshot:
            aggregate = self._rebuild_from_snapshot(snapshot)
            start_version = snapshot.version + 1
        else:
            aggregate = self._create_empty()
            start_version = 0

        events = await self._event_store.read_stream(
            stream_id, from_version=start_version
        )
        for event in events:
            aggregate.apply(event)

        return aggregate

    async def save(self, aggregate):
        new_events = aggregate.uncommitted_events
        current_version = aggregate.version

        await self._event_store.append(
            aggregate.stream_id, new_events, current_version
        )

        # Create snapshot if threshold reached
        if current_version % self.SNAPSHOT_INTERVAL == 0:
            await self._snapshot_store.save_snapshot(Snapshot(
                stream_id=aggregate.stream_id,
                version=current_version,
                state=aggregate.to_dict(),
                created_at=self._now(),
            ))

import asyncio
from dataclasses import dataclass

@dataclass
class ProjectionRebuildConfig:
    batch_size: int = 1000
    max_concurrent_batches: int = 5
    checkpoint_interval: int = 10_000
    progress_log_interval: int = 50_000

class ProjectionRebuilder:
    """Optimized projection rebuild with batching and checkpointing."""

    def __init__(self, event_store, projection_handler, checkpoint_store,
                 config: ProjectionRebuildConfig = None):
        self._event_store = event_store
        self._handler = projection_handler
        self._checkpoint_store = checkpoint_store
        self._config = config or ProjectionRebuildConfig()

    async def rebuild(self, projection_name: str):
        # Resume from last checkpoint if rebuild was interrupted
        last_position = await self._checkpoint_store.get(
            f"rebuild:{projection_name}"
        )
        start_position = last_position + 1 if last_position else 0

        total_processed = 0
        position = start_position

        while True:
            events = await self._event_store.read_all(
                from_position=position,
                count=self._config.batch_size,
            )
            if not events:
                break

            # Process batch
            await self._handler.handle_batch(events)
            position = events[-1].global_position + 1
            total_processed += len(events)

            # Checkpoint periodically
            if total_processed % self._config.checkpoint_interval == 0:
                await self._checkpoint_store.save(
                    f"rebuild:{projection_name}",
                    events[-1].global_position,
                )

            if total_processed % self._config.progress_log_interval == 0:
                print(
                    f"Rebuild progress: {total_processed} events processed, "
                    f"position: {position}"
                )

        print(f"Rebuild complete: {total_processed} total events processed")

// BAD: Property sourcing - no business intent
const badEvents = [
  { type: 'FieldUpdated', field: 'status', value: 'suspended' },
  { type: 'FieldUpdated', field: 'reason', value: 'payment_failed' },
]

// GOOD: Domain events with business meaning
const goodEvents = [
  {
    type: 'AccountSuspendedDueToPaymentFailure',
    accountId: 'ACC-123',
    failedPaymentId: 'PAY-456',
    suspendedAt: '2026-03-07T10:30:00Z',
    retryScheduledAt: '2026-03-10T10:30:00Z',
  },
]

-- BAD: Blind insert (duplicates on redelivery)
INSERT INTO order_summary (order_id, status, total)
VALUES ('ORD-001', 'placed', 150.00);

-- GOOD: Idempotent upsert
INSERT INTO order_summary (order_id, status, total, last_event_position)
VALUES ('ORD-001', 'placed', 150.00, 42)
ON CONFLICT (order_id) DO UPDATE
SET status = EXCLUDED.status,
    total = EXCLUDED.total,
    last_event_position = EXCLUDED.last_event_position
WHERE order_summary.last_event_position < EXCLUDED.last_event_position;

+------------------+     +-----------------+     +------------------+
|   API Gateway    |---->| Command Handler |---->| Domain Aggregate |
+------------------+     +-----------------+     +--------+---------+
                                                          |
                              +---------------------------+
                              | Domain Events
                              v
                    +--------------------+
                    |    Event Store     |
                    | (append-only log)  |
                    +--------+-----------+
                             |
              +--------------+--------------+
              |              |              |
              v              v              v
     +--------+---+  +------+------+  +----+--------+
     | Projection |  | Projection  |  | Projection  |
     | Handler A  |  | Handler B   |  | Handler C   |
     +--------+---+  +------+------+  +----+--------+
              |              |              |
              v              v              v
     +--------+---+  +------+------+  +----+--------+
     | PostgreSQL |  |Elasticsearch|  |    Redis     |
     | Read Model |  |Search Index |  |   Cache      |
     +------------+  +-------------+  +-------------+

Event Schema Evolution Pipeline:

  Event Store (raw bytes)
       |
       v
  +----------+     +-----------+     +-----------+
  | Upcaster |---->| Upcaster  |---->| Upcaster  |
  |  V1->V2  |     |  V2->V3   |     |  V3->V4   |
  +----------+     +-----------+     +-----------+
                                          |
                                          v
                                  +---------------+
                                  | Current Event |
                                  |  (Version 4)  |
                                  +---------------+
                                          |
                              +-----------+-----------+
                              |                       |
                              v                       v
                     +--------+------+      +---------+-----+
                     | Command Side  |      | Projection    |
                     | (Aggregate)   |      | (Read Model)  |
                     +---------------+      +---------------+