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 a query serves what it wrote.

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.

When a view isn’t the right store

A view answers one question about one key, and the model serves it. That covers most screens. Some questions don’t fit that shape. An accountant wants every money movement across every account this quarter, filtered by amount and sorted by date. A search box wants to find a customer by any part of their name. A data team wants the whole history in their warehouse. Each needs a store built for its kind of query, a SQL table, a search index, a warehouse, and each is a store you run.

A projection is how events reach a store like that. It’s still a read model in the CQRS sense: built from events, and never the source of truth. Add one to a model that already has history, and it receives that history first, so its store starts complete (see Consumers added later receive the history).

EventViewProjection

The same events feed a view the model serves and a table you query yourself.

Reach for a view first. Reach for a projection when the query needs a store a view doesn’t give you, or when the data has to leave the model altogether.

Declare what it consumes

A projection names its sources, the key that partitions it, and the events it consumes. In the example bank, a 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.

Describe it

A projection’s last argument is its annotations. No client calls a projection, so its description is for the people and the tools that read the model, as Ledger.annotations.

export class Ledger extends Domain.Projection(
  "Ledger",
  {
    from: [Account],
    key: Account,
    on: [
      MoneyDeposited,
      MoneyWithdrawn,
      TransferDebited,
      TransferCredited,
      TransferRefunded,
      SettlementReceived,
    ],
  },
  { description: "Every money movement on an account, oldest first." },
) {}

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. The query that serves the ledger reads through the same port.

export class LedgerStore extends Context.Service<
  LedgerStore,
  {
    readonly record: (rows: ReadonlyArray<LedgerRow>) => Effect.Effect<void, never, RuntimeContext>;
    /** One page of an account's movements, oldest first. */
    readonly forAccount: (
      accountId: string,
      page: Domain.Query.PageRequest,
    ) => Effect.Effect<Domain.Query.Page<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.

The key can also be a string Schema, such as an org id, when there is no aggregate to partition by. keyOf then returns the value, and the handler gets it as key.

export class OrgLedger extends Domain.Projection("OrgLedger", {
  from: [Account],
  key: OrgId,
  keyOf: { Account: ({ event }) => event.orgId },
  on: [MoneyDeposited],
}) {}

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.

version is tuning too. Bump it and the projection runs again over the whole history on the next deploy, for example after its table was lost, or once the handler writes a new shape. Empty or migrate the table first if a replayed write wouldn’t overwrite the old row.

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

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.

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, { after: undefined, take: 50 });
});

caughtUp waits until the projection has handled every event of the command that it consumes directly from that aggregate. A client’s receipt carries only stream and version; given that, caughtUp reads the command’s events back from the aggregate. That’s how a query waits for its caller’s write: atLeast: receipt.

Serve it with a query

A projection’s store is yours, and a query serves it to clients: a typed method on the Api, with middleware, rejections, a handler that checks the caller, cursor paging and read-your-own-write. The example bank’s ledger is the AccountLedger query, as private as the account it’s about.

export class AccountLedger extends Domain.Query("AccountLedger", {
  input: { account: AccountId },
  success: LedgerRow,
  paged: true,
  waitsFor: [Ledger],
  middleware: [Session],
  rejects: [NotOwner],
}) {}

To have the model keep a projection’s output per key for you, with live updates, write a stream instead.

Test it

Projections run for real in a story. Read back what one wrote through the query that serves it.

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 })),
query(AccountLedger, { account: a1.id }, [
  { kind: "MoneyDeposited", amount: 10_000, balanceAfter: 10_000 },
  { kind: "MoneyWithdrawn", amount: 2_500, balanceAfter: 7_500 },
]),

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