Level 3 — Evolution: upcasters, snapshots, time travel

The event log is immutable — that’s the point. But your understanding of the domain keeps changing: fields get renamed, units change, aggregates grow histories with thousands of events. Level 3 is the toolkit for a log that lives for years.

Upcasters: change the shape, not the history

Say v1 of invoice-approved stored amount as a decimal string, and you now want amountCents as an integer. You never rewrite stored events — you declare a migration that upgrades old payloads on read:

  // Two domain events. "approved" is versioned — v1 stored `amount` as
  // string, v2 uses `amountCents` integer. The DECLARATIVE migration form
  // covers rename/default/map without an imperative function; the chain
  // walks the upcast on read.
  const approved = r.defineEvent(
    "invoice-approved",
    z.object({ amountCents: z.number().int(), approvedBy: z.string() }),
    { version: 2 },
  );
  r.eventMigration("invoice-approved", 1, 2, {
    rename: { amount: "amountCents" },
    map: { amountCents: (v) => Math.round(Number.parseFloat(String(v)) * 100) },
  });

  const paid = r.defineEvent("invoice-paid", z.object({ amountCents: z.number().int() }));

The declarative form covers the common cases without a hand-written function: rename old keys, default missing ones, map values. Bump version on the event, add the migration, done — every consumer sees v2, whether the stored event is v1 or v2.

When a migration needs more than the payload itself, use the function form — it can be async and read from the database:

  // Acknowledged event with an ASYNC upcaster: v1 had only the approverId,
  // v2 carries the human-readable display name too. The migration looks the
  // name up from the directory table at read time. This is Marten's
  // AsyncOnlyEventUpcaster pattern — DB enrichment without rewriting the
  // event log.
  const acknowledged = r.defineEvent(
    "invoice-acknowledged",
    z.object({ approverId: z.string(), approverDisplayName: z.string() }),
    { version: 2 },
  );
  r.eventMigration("invoice-acknowledged", 1, 2, async (payload, ctx) => {
    const p = payload as { approverId: string };
    const row = await fetchOne<{ displayName: string }>(ctx.db, approverDirectoryTable, {
      approverId: p.approverId,
    });
    return {
      approverId: p.approverId,
      approverDisplayName: row?.displayName ?? `unknown:${p.approverId}`,
    };
  });

📄 On GitHub: samples/recipes/event-sourcing/src/feature.ts

Migrations chain: v1→v2→v3 payloads walk the whole chain on read, and boot validation fails fast if a version gap has no migration.

Snapshots: long histories, constant reads

Reducing 10 events into state is instant; reducing 50,000 is not. A snapshot stores the reduced state at a version, so a read only replays the delta since:

  // Snapshot-aware fast path. Uses the latest snapshot if available and
  // only replays delta events past it. Identical state shape to
  // invoice:state — the two are interchangeable modulo the perf profile.
  r.queryHandler(
    "invoice:fast-state",
    z.object({ id: z.uuid() }),
    async (query, ctx) => {
      const result = await ctx.loadAggregateWithSnapshot<InvoiceStateRecord>(
        query.payload.id,
        (state, evt) => {
          const next: InvoiceStateRecord = { ...state };
          reduceInvoice(next as unknown as InvoiceState, evt);
          return next;
        },
        { ...initialInvoiceState } as InvoiceStateRecord,
        // Auto-snapshot policy: once 100+ delta events pile up past the last
        // snapshot, the read path persists a fresh one (best-effort). Bump
        // snapshotVersion whenever the reducer's state SHAPE changes — a
        // stale-shape snapshot is then ignored and rebuilt from the log.
        { snapshotEvery: 100, snapshotVersion: 1 },
      );
      return {
        state: result.state,
        version: result.version,
        snapshotHit: result.snapshotHit,
      };
    },
    { access: { openToAll: true } },
  );

The snapshotEvery option is the auto-policy: whenever a read replays more than that many delta events, the framework persists a fresh snapshot in the background — no cron, no lifecycle hook. snapshotVersion is your state shape generation: bump it when the reducer’s output changes, and stale snapshots are ignored and rebuilt from the log instead of poisoning reads.

You can also snapshot explicitly (e.g. from a scheduled job):

  // Persist a snapshot of the current state. In practice, features schedule
  // this from a lifecycle hook (every N events, every M minutes) — here it's
  // an explicit write-handler so the integration test can drive it.
  r.writeHandler(
    "invoice:take-snapshot",
    z.object({ id: z.uuid() }),
    async (event, ctx) => {
      // Rebuild the state from the current stream so the snapshot reflects
      // the committed event log, not an in-memory guess. Use the upcaster-
      // aware ctx.loadAggregate to stay consistent with read-time semantics.
      const events = await ctx.loadAggregate(event.payload.id);
      const state: InvoiceState = { ...initialInvoiceState };
      for (const evt of events) {
        reduceInvoice(state, evt);
      }
      const version = events.length > 0 ? (events[events.length - 1]?.version ?? 0) : 0;
      await ctx.snapshotAggregate({
        aggregateId: event.payload.id,
        aggregateType: "showcase-invoice",
        version,
        state: state as unknown as Record<string, unknown>,
      });
      return { isSuccess: true as const, data: { id: event.payload.id, snapshotVersion: version } };
    },
    { access: { roles: ["Admin"] } },
  );

Snapshots are a pure performance cache — deleting them all costs you nothing but the next read’s latency.

Time travel: asOf

Because every fact has a timestamp, “what did this invoice look like on March 31st?” is a query parameter, not a data-warehouse project:

  // Live aggregation via ctx.loadAggregate — reduces events into a state
  // snapshot. Supports asOf for point-in-time reads.
  r.queryHandler(
    "invoice:state",
    z.object({ id: z.uuid(), asOf: z.iso.datetime().optional() }),
    async (query, ctx) => {
      const events = await ctx.loadAggregate(query.payload.id, {
        ...(query.payload.asOf ? { asOf: Temporal.Instant.from(query.payload.asOf) } : {}),
      });
      const state: InvoiceState = { ...initialInvoiceState };
      for (const evt of events) {
        reduceInvoice(state, evt);
      }
      return state;
    },
    { access: { openToAll: true } },
  );

ctx.loadAggregate(id, { asOf }) replays only events up to that instant — upcasters included, so historical reads see today’s payload shapes.

When to climb to Level 4

Everything so far lives inside one aggregate. When a read model spans many — revenue per customer across all invoices — or you operate rebuilds and consumers in production, you’re ready for Level 4.