Build a read model across multiple aggregates

Some read models can’t be written by r.projection(def) against a single entity. A “revenue per customer” number aggregates events from invoices, refunds, and credit notes, three streams. A “components-with-active-incidents” view crosses component and incident streams. For these you use r.multiStreamProjection.

Prerequisites

• You’ve read Events and projections so the difference between single-stream and multi-stream projections is clear.
• The events you want to aggregate are already declared with r.defineEvent somewhere.

The code

Define the projection’s table, then the projection itself. The tested event-sourcing recipe contains the complete customer-revenue table and projection. Its projection registration is the source-backed example for this guide:

  // event-dispatcher (runOnce in tests, NOTIFY/LISTEN in production).
  r.multiStreamProjection({
    name: "customer-revenue",
    table: customerRevenueTable,
    apply: {
      [paid.name]: async (event, tx) => {
        // Resolve the customer: pull it from the invoice-detail projection
        // that the inline projection just populated. Cross-projection reads
        // are fine inside apply() — we're at the read model layer.
        const detail = await fetchOne<{ customer: string }>(tx, invoiceDetailTable, {
          invoiceId: event.aggregateId,
        });
        if (!detail) return;
        const p = typedPayload(event, paid);
        await incrementCounter(
          tx,
          customerRevenueTable,
          {
            customer: detail.customer,
            tenantId: event.tenantId,
            paidInvoices: 1,
            totalCents: p.amountCents,
          },
          { paidInvoices: 1, totalCents: p.amountCents },
          { conflictKeys: ["customer"] },
        );
      },
    },
  });

How it differs from single-stream

	r.projection	r.multiStreamProjection
Lifecycle	inline (write transaction)	async (event-dispatcher cursor)
Source	one entity	event-type-driven, cross-aggregate
Delivery	same-transaction with the event	at-least-once, cursor-driven
Latency	zero (same TX)	asynchronous; event-dispatcher cursor

A multi-stream projection runs after the transaction commits and catches up via a cursor. If the process crashes, the cursor records where it stopped; on restart the projection replays from there. The apply function therefore must be idempotent, onConflictDoUpdate or update … where, not blind insert.

Side-effect-only projections

You can omit the table: field entirely. A table-less multi-stream projection is the canonical place for “send a webhook on every paid invoice”, the apply function does its work, the cursor records progress, and there is no read model to query:

r.multiStreamProjection({
  name: "stripe-webhooks",
  apply: {
    "invoice:event:paid": async (event, _tx) => {
      await fetch(STRIPE_WEBHOOK_URL, {
        method: "POST",
        body: JSON.stringify(event.payload),
      });
    },
  },
});

This replaces the classic pub/sub subscriber pattern: the cursor gives you ordering, retry, and deduplication out of the box.

Common gotchas

• event.tenantId is the source of truth for scoping. Always carry it into the rows you write, otherwise you leak across tenants in the read model.
• Apply keys must match registered events. A typo like "invoice:event:payd" is a boot error, not a silent no-op.
• MSP rebuilds are operational work. A table-backed MSP can be rebuilt with the framework’s rebuildMultiStreamProjection path, which resets its consumer cursor, replays subscribed events through a shadow table, and swaps the result. A side-effect-only MSP has no safe rebuild because replay would resend webhooks or notifications; use consumer skip/restart operations for poison events instead. The single-stream kumiko project rebuild command does not rebuild MSPs.

Live example

A revenue-per-customer projection fed by paid events from the invoice stream, using onConflictDoUpdate to stay idempotent:

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

  // Multi-stream projection: one row per customer. Fires ASYNC via the
  // event-dispatcher (runOnce in tests, NOTIFY/LISTEN in production).
  r.multiStreamProjection({
    name: "customer-revenue",
    table: customerRevenueTable,
    apply: {
      [paid.name]: async (event, tx) => {
        // Resolve the customer: pull it from the invoice-detail projection
        // that the inline projection just populated. Cross-projection reads
        // are fine inside apply() — we're at the read model layer.
        const detail = await fetchOne<{ customer: string }>(tx, invoiceDetailTable, {
          invoiceId: event.aggregateId,
        });
        if (!detail) return;
        const p = typedPayload(event, paid);
        await incrementCounter(
          tx,

Full source: samples/recipes/event-sourcing.

See also

• Events and projections, single-stream vs. multi-stream contracts in detail.