Event Sourcing in Practice: Building an Append-Only Event Store with Projections and Snapshots

Most applications store current state. Your orders table has one row per order, and that row reflects what the order looks like right now. When an order ships, you update the row. When it is cancelled, you update the row again. The previous state is gone.

This is fine for a large class of problems. It is a serious liability for the rest.

When your support team asks “what was the order total before the customer applied the coupon code?”, you cannot answer. When a billing discrepancy surfaces three months after the fact, you cannot reconstruct the sequence of changes that caused it. When you add a new feature that needs historical data, you have no history to backfill from.

Event sourcing solves this by inverting the model. Instead of storing current state and discarding the changes that produced it, you store every change as an immutable event and derive current state from the event sequence. The event log is the source of truth. Current state is a projection.

What CRUD Loses

Consider an Order aggregate. A typical CRUD lifecycle looks like this:

// What CRUD stores: the current snapshot only
interface Order {
  id: string;
  status: 'pending' | 'confirmed' | 'shipped' | 'cancelled';
  total: number;
  discountCode: string | null;
  updatedAt: Date;
}

By the time an order reaches shipped, you have no record of whether it was ever partially cancelled and reinstated, what the original total was before a price adjustment, or who changed what and when. The database reflects the outcome, not the journey.

Event sourcing keeps the journey:

type OrderEvent =
  | { type: 'OrderPlaced'; payload: { customerId: string; lineItems: LineItem[]; total: number } }
  | { type: 'CouponApplied'; payload: { code: string; discountAmount: number } }
  | { type: 'OrderConfirmed'; payload: { confirmedAt: string } }
  | { type: 'OrderShipped'; payload: { trackingNumber: string; shippedAt: string } }
  | { type: 'OrderCancelled'; payload: { reason: string } };

Every state change is a named event with a typed payload. You can replay these events in sequence to reconstruct the order at any point in time.

The Event Store

An event store is an append-only log scoped to aggregate streams. The two core operations are: append events to a stream, and read events from a stream. Nothing is updated. Nothing is deleted.

Here is the schema in Postgres:

CREATE TABLE events (
  id          BIGSERIAL PRIMARY KEY,
  stream_id   TEXT        NOT NULL,
  stream_type TEXT        NOT NULL,
  event_type  TEXT        NOT NULL,
  payload     JSONB       NOT NULL,
  metadata    JSONB       NOT NULL DEFAULT '{}',
  version     INTEGER     NOT NULL,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  UNIQUE (stream_id, version)
);

CREATE INDEX idx_events_stream ON events (stream_id, version);
CREATE INDEX idx_events_type   ON events (event_type, created_at);

The UNIQUE (stream_id, version) constraint is load-bearing. It enforces optimistic concurrency control: two concurrent writers trying to append at the same version will fail, and one will need to retry. This prevents lost updates without requiring locks.

The TypeScript interface:

interface StoredEvent {
  id: number;
  streamId: string;
  streamType: string;
  eventType: string;
  payload: Record<string, unknown>;
  metadata: Record<string, unknown>;
  version: number;
  createdAt: Date;
}

interface AppendOptions {
  expectedVersion: number; // -1 means "stream must not exist"
}

The append implementation:

import { Pool } from 'pg';

class PostgresEventStore {
  constructor(private readonly pool: Pool) {}

  async append(
    streamId: string,
    streamType: string,
    events: Array<{ type: string; payload: Record<string, unknown>; metadata?: Record<string, unknown> }>,
    options: AppendOptions
  ): Promise<void> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');

      // Verify expected version to detect concurrent writes
      const { rows } = await client.query<{ max_version: number | null }>(
        'SELECT MAX(version) AS max_version FROM events WHERE stream_id = $1',
        [streamId]
      );
      const currentVersion = rows[0].max_version ?? -1;

      if (currentVersion !== options.expectedVersion) {
        throw new ConcurrencyError(
          `Stream ${streamId}: expected version ${options.expectedVersion}, got ${currentVersion}`
        );
      }

      // Append each event with an incrementing version
      for (let i = 0; i < events.length; i++) {
        const event = events[i];
        const nextVersion = currentVersion + 1 + i;
        await client.query(
          `INSERT INTO events (stream_id, stream_type, event_type, payload, metadata, version)
           VALUES ($1, $2, $3, $4, $5, $6)`,
          [
            streamId,
            streamType,
            event.type,
            JSON.stringify(event.payload),
            JSON.stringify(event.metadata ?? {}),
            nextVersion,
          ]
        );
      }

      await client.query('COMMIT');
    } catch (err) {
      await client.query('ROLLBACK');
      throw err;
    } finally {
      client.release();
    }
  }

  async readStream(streamId: string, fromVersion = 0): Promise<StoredEvent[]> {
    const { rows } = await this.pool.query<StoredEvent>(
      `SELECT id, stream_id AS "streamId", stream_type AS "streamType",
              event_type AS "eventType", payload, metadata, version, created_at AS "createdAt"
       FROM events
       WHERE stream_id = $1 AND version >= $2
       ORDER BY version ASC`,
      [streamId, fromVersion]
    );
    return rows;
  }
}

class ConcurrencyError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'ConcurrencyError';
  }
}

Aggregates and Rehydration

With an event store in place, you load an aggregate by replaying its event stream. This is called rehydration:

type OrderState = {
  id: string;
  status: 'pending' | 'confirmed' | 'shipped' | 'cancelled';
  total: number;
  discountApplied: number;
  version: number;
};

function applyOrderEvent(state: OrderState | null, event: StoredEvent): OrderState {
  switch (event.eventType) {
    case 'OrderPlaced': {
      const p = event.payload as { total: number };
      return { id: event.streamId, status: 'pending', total: p.total, discountApplied: 0, version: event.version };
    }
    case 'CouponApplied': {
      const p = event.payload as { discountAmount: number };
      return { ...state!, total: state!.total - p.discountAmount, discountApplied: p.discountAmount, version: event.version };
    }
    case 'OrderConfirmed':
      return { ...state!, status: 'confirmed', version: event.version };
    case 'OrderShipped':
      return { ...state!, status: 'shipped', version: event.version };
    case 'OrderCancelled':
      return { ...state!, status: 'cancelled', version: event.version };
    default:
      return state!; // Unknown events are ignored, not errors
  }
}

async function loadOrder(store: PostgresEventStore, orderId: string): Promise<OrderState | null> {
  const events = await store.readStream(orderId);
  if (events.length === 0) return null;
  return events.reduce<OrderState | null>((state, event) => applyOrderEvent(state, event), null);
}

The version field on the loaded aggregate is what you pass as expectedVersion on the next append. This closes the optimistic concurrency loop.

Projections

Replaying all events every time you need current state is impractical at scale. Projections solve this by maintaining a materialized view, updated incrementally as new events arrive.

A projection subscribes to the event stream and folds events into a denormalized read model. The projection state lives in a separate table, optimized for queries rather than consistency.

// Read model: pre-computed for the dashboard query
interface OrderSummary {
  orderId: string;
  customerId: string;
  status: string;
  total: number;
  itemCount: number;
  lastUpdatedAt: Date;
}

class OrderSummaryProjection {
  constructor(private readonly pool: Pool) {}

  async handle(event: StoredEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderPlaced': {
        const p = event.payload as { customerId: string; lineItems: unknown[]; total: number };
        await this.pool.query(
          `INSERT INTO order_summaries (order_id, customer_id, status, total, item_count, last_updated_at)
           VALUES ($1, $2, 'pending', $3, $4, NOW())
           ON CONFLICT (order_id) DO NOTHING`,
          [event.streamId, p.customerId, p.total, p.lineItems.length]
        );
        break;
      }
      case 'CouponApplied': {
        const p = event.payload as { discountAmount: number };
        await this.pool.query(
          `UPDATE order_summaries
           SET total = total - $2, last_updated_at = NOW()
           WHERE order_id = $1`,
          [event.streamId, p.discountAmount]
        );
        break;
      }
      case 'OrderConfirmed':
      case 'OrderShipped':
      case 'OrderCancelled': {
        const status = event.eventType.replace('Order', '').toLowerCase();
        await this.pool.query(
          `UPDATE order_summaries SET status = $2, last_updated_at = NOW() WHERE order_id = $1`,
          [event.streamId, status]
        );
        break;
      }
    }
  }
}

Projections can be rebuilt from scratch at any time by replaying the full event log. This is one of the more powerful properties of event sourcing: you can add a new projection today and populate it retroactively with months of history, without any migration.

A projection runner polls for new events and tracks its position using a checkpoint:

class ProjectionRunner {
  private checkpoint = 0;

  constructor(
    private readonly store: PostgresEventStore,
    private readonly projection: OrderSummaryProjection,
    private readonly pool: Pool
  ) {}

  async run(): Promise<void> {
    const saved = await this.loadCheckpoint();
    this.checkpoint = saved;

    while (true) {
      const events = await this.fetchEventsBatch(this.checkpoint);
      if (events.length === 0) {
        await new Promise(resolve => setTimeout(resolve, 500));
        continue;
      }

      for (const event of events) {
        await this.projection.handle(event);
        this.checkpoint = event.id;
      }

      await this.saveCheckpoint(this.checkpoint);
    }
  }

  private async fetchEventsBatch(afterId: number): Promise<StoredEvent[]> {
    const { rows } = await this.pool.query<StoredEvent>(
      `SELECT id, stream_id AS "streamId", stream_type AS "streamType",
              event_type AS "eventType", payload, metadata, version, created_at AS "createdAt"
       FROM events WHERE id > $1 ORDER BY id ASC LIMIT 500`,
      [afterId]
    );
    return rows;
  }

  private async loadCheckpoint(): Promise<number> {
    const { rows } = await this.pool.query<{ position: number }>(
      `SELECT position FROM projection_checkpoints WHERE name = 'order_summary'`
    );
    return rows[0]?.position ?? 0;
  }

  private async saveCheckpoint(position: number): Promise<void> {
    await this.pool.query(
      `INSERT INTO projection_checkpoints (name, position) VALUES ('order_summary', $1)
       ON CONFLICT (name) DO UPDATE SET position = EXCLUDED.position`,
      [position]
    );
  }
}

Snapshots

As streams accumulate events over months or years, rehydrating an aggregate by replaying every event from the beginning becomes slow. Snapshots short-circuit this by persisting the aggregate state at a particular version.

CREATE TABLE snapshots (
  stream_id   TEXT        NOT NULL,
  version     INTEGER     NOT NULL,
  state       JSONB       NOT NULL,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  PRIMARY KEY (stream_id, version)
);

class SnapshotStore {
  constructor(private readonly pool: Pool) {}

  async save(streamId: string, version: number, state: unknown): Promise<void> {
    await this.pool.query(
      `INSERT INTO snapshots (stream_id, version, state)
       VALUES ($1, $2, $3)
       ON CONFLICT (stream_id, version) DO UPDATE SET state = EXCLUDED.state`,
      [streamId, version, JSON.stringify(state)]
    );
  }

  async load(streamId: string): Promise<{ version: number; state: unknown } | null> {
    const { rows } = await this.pool.query<{ version: number; state: unknown }>(
      `SELECT version, state FROM snapshots WHERE stream_id = $1 ORDER BY version DESC LIMIT 1`,
      [streamId]
    );
    return rows[0] ?? null;
  }
}

Loading an aggregate with snapshot support:

async function loadOrderWithSnapshot(
  store: PostgresEventStore,
  snapshots: SnapshotStore,
  orderId: string
): Promise<OrderState | null> {
  const snapshot = await snapshots.load(orderId);
  const fromVersion = snapshot ? snapshot.version + 1 : 0;
  const initialState = snapshot ? (snapshot.state as OrderState) : null;

  const events = await store.readStream(orderId, fromVersion);
  if (!snapshot && events.length === 0) return null;

  return events.reduce<OrderState | null>(
    (state, event) => applyOrderEvent(state, event),
    initialState
  );
}

A common strategy: snapshot every N events (50 or 100 is typical), or whenever the event count on load exceeds a threshold. Keep only the last one or two snapshots per stream. They are a performance optimization, not a source of truth, so you do not need to preserve every historical snapshot.

CRUD vs Event Sourcing

Dimension	CRUD	Event Sourcing
Storage model	Current state only	Full event history
Audit trail	Requires extra effort (change tables, triggers)	Built-in, always accurate
Read performance	Simple indexed queries	Requires projections for query performance
Write performance	Single row update	Append to log, eventually update projections
Schema changes	ALTER TABLE, migrations	Event upcasting or new event versions
Debugging	Query current state	Replay events to any point in time
Backfill new features	Historical data often lost	Replay events to populate new projections
Operational complexity	Low	High: projections, snapshots, ordering guarantees
Consistency model	Immediate	Eventual (projection lag)
Temporal queries	Hard or impossible	Native

Production Tradeoffs

Schema Evolution

This is the hardest long-term problem in event sourcing. You cannot alter events after they are appended. If you rename a field in OrderPlaced, events already in the log still use the old name.

Two approaches:

Upcasting: Transform old event formats on read before passing them to the aggregate. Keep a registry of transforms keyed on event type and version. This works but adds complexity to every read path.

New event versions: When the shape changes significantly, introduce OrderPlacedV2 and let the old events remain as-is. Apply events accumulate in both versions until all streams have naturally migrated.

The safest rule: treat every event payload field as public API from day one. Adding optional fields is safe. Removing or renaming fields requires a migration strategy.

Eventual Consistency

Projections are eventually consistent with the event log. The lag is usually milliseconds on a healthy system, but it is not zero. Callers who write a command and immediately query a projection may see stale state.

Common mitigations:

• Return the new version number from the command handler and let the client poll the read model until it reflects that version.
• Keep command handlers and projection reads in the same process for low-latency paths, bypassing the projection entirely when freshness is critical.
• Accept the lag and design the UI around it (optimistic updates, “processing” states).

Debugging

Event sourcing actually makes some debugging easier: you have a complete, ordered history of everything that happened. But this only helps if the events are meaningful. Anemic events like StateChanged with a JSON diff are as useless as change tracking in a CRUD system.

Name events after business facts, not technical operations. FraudCheckPassed, not StatusUpdated. ShippingAddressOverridden, not OrderModified. The event log should read like a business audit trail, not a database changelog.

For debugging projection bugs specifically, the ability to drop and rebuild a projection from the event log is extremely useful. This lets you ship a projection fix and backfill it in minutes rather than writing complex data repair scripts.

Ordering Guarantees

Within a single stream, ordering is guaranteed by the version sequence. Across streams, events are ordered only by insertion time (id in the global events table), which provides a rough global order but not strict causal ordering across aggregates.

If your projections join data from multiple streams, be careful about race conditions where the events from one stream arrive out of order relative to the other. Handling this typically requires idempotent projection logic or explicit causality tracking via correlation IDs in event metadata.

When Not to Use Event Sourcing

Event sourcing adds real complexity. The operational surface area is larger: you are maintaining an event log, multiple projections, checkpoints, and a snapshot store. Projection rebuild times grow with data volume.

It earns its cost when:

• The business needs a full audit trail and reconstructing it after the fact would be impractical.
• You need temporal queries: “what did this record look like on a specific date?”
• Multiple downstream systems need to react to state changes in different ways.
• You need the ability to add new projections retroactively without losing history.

It is probably overkill when:

• Your domain is simple CRUD with no audit requirements.
• Your team is small and operational complexity has a steep cost.
• You can achieve audit requirements more cheaply with a change log or soft deletes.

Closing

Event sourcing is not a design pattern you reach for by default. The event log, projections, snapshots, checkpoint management, and schema evolution strategies are a meaningful tax on every future change. But for domains where history is first-class, where the business legitimately needs to know what happened and when, and where you want the flexibility to build new read models from existing data, that tax pays for itself.

The core implementation is not complicated: an append-only table, a concurrency check on insert, a reduce over events to rebuild state. The complexity lives in the operational concerns and in keeping your event schema clean as the domain evolves. Start there before adding projections, then add projections before adding snapshots. Add complexity only when the performance data demands it.