Level 2 — Events & projections

At Level 1, “the invoice was approved” would be update({ status: "approved" }) — the fact gets flattened into a field. Level 2 stores the fact itself. Now is the moment for the two words this model is built on:

• The event log is the source of truth: an append-only record of every fact (kumiko_events in Postgres — no Kafka, no extra infrastructure).
• Your tables are projections: read models derived from those facts, disposable and rebuildable.

Level 1 used both without telling you — r.crud appends task.created/task.updated events and maintains the entity table as a projection. Level 2 is you doing the same thing with your own vocabulary.

All snippets on this page come from the event-sourcing recipe, an invoice domain with approve and pay verbs.

Define events

An event has a name and a Zod-validated payload shape:

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

(Ignore version and r.eventMigration for now — they are Level 3. A first event needs neither.)

Append events in write handlers

A write handler validates the command, then records the fact with ctx.appendEvent:

  r.writeHandler(
    "invoice:approve",
    z.object({
      id: z.uuid(),
      amountCents: z.number().int(),
      approvedBy: z.string(),
      // Optional ops metadata that the caller wants threaded into the
      // event for later filtering/auditing without bloating the payload.
      geoRegion: z.string().optional(),
      abTestBucket: z.string().optional(),
    }),
    async (event, ctx) => {
      const headers: Record<string, string> = {};
      if (event.payload.geoRegion) headers["geoRegion"] = event.payload.geoRegion;
      if (event.payload.abTestBucket) headers["abTestBucket"] = event.payload.abTestBucket;
      await ctx.appendEvent({
        aggregateId: event.payload.id,
        aggregateType: "showcase-invoice",
        type: approved.name,
        payload: { amountCents: event.payload.amountCents, approvedBy: event.payload.approvedBy },
        ...(Object.keys(headers).length > 0 ? { headers } : {}),
      });
      return { isSuccess: true as const, data: { id: event.payload.id } };
    },
    { access: { roles: ["Admin"] } },
  );

Note what the handler does not do: it doesn’t update any table. Recording the fact is the write; everything downstream reacts to it. The optional headers carry ops metadata (region, A/B bucket) without polluting the domain payload.

Derive a read model

A projection listens to events and maintains a table. This one keeps a per-invoice detail row and runs inline — in the same transaction as the event append, so the read model is never behind:

  // Single-stream projection: one row per invoice, reacts to the auto
  // CRUD event + both domain events. Runs INLINE in the write TX.
  r.projection({
    name: "invoice-detail",
    source: "showcase-invoice",
    table: invoiceDetailTable,
    apply: {
      "showcase-invoice.created": async (event, tx) => {
        const p = event.payload as { customer: string; status: string };
        await insertOne(tx, invoiceDetailTable, {
          invoiceId: event.aggregateId,
          tenantId: event.tenantId,
          customer: p.customer,
          status: p.status,
          amountCents: 0,
        });
      },
      [approved.name]: async (event, tx) => {
        const p = typedPayload(event, approved);
        await updateMany(
          tx,
          invoiceDetailTable,
          { status: "approved", amountCents: p.amountCents },
          { invoiceId: event.aggregateId },
        );
      },
      [paid.name]: async (event, tx) => {
        await updateMany(
          tx,
          invoiceDetailTable,
          { status: "paid" },
          { invoiceId: event.aggregateId },
        );
      },
    },
  });

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

Each apply entry is plain SQL against the projection table. If you later want a different table shape, change the apply functions and rebuild — the events replay, the new table appears. No data migration.

Read it back

Two ways to query:

• ctx.queryProjection / defineProjectionQueryHandler — read the projection table like any other table.
• ctx.loadAggregate(id) — load the raw events of one aggregate and reduce them in memory, when you need the full story rather than the current row.

What this buys you

The moment “approved” is an event instead of a field update, you get: the amount and approver of every approval ever (not just the last), projections you can reshape after the fact, and other features reacting to the fact (notifications, counters) without the write handler knowing about them.

When to climb to Level 3

Level 2 is where most features live. Climb when time starts to hurt: your event payloads need to change shape without breaking two years of history, or aggregates accumulate thousands of events and loading them gets slow. That’s Level 3.