Domain.Projection reference

Projection

Source: packages/domain/src/Projection.ts

Consumes events into a store you own: a database, a search index, a warehouse, or a Domain.Stream. Inputs are routed to keys and handled in order per key, at least once, and batched if its Layer says so. A projection never sends commands: replaying it must never act on the world again.

The declaration holds what it means (sources, partitioning, events, the streams it writes). Its Layer holds how it runs: the handler, and options such as batching.

Projection: Writing to a database

import * as Domain from "@alchemy.run/domain";
import * as Effect from "effect/Effect";
import { Account } from "./Account.ts";
import { MoneyDeposited } from "./DepositMoney.ts";
import { LedgerStore } from "./LedgerStore.ts";

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

export const LedgerLive = Ledger.toLayer(
  Effect.gen(function* () {
    const store = yield* LedgerStore;
    return Effect.fn(function* (batch) {
      yield* store.upsert(
        batch.map(({ key, event, envelope }) => ({
          id: `${envelope.stream}:${envelope.seq}`,
          account: key.id,
          amount: event.amount,
        })),
      );
    });
  }),
  { batch: { size: 500, wait: "2 seconds" } },
);

Projection: Reading your own write

const receipt = yield* accounts.send(accountId, new DepositMoney({ amount: 100 }));
yield* Ledger.caughtUp(receipt);