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, guarded like a view: middleware and rejections on the declaration, validate on its Layer. 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: Validating reads

validate judges the key’s latest entry, on its Layer. A stream that declares middleware or rejects needs it.

export const AccountCardLive = AccountCard.toLayer(
  Effect.succeed(
    Effect.fn(function* (_request, latest) {
      const actor = yield* CurrentActor;
      if (latest && latest.event.owner !== actor.customerId) return yield* new NotOwner();
    }),
  ),
);

Stream: Reading it

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

Stream: Annotating streams

The last argument is its annotations: data for what is built on the model, read back as X.annotations. Add your own keys by augmenting Domain.Annotations.

An Api’s read tools, CLI help and OpenAPI document show its description after the sentence for each call.

export class AccountCard extends Domain.Stream(
  "AccountCard",
  { key: Account, entries: [CardChanged] },
  { description: "An account's card, with an entry each time it changes." },
) {}