Domain.Stream reference

Stream

Source: packages/domain/src/Stream.ts

A keyed, append-only log: one ordered list of entries per key, kept and replayable. A projection writes it; clients read it through an Api with the same middleware and validate as a view. Its latest entry is its current value, so a stream of snapshots serves both “what is it now?” (peek) and “tell me every change” (watch, events).

Stream: Declaring a stream

import * as Domain from "@alchemy.run/domain";
import * as Schema from "effect/Schema";
import { Account } from "./Account.ts";

export class CardChanged extends Domain.Event("CardChanged", {
  owner: Schema.String,
  balance: Schema.Int,
}) {}

export class AccountCard extends Domain.Stream("AccountCard", {
  key: Account,
  entries: [CardChanged],
}) {}

Stream: Writing it from a projection

A projection lists the streams it writes, then appends for the input it is handling.

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

export const AccountCardsLive = AccountCards.toLayer(
  Effect.gen(function* () {
    const cards = yield* AccountCard;
    return Effect.fn(function* (batch) {
      for (const input of batch) {
        yield* cards.append(input, new CardChanged({ owner: input.state.owner, balance: input.state.balance }));
      }
    });
  }),
);

Stream: Reading it

const latest = yield* bank.AccountCard.peek(accountId);
yield* bank.AccountCard.watch(accountId).pipe(Stream.runForEach((card) => Effect.log(card)));