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:

  // 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: {
            rename: { amount: "amountCents" },

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:

          },
        },
      ],
    },
  );

  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" } },
      version: 2,

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

    schema: z.object({ id: z.uuid(), asOf: z.iso.datetime().optional() }),
    handler: async (query, ctx) => {
      const events = await ctx.events.load(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({
    name: "invoice:fast-state",
    schema: z.object({ id: z.uuid() }),
    handler: async (query, ctx) => {
      const result = await ctx.events.loadWithSnapshot<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

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

    // → access → validation → handler).
    rateLimit: { per: "user", limit: 5, windowSeconds: 60 },
  });

  r.writeHandler({
    name: "invoice:archive",
    schema: z.object({ id: z.uuid() }),
    handler: async (event, ctx) => {
      await ctx.events.archive(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({
    name: "invoice:take-snapshot",
    schema: z.object({ id: z.uuid() }),
    handler: 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.events.load to stay consistent with read-time semantics.
      const events = await ctx.events.load(event.payload.id);
      const state: InvoiceState = { ...initialInvoiceState };

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:

      const version = events.length > 0 ? (events[events.length - 1]?.version ?? 0) : 0;
      await ctx.events.snapshot({
        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"] },
  });

  // --- Query handlers ---

  // Live aggregation via ctx.events.load — reduces events into a state
  // snapshot. Supports asOf for point-in-time reads.
  r.queryHandler({

ctx.events.load(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.