Projection

A projection consumes events into a store you own: a database table, a search index, a warehouse, or a Domain.Stream. Where a view is a read model the platform stores and serves for you, a projection writes wherever you choose and leaves serving it to you.

It shares the view’s routing and ordering, and differs from a policy in one way that shapes everything else: a projection never sends commands. Its only effect is on its own store, so it can always be replayed.

Declare what it consumes

A projection names its sources, the key that partitions it, and the events it consumes. The bank’s ledger records every money movement on an account.

export class Ledger extends Domain.Projection("Ledger", {
  from: [Account],
  key: Account,
  on: [
    MoneyDeposited,
    MoneyWithdrawn,
    TransferDebited,
    TransferCredited,
    TransferRefunded,
    SettlementReceived,
  ],
}) {}

Add it to the model’s projections, like a policy.

AggregateProjection

The account commits first. Shortly after, the projection writes the movement to a table outside the model.

Implement it as a Layer

The handler lives in a Layer, built once with everything it needs. It always receives a batch: an array of inputs for one key, in order.

export const LedgerLive = Ledger.toLayer(
  Effect.gen(function* () {
    const store = yield* LedgerStore;
    return Effect.fn(function* (batch) {
      yield* store.record(
        batch.map(({ key, event, envelope }) => ({
          id: `${envelope.stream}:${envelope.seq}`,
          accountId: key.id,
          kind: event._tag,
          amount: event.amount,
          balanceAfter: event.balanceAfter,
          at: DateTime.toEpochMillis(envelope.at),
        })),
      );
    });
  }),
  { batch: { size: 100, wait: "1 second" } },
);

Each input carries the key it was routed to, the event, its envelope, the source it came from, and, for a view source, the view’s state after the event.

Depend on a store, not a database

The projection asks for LedgerStore, a service you define in the domain package, rather than a D1 binding. The server package provides it with D1; a story provides it with memory.

export class LedgerStore extends Context.Service<
  LedgerStore,
  {
    readonly record: (rows: ReadonlyArray<LedgerRow>) => Effect.Effect<void, never, RuntimeContext>;
    readonly forAccount: (
      accountId: string,
    ) => Effect.Effect<ReadonlyArray<LedgerRow>, never, RuntimeContext>;
  }
>()("Bank/LedgerStore") {}

Its methods require RuntimeContext, because a database binding only exists inside a running Worker. A projection’s handler may require it; the platform runs handlers inside the runtime.

Make every write idempotent

Inputs arrive at least once. If a handler fails halfway, or its host restarts, the same inputs run again, so a write must be safe to repeat. The ledger keys each row by the event it came from, stream:seq, and ignores a row it already has.

"INSERT INTO ledger (id, account_id, kind, amount, balance_after, at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT (id) DO NOTHING",

The platform helps on its side too: every key keeps an input log, and a delivery it has already accepted is dropped on arrival.

Partition it by key

key decides which inputs are handled in order. Inputs for one key are handled one batch at a time, in the order they were accepted; different keys run in parallel.

AggregateProjection

One ordered partition per key. Keys don't wait for each other.

A source keyed by the same aggregate routes itself. For any other source, keyOf names the key, by event tag or source name, and can send one event to several keys with named routes, exactly as a view’s keyOf does. Each input then carries its route.

Within a key, events from one source stay in that source’s order. Events from different sources have no order between them: the model is distributed on purpose, and there is no total order to give.

Batch it on its Layer

Batching changes how a projection runs, not what it means, so it lives on the Layer. batch: { size, wait } hands the handler up to size inputs of one key at once, waiting up to wait for a batch to fill. Without it, each input is handled on its own, as soon as it arrives.

{ batch: { size: 100, wait: "1 second" } },

The handler’s code is the same either way: it always receives an array. Changing the batch size is pure tuning.

It never sends commands

A projection’s effects stay in its own store. That’s what makes it safe to replay: run it again over the same inputs, and the only thing that happens is the same rows are written again.

A policy is the opposite. It acts on the world, through commands and external systems, so it handles each new event once and never replays. When you need both, split the work: the projection writes the store, and a policy on the same events sends the commands.

CommandEventProjectionPolicy

The same event, consumed twice: once into a store that can be rebuilt, once into an action that can't be undone.

Read your own write

Your store is eventually consistent with the model. A caller who has just sent a command and wants to read the result from your store waits for the projection first.

import * as Effect from "effect/Effect";
import { Account } from "../src/Account/Account.aggregate.ts";
import type { AccountId } from "../src/Account/Account.id.ts";
import { WithdrawMoney } from "../src/Account/WithdrawMoney.command.ts";
import { Ledger } from "../src/Ledger/Ledger.projection.ts";
import { LedgerStore } from "../src/Ledger/LedgerStore.ts";

export const ledgerAfterWithdrawal = Effect.fn(function* (accountId: AccountId) {
  const accounts = yield* Account;
  const receipt = yield* accounts.send(accountId, new WithdrawMoney({ amount: 2_000 }));
  yield* Ledger.caughtUp(receipt);
  return yield* (yield* LedgerStore).forAccount(accountId);
});

caughtUp waits until the projection has handled every event in the receipt that it consumes directly from that aggregate.

Serve it yourself

A projection’s store is yours, so its API is too. The bank serves the ledger from a plain route on its Worker, authorizing the caller itself.

const summary = yield* summaries.query(Account.ref(accountId)).pipe(Effect.orDie);
const owner = Option.getOrNull(summary)?.customerId;
if (actor.value._tag === "Customer" && owner !== actor.value.customerId) {
  return HttpServerResponse.empty({ status: 404 });
}
return yield* HttpServerResponse.json(yield* ledger.forAccount(accountId));

Clients that want to know when to refetch can follow the same events the projection consumed, through the Api.

To have the model serve a projection’s output for you, with live updates and validate, write a stream instead.

Test it

Projections run for real in a story. Give the story a store you can read back.

const ledger = memoryLedger();
yield* ledgerStory(
  given(a1, opened(), new MoneyDeposited({ amount: 10_000, balanceAfter: 10_000 })),
  as(sam),
  when(a1, new WithdrawMoney({ amount: 2_500 })),
  then(a1, new MoneyWithdrawn({ amount: 2_500, balanceAfter: 7_500 })),
).pipe(Effect.provide(ledger.layer));

Next, Stream is the store the model keeps and serves for a projection.