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() }),
    {
      piiFields: "none",
      version: 2,
      migrations: [
        {
          fromVersion: 1,
          toVersion: 2,
          transform: {

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:

            map: { amountCents: (v) => Math.round(Number.parseFloat(String(v)) * 100) },
          },
        },
      ],
    },
  );

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

  // 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() }),
    {
      piiFields: { approverDisplayName: { subjectField: "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:

  // --- Query handlers ---

  // 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: {
          reason:
            "demo recipe: any signed-in user may read the event-sourced invoice state; " +
            "there is no per-user invoice ownership in this sample",
        },
      },
    },
  );

  // 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,

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):

      // Per-user cap: a single admin shouldn't fire more than 5 pay
      // operations per minute. Real production caps would be per
      // tenant + handler ("tenant+handler") to keep one admin from
      // monopolising the tenant's quota — kept simple here for the
      // sample. Demonstrates the fourth dispatcher gate (rate-limit
      // → access → validation → handler).
      rateLimit: { per: "user", limit: 5, windowSeconds: 60 },
    },
  );

  r.writeHandler(
    "invoice:archive",
    z.object({ id: z.uuid() }),
    async (event, ctx) => {
      await ctx.archiveStream(event.payload.id, { aggregateType: "showcase-invoice" });
      return { isSuccess: true as const, data: { id: event.payload.id } };
    },
    { access: { roles: ["Admin"] } },
  );

  // 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() }),

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:

      // 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"] } },
  );

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.