Event Store: Storing and Replaying Events

You've learned what events are and how projections use them. But where do these events actually live? How do you store billions of events and replay them efficiently?

Enter the Event Store - a database optimized for append-only event streams.

Event Store Fundamentals

An event store is NOT a regular database. It has specific properties:

Event Store Schema

-- Simplified event store table structure
CREATE TABLE events (
    -- Global ordering across all streams
    global_position BIGSERIAL PRIMARY KEY,

    -- Stream identification (e.g., "Order-123")
    stream_id VARCHAR(255) NOT NULL,

    -- Position within the stream (0, 1, 2, ...)
    stream_position INTEGER NOT NULL,

    -- Event metadata
    event_type VARCHAR(255) NOT NULL,
    event_data JSONB NOT NULL,
    metadata JSONB,

    -- Timestamp
    created_at TIMESTAMP DEFAULT NOW(),

    -- Ensure ordering within stream
    UNIQUE (stream_id, stream_position)
);

-- Index for reading streams
CREATE INDEX idx_stream_id ON events(stream_id, stream_position);

Core Operations

Appending Events

class EventStore {
  async append(
    streamId: string,
    events: Event[],
    expectedVersion: number
  ): Promise<void> {
    // Check optimistic concurrency
    const currentVersion = await this.getStreamVersion(streamId);

    if (currentVersion !== expectedVersion) {
      throw new ConcurrencyError(
        `Expected version ${expectedVersion}, but stream is at ${currentVersion}`
      );
    }

    // Append events
    for (let i = 0; i < events.length; i++) {
      await this.db.insert('events', {
        stream_id: streamId,
        stream_position: expectedVersion + i + 1,
        event_type: events[i].type,
        event_data: events[i].data,
        metadata: events[i].metadata
      });
    }
  }
}

Reading a Stream

async readStream(streamId: string): Promise<Event[]> {
  const rows = await this.db.query(`
    SELECT event_type, event_data, stream_position
    FROM events
    WHERE stream_id = $1
    ORDER BY stream_position ASC
  `, [streamId]);

  return rows.map(row => ({
    type: row.event_type,
    data: row.event_data,
    position: row.stream_position
  }));
}

Optimistic Concurrency in Action

Timeline with Concurrent Writes:
────────────────────────────────────────────────────────────

Order-123 Stream:
Position:  0          1           2
           │          │           │
           ▼          ▼           ▼
Events:    [OrderPlaced] → [ItemAdded] → ???

User A reads at version 1, prepares: AddItem (expects version 1)
User B reads at version 1, prepares: AddItem (expects version 1)

User A writes first: SUCCESS → stream now at version 2
User B writes:       CONFLICT! Expected 1, found 2

User B must: reload stream, re-apply business logic, retry

The Snapshot Pattern

Problem: Loading an aggregate with 10,000 events is slow.

Solution: Snapshots - periodic captures of aggregate state.

Load Order-123:
  Read 10,000 events
  Apply event 1
  Apply event 2
  ...
  Apply event 10,000
  Time: 2-3 seconds

Load Order-123:
  Read snapshot @ version 9,950
  Read 50 events since snapshot
  Apply 50 events
  Time: 50ms

Implementing Snapshots

interface Snapshot<T> {
  streamId: string;
  version: number;
  state: T;
  createdAt: Date;
}

class AggregateRepository<T> {
  private snapshotInterval = 100; // Snapshot every 100 events

  async load(streamId: string): Promise<T> {
    // Try to load snapshot first
    const snapshot = await this.loadLatestSnapshot(streamId);

    // Read events since snapshot (or all events if no snapshot)
    const fromVersion = snapshot ? snapshot.version + 1 : 0;
    const events = await this.eventStore.readStream(streamId, fromVersion);

    // Start from snapshot state or empty state
    let state = snapshot ? snapshot.state : this.createEmpty();

    // Apply events
    for (const event of events) {
      state = this.apply(state, event);
    }

    return state;
  }

  async save(streamId: string, state: T, events: Event[], version: number) {
    await this.eventStore.append(streamId, events, version);

    // Create snapshot if needed
    const newVersion = version + events.length;
    if (newVersion % this.snapshotInterval === 0) {
      await this.saveSnapshot(streamId, newVersion, state);
    }
  }
}

Event Schema Evolution

Events are immutable, but your code evolves. How do you handle old event formats?

The Upcaster Pattern

// Original event (v1)
interface OrderPlacedV1 {
  orderId: string;
  items: string[];  // Just product IDs
}

// New event (v2)
interface OrderPlacedV2 {
  orderId: string;
  items: Array<{
    productId: string;
    quantity: number;  // Added!
    price: number;     // Added!
  }>;
}

// Upcaster transforms v1 → v2 when reading
class OrderPlacedUpcaster {
  upcast(event: OrderPlacedV1): OrderPlacedV2 {
    return {
      orderId: event.orderId,
      items: event.items.map(productId => ({
        productId,
        quantity: 1,        // Default value
        price: 0            // Unknown, requires lookup
      }))
    };
  }
}

Schema Evolution Strategies

Popular Event Store Implementations

Real-World: Event Sourcing at LinkedIn

Global vs Per-Stream Ordering

Global Position: Total ordering across ALL streams
─────────────────────────────────────────────────
Position 1: Order-123 → OrderPlaced
Position 2: User-456  → ProfileUpdated
Position 3: Order-123 → ItemAdded
Position 4: Order-789 → OrderPlaced
Position 5: Order-123 → PaymentReceived

Stream Position: Ordering within ONE stream
─────────────────────────────────────────────────
Order-123 Stream:
  Position 0: OrderPlaced
  Position 1: ItemAdded
  Position 2: PaymentReceived

Key Takeaways

Next up: Sagas - Coordinating distributed transactions across multiple services.