Events and projections

Kumiko writes every state change as an event in an append-only log before it updates any read model. The event log is the source of truth; the entity tables and custom projections are derived from it. You never read the event log directly — you read the projections — but the log is what survives. Drop a projection table, replay the events, and the data is back.

This is the same model Marten follows on .NET: a single Postgres database, events written in the same transaction as their projections, and a clear separation between the single-stream “this is the row” projections and the multi-stream “this is a view across many things” projections.

What gets written on a write

When a write handler runs and ends successfully, the CrudExecutor writes two things in the same transaction:

An event row on the entity’s aggregate stream (<tenantId>:<entity>:<id>).
The corresponding update to the entity table (and any projections that subscribe).

Either both succeed or both roll back. There is no window in which the event log shows a change that did not happen, and no window in which the entity table shows a change that no event records.

const incidentResolved = r.defineEvent("incident.resolved", z.object({
  resolution: z.string(),
  resolvedAt: z.date(),
}));

const resolveIncident = defineWriteHandler({
  name: "resolve",
  schema: z.object({ id: z.string(), resolution: z.string() }),
  handler: async (event, ctx) => {
    await ctx.appendEvent(incidentResolved, event.payload.id, {
      resolution: event.payload.resolution,
      resolvedAt: ctx.now(),
    });
  },
});

r.writeHandler(resolveIncident);

ctx.appendEvent is the primitive: append this event to this aggregate stream, in this transaction. The default CRUD handlers append <entity>.created, .updated, .deleted, and .restored automatically. Domain events that mean something specific in your business — incident.resolved, order.shipped, payment.captured — you declare with r.defineEvent(name, schema) and emit explicitly.

What an event carries

Every event has a typed payload (validated against the Zod schema you supplied) plus a small fixed envelope: tenant ID, aggregate ID, event type, schema version, who triggered it, and when. For updates and deletes, the payload also carries previous — the state before the change. That sounds redundant; it is deliberate.

Cross-aggregate projections need it. A counter that tracks “units per property” needs to decrement the old property and increment the new one when a unit is moved. Without previous in the event, that’s an extra database read in every projection apply, and it can race. Kumiko pays the storage cost so that projections remain a pure function of the event stream.

Sensitive fields and the immutable log

Append-only is a feature for audit and replay; it is a problem for password hashes and API tokens. Once written into an event payload, those values are permanently in history.

Mark the field sensitive: true and Kumiko strips it out of every event payload — create.data, update.changes, update.previous, delete.previous, restore.previous. The entity table still stores the value; only the event log is sanitised. Two consequences follow:

A projection cannot read sensitive fields. They never appear in apply().
Replaying events does not reconstruct sensitive fields. If a password hash is lost, the user resets their password — there is no historical recovery.

That is the right trade. The audit trail captures “this user changed their password at 14:32”; it does not need to capture what they changed it to.

Two kinds of projection

A projection is a read model fed from events. Kumiko distinguishes two varieties because they have different transactional contracts.

Single-stream projections (r.projection(def)) run in the same transaction as the event append. They are exactly-once by construction: the event row, the entity row, and the projection row all live or die together. The projection’s apply function gets the event and the transaction-scoped database connection, and writes are typically INSERT … ON CONFLICT DO UPDATE so two concurrent events for the same key cannot race.

r.projection({
  name: "units-per-property",
  source: "unit",
  table: unitsPerPropertyTable,
  apply: {
    "unit.created": async (event, tx) => {
      const propertyId = event.payload.propertyId;
      await tx.insert(unitsPerPropertyTable)
        .values({ propertyId, tenantId: event.tenantId, unitCount: 1 })
        .onConflictDoUpdate({
          target: unitsPerPropertyTable.propertyId,
          set: { unitCount: sql`${unitsPerPropertyTable.unitCount} + 1` },
        });
    },
  },
});

If apply throws, the transaction rolls back. The event is not in the log; the entity update did not happen; the projection table is unchanged. The write call returns the error to the user. There is no inconsistent half-state to clean up.

Multi-stream projections (r.multiStreamProjection(def)) run after the transaction commits, asynchronously, driven by a database cursor. They listen to event types from many entities and aggregate them in a view that crosses aggregate boundaries — “revenue per customer across all invoices and refunds”, “open incidents per service across multiple component streams”. Their delivery is at-least-once: the cursor records progress, a crash and restart replays from the last checkpoint, and apply() must be idempotent.

The trade-off is explicit: single-stream projections are consistent and synchronous (zero lag), multi-stream projections cost a network roundtrip and can be a few hundred milliseconds behind the write. Pick based on whether your read needs to be consistent with the write that triggered it, or whether eventual consistency is fine for the use case.

Two-phase hooks

The same dual contract appears in lifecycle hooks. A postSave hook runs in one of two phases:

inTransaction — runs before commit, in the transaction. Throwing here rolls everything back. Use this for derived database writes that must stay atomic with the change: audit rows, denormalised counters, dependent entity updates.
afterCommit (the default) — runs after commit, outside the transaction. Throwing here is logged but does not roll back. Use this for external side effects: sending email, pushing SSE updates, enqueueing jobs, posting webhooks.

The reasoning is the dual-write problem: an email cannot be unsent, an SSE event cannot be un-broadcast, a Bull queue job already scheduled cannot be un-scheduled. If those run inside the transaction and the transaction rolls back, you have a side effect that should not have happened. Kumiko forces you to pick the phase explicitly so that the question “is this rollback-safe?” is answered at write time, not after the first incident.

Optimistic locking, automatically

Every entity carries a version column. Updates run with WHERE version = expectedVersion; if the row’s version moved on between read and write, the write fails with version_conflict. The handler does nothing special — the framework writes the predicate, and the renderer shows a default “this record changed, reload?” dialog when the error reaches the browser.

Two parallel edits with the same starting version are impossible. The first write wins, the second sees the conflict and is told to reload. There is no silent overwrite.

Replay and rebuild

Because the event log is the source of truth, a projection is by definition disposable. bun kumiko project rebuild <name> truncates the projection table, replays every event in order through apply(), and writes the result back — all in a single transaction. If the replay throws, the old projection stays intact; only the next attempt sees the change.

This matters more than it sounds:

A projection schema change (add a column, change a derivation) becomes a rebuild, not a migration.
A projection bug becomes a fix-and-rebuild, not a data archaeology project.
A new projection added six months after the events landed gets the historical data for free.

The events are the contract; the projections are caches. As long as the events remain stable (and r.eventMigration() covers the cases where they cannot), you can rebuild any read model from any point in history.

What this gives you

The append-only-log choice is not “for audit purposes” — audit is a side effect. The real outcomes are:

Atomic dual writes — event and projection commit together, no out-of-sync windows.
Disposable read models — projection schema is not a migration target; it is a function from events.
Time travel — replay events up to a timestamp and see the world as it was. Useful for debugging, mandatory for some compliance regimes.
No silent overwrites — optimistic locking is the default, not an opt-in.

The cost is bookkeeping: events have envelopes, payloads include previous, sensitive fields need flagging, and writes are slightly more expensive than a plain UPDATE. For an app you intend to live with for years, that is the right side of the trade.

Live example

A single-stream projection: one row per invoice, fed by the auto CRUD event plus two declared domain events, all in the same write transaction:

      approverId: p.approverId,
      approverDisplayName: row?.displayName ?? `unknown:${p.approverId}`,
    };
  });

  // 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 tx.insert(invoiceDetailTable).values({
          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 tx
          .update(invoiceDetailTable)
          .set({ status: "approved", amountCents: p.amountCents })
          .where(sql`${invoiceDetailTable["invoiceId"]} = ${event.aggregateId}`);
      },
      [paid.name]: async (event, tx) => {
        await tx
          .update(invoiceDetailTable)

Full source: samples/recipes/event-sourcing — also covers multi-stream projections, custom event types, and rebuild.