AliveMCP

Guide · Event-Driven & Async Patterns

MCP Server Event Sourcing — aggregate replay, snapshots, CQRS read models

Three Event Sourcing behaviours catch MCP server authors off-guard: aggregate state must be rebuilt by replaying events ordered by sequence_number, not created_at — wall-clock timestamps are unreliable across distributed systems due to clock drift, NTP corrections, and daylight-saving adjustments; if two events are written milliseconds apart from different nodes with slightly different clocks, sorting by created_at produces a different order on different replays, leading to divergent aggregate state; always use an auto-incrementing sequence_number column (or stream position) scoped per aggregate stream as the canonical ordering key; event schemas must never be mutated — create versioned event types when the shape needs to change — the event log is immutable by design; a stored event with type user.created and schema version 1 may have been written years before the current code was deployed; if you add required fields to an existing event type, all historical events of that type fail deserialization; the correct pattern is to introduce user.created.v2 as a new event type and write new code that handles both versions, then optionally run a migration to upcast old events at read time; and snapshots are optimization hints, not sources of truth — always validate a snapshot's schema version before trusting its aggregate state — a snapshot captures the aggregate state at a specific sequence number and is used to skip replaying thousands of events from the beginning; but if the snapshot was created by an older version of the aggregate's apply function, its serialized state may reflect a different data shape; always compare the snapshot's snapshot_schema_version against the current aggregate class version and fall back to full replay if they differ.

TL;DR

Use an append-only events table with columns: aggregate_id, aggregate_type, sequence_number (UNIQUE per aggregate), event_type, event_version, payload (JSONB), created_at. Rebuild state with SELECT … ORDER BY sequence_number ASC. Append with an optimistic concurrency check: INSERT … WHERE NOT EXISTS (SELECT 1 FROM events WHERE aggregate_id = $1 AND sequence_number = $2). Store snapshots separately with snapshot_schema_version — discard if mismatched. Project events into read-model tables in separate projector processes, not in the same transaction as the write.

Event store schema and appending events

The core event store is an append-only table. The sequence_number is the critical field — it must be auto-incrementing per aggregate stream (not globally), enforced with a UNIQUE constraint on (aggregate_id, sequence_number). The event_version column tracks the schema version of the event payload, which is essential for safe deserialization when the event's data shape evolves. The aggregate_type column enables efficient partitioning and querying of events by domain boundary.

-- PostgreSQL event store schema
CREATE TABLE events (
  id               BIGSERIAL PRIMARY KEY,
  aggregate_id     UUID        NOT NULL,
  aggregate_type   TEXT        NOT NULL,
  sequence_number  BIGINT      NOT NULL,
  event_type       TEXT        NOT NULL,    -- e.g. 'user.created', 'order.item.added'
  event_version    INT         NOT NULL DEFAULT 1,
  payload          JSONB       NOT NULL,
  metadata         JSONB,                  -- correlation_id, causation_id, actor_id
  created_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),

  UNIQUE (aggregate_id, sequence_number)   -- optimistic concurrency lock
);

CREATE INDEX ON events (aggregate_id, sequence_number);
CREATE INDEX ON events (aggregate_type, created_at);  -- for projector catch-up queries
import { Pool } from 'pg';
import { z } from 'zod';
import crypto from 'crypto';

const db = new Pool({ connectionString: process.env.DATABASE_URL });

// MCP tool — append an event to an aggregate's stream
server.tool(
  'append_event',
  {
    aggregate_id:    z.string().uuid(),
    aggregate_type:  z.string().min(1),
    expected_seq:    z.number().int().min(-1),  // -1 = aggregate must not exist yet
    event_type:      z.string().min(1),
    event_version:   z.number().int().min(1).default(1),
    payload:         z.record(z.unknown()),
    metadata:        z.record(z.unknown()).optional(),
  },
  async ({ aggregate_id, aggregate_type, expected_seq, event_type, event_version, payload, metadata }) => {
    const next_seq = expected_seq + 1;

    // Optimistic concurrency: unique constraint on (aggregate_id, sequence_number) prevents races
    const result = await db.query(
      `INSERT INTO events
         (aggregate_id, aggregate_type, sequence_number, event_type, event_version, payload, metadata)
       VALUES ($1, $2, $3, $4, $5, $6, $7)
       RETURNING id, sequence_number, created_at`,
      [aggregate_id, aggregate_type, next_seq, event_type, event_version, payload, metadata ?? {}]
    );

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          appended:        true,
          event_id:        result.rows[0].id,
          sequence_number: result.rows[0].sequence_number,
          created_at:      result.rows[0].created_at,
        }),
      }],
    };
  }
);

Aggregate replay and snapshot optimization

To rebuild the current state of an aggregate, load all events for that aggregate in sequence_number order and apply each event's payload to the aggregate's state using a switch on event_type. When an aggregate has accumulated thousands of events, this replay can become slow — snapshots address this by storing the aggregate's state at a specific sequence number so replay can start from there rather than from event 0. Always store a snapshot_schema_version alongside each snapshot so your code can detect when the snapshot was created by an older version of the aggregate and fall back to full replay.

-- Snapshots table
CREATE TABLE aggregate_snapshots (
  aggregate_id          UUID    NOT NULL,
  aggregate_type        TEXT    NOT NULL,
  sequence_number       BIGINT  NOT NULL,
  snapshot_schema_version INT   NOT NULL,   -- must match current aggregate class version
  state                 JSONB   NOT NULL,
  created_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),

  PRIMARY KEY (aggregate_id, sequence_number)
);

CREATE INDEX ON aggregate_snapshots (aggregate_id, sequence_number DESC);
const SNAPSHOT_THRESHOLD    = 200;  // take a snapshot every N events
const CURRENT_SCHEMA_VERSION = 3;   // increment when aggregate apply() logic changes

// MCP tool — load and replay an aggregate
server.tool(
  'load_aggregate',
  {
    aggregate_id:   z.string().uuid(),
    aggregate_type: z.string().min(1),
  },
  async ({ aggregate_id, aggregate_type }) => {
    // 1. Try to load the most recent valid snapshot
    const snapResult = await db.query(
      `SELECT sequence_number, snapshot_schema_version, state
       FROM aggregate_snapshots
       WHERE aggregate_id = $1
         AND snapshot_schema_version = $2  -- only trust current-version snapshots
       ORDER BY sequence_number DESC
       LIMIT 1`,
      [aggregate_id, CURRENT_SCHEMA_VERSION]
    );

    let state: Record<string, unknown> = {};
    let replayFromSeq = 0;

    if (snapResult.rows.length > 0) {
      state         = snapResult.rows[0].state;
      replayFromSeq = snapResult.rows[0].sequence_number;
    }

    // 2. Load events after the snapshot (or from the beginning)
    const eventsResult = await db.query(
      `SELECT sequence_number, event_type, event_version, payload
       FROM events
       WHERE aggregate_id = $1
         AND sequence_number > $2
       ORDER BY sequence_number ASC`,  // ORDER BY sequence_number, not created_at
      [aggregate_id, replayFromSeq]
    );

    // 3. Apply each event to rebuild state
    for (const row of eventsResult.rows) {
      state = applyEvent(state, row.event_type, row.event_version, row.payload);
    }

    const lastSeq = eventsResult.rows.length > 0
      ? eventsResult.rows[eventsResult.rows.length - 1].sequence_number
      : replayFromSeq;

    // 4. Take a new snapshot if we replayed many events
    if (eventsResult.rows.length >= SNAPSHOT_THRESHOLD) {
      await db.query(
        `INSERT INTO aggregate_snapshots
           (aggregate_id, aggregate_type, sequence_number, snapshot_schema_version, state)
         VALUES ($1, $2, $3, $4, $5)
         ON CONFLICT (aggregate_id, sequence_number) DO NOTHING`,
        [aggregate_id, aggregate_type, lastSeq, CURRENT_SCHEMA_VERSION, state]
      );
    }

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          aggregate_id,
          aggregate_type,
          sequence_number: lastSeq,
          state,
          replayed_events: eventsResult.rows.length,
          loaded_from_snapshot: replayFromSeq > 0,
        }, null, 2),
      }],
    };
  }
);

function applyEvent(
  state: Record<string, unknown>,
  eventType: string,
  eventVersion: number,
  payload: Record<string, unknown>
): Record<string, unknown> {
  switch (eventType) {
    case 'user.created':
      if (eventVersion === 1) {
        return { ...state, id: payload.id, email: payload.email, created_at: payload.created_at };
      }
      if (eventVersion === 2) {
        // v2 added display_name
        return { ...state, id: payload.id, email: payload.email, display_name: payload.display_name, created_at: payload.created_at };
      }
      return state;

    case 'user.email.changed':
      return { ...state, email: payload.new_email, email_updated_at: payload.changed_at };

    case 'user.deactivated':
      return { ...state, active: false, deactivated_at: payload.deactivated_at };

    default:
      // Unknown event type — return state unchanged (forward compatibility)
      return state;
  }
}

CQRS read models — projecting events into query tables

Command Query Responsibility Segregation (CQRS) separates the write model (the event store) from the read model (denormalized query tables optimized for fast reads). Projectors are processes that subscribe to the event stream and maintain read-model tables by applying each event. The key design rule for MCP server tools is to query the read-model tables directly for all read operations — never rebuild aggregate state on every query. Projectors must track their last-processed event to support catch-up from any offset after a restart or deployment.

-- Read model: user summary table for fast agent queries
CREATE TABLE users_read_model (
  user_id      UUID  PRIMARY KEY,
  email        TEXT  NOT NULL,
  display_name TEXT,
  active       BOOLEAN NOT NULL DEFAULT TRUE,
  created_at   TIMESTAMPTZ,
  updated_at   TIMESTAMPTZ
);

-- Projector checkpoint — tracks how far each projector has processed
CREATE TABLE projector_checkpoints (
  projector_name TEXT    PRIMARY KEY,
  last_event_id  BIGINT  NOT NULL DEFAULT 0
);
// Projector — maintains users_read_model from the events table
async function runUserProjector(): Promise<void> {
  const { rows: [checkpoint] } = await db.query(
    `INSERT INTO projector_checkpoints (projector_name, last_event_id)
     VALUES ('user_projector', 0)
     ON CONFLICT (projector_name) DO UPDATE SET projector_name = EXCLUDED.projector_name
     RETURNING last_event_id`
  );

  let lastEventId: bigint = BigInt(checkpoint?.last_event_id ?? 0);

  while (true) {
    const { rows } = await db.query(
      `SELECT id, aggregate_id, event_type, event_version, payload
       FROM events
       WHERE aggregate_type = 'user'
         AND id > $1
       ORDER BY id ASC
       LIMIT 500`,
      [lastEventId]
    );

    if (rows.length === 0) {
      await sleep(500);  // poll interval — replace with LISTEN/NOTIFY in production
      continue;
    }

    for (const event of rows) {
      await projectUserEvent(event.aggregate_id, event.event_type, event.event_version, event.payload);
      lastEventId = BigInt(event.id);
    }

    await db.query(
      `UPDATE projector_checkpoints SET last_event_id = $1 WHERE projector_name = 'user_projector'`,
      [lastEventId]
    );
  }
}

async function projectUserEvent(
  userId: string, type: string, version: number, payload: Record<string, unknown>
): Promise<void> {
  switch (type) {
    case 'user.created':
      await db.query(
        `INSERT INTO users_read_model (user_id, email, display_name, active, created_at, updated_at)
         VALUES ($1, $2, $3, TRUE, $4, NOW())
         ON CONFLICT (user_id) DO NOTHING`,
        [userId, payload.email, payload.display_name ?? null, payload.created_at]
      );
      break;
    case 'user.email.changed':
      await db.query(
        `UPDATE users_read_model SET email = $2, updated_at = NOW() WHERE user_id = $1`,
        [userId, payload.new_email]
      );
      break;
    case 'user.deactivated':
      await db.query(
        `UPDATE users_read_model SET active = FALSE, updated_at = NOW() WHERE user_id = $1`,
        [userId]
      );
      break;
  }
}

// MCP tool — query the read model (fast, no event replay)
server.tool(
  'query_users',
  {
    email_contains: z.string().optional(),
    active_only:    z.boolean().default(true),
    limit:          z.number().int().min(1).max(100).default(20),
    offset:         z.number().int().min(0).default(0),
  },
  async ({ email_contains, active_only, limit, offset }) => {
    const conditions: string[] = [];
    const params: unknown[] = [];
    let paramIdx = 1;

    if (active_only) {
      conditions.push(`active = TRUE`);
    }
    if (email_contains) {
      conditions.push(`email ILIKE $${paramIdx}`);
      params.push(`%${email_contains}%`);
      paramIdx++;
    }

    const where = conditions.length ? `WHERE ${conditions.join(' AND ')}` : '';
    params.push(limit, offset);

    const { rows } = await db.query(
      `SELECT user_id, email, display_name, active, created_at, updated_at
       FROM users_read_model ${where}
       ORDER BY created_at DESC
       LIMIT $${paramIdx} OFFSET $${paramIdx + 1}`,
      params
    );

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ users: rows, count: rows.length }, null, 2),
      }],
    };
  }
);

function sleep(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms));
}