Stream

A stream is a keyed, append-only log: one ordered list of entries per key, kept for good and readable from any point. 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 one stream answers both “what is it now?” and “tell me every change”, which is what a client syncing state needs: the current value first, then everything after it.

Declare a stream

A stream names its key and the events its entries may be. The bank keeps a card per account, one entry each time the card changes.

export class CardChanged extends Domain.Event("CardChanged", {
  customerId: CustomerId,
  owner: Schema.String,
  balance: Cents,
  frozen: Schema.Boolean,
}) {}

export class AccountCard extends Domain.Stream("AccountCard", {
  key: Account,
  entries: [CardChanged],
  middleware: [Session],
  rejects: [NotOwner],

Add it to the model’s streams.

Write it from a projection

A stream has exactly one writer: a projection that lists it in writes, keyed by the same aggregate. AccountCards reads the account summary, whose events arrive with the summary’s state after each one, and appends a card for every change.

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

append takes the input being handled. That’s how the stream knows which key it belongs to, and how it keeps writes exactly once: each append is identified by the input’s place in the projection’s log, so a redelivered or replayed input appends nothing twice.

AggregateViewProjectionStream

Each change to the summary reaches the projection, which appends one card to the account's stream.

One writer per key

Order per key only holds if each key has a single writer. The writer says so in its declaration:

writes: [AccountCard],

The model checks this when it’s built: a stream listed in two projections’ writes, or written by a projection keyed differently, is a definition error.

Read it

Exposed on an Api, a stream offers four reads for a key.

import * as Effect from "effect/Effect";
import * as Stream from "effect/Stream";
import type { AccountId } from "@bank/domain/Account/Account.id";
import type { CustomerClient } from "./flows.ts";

export const cardReads = Effect.fn(function* (bank: CustomerClient, accountId: AccountId) {
  // The card now: the latest entry, or null.
  const now = yield* bank.AccountCard.peek(accountId);

  // Every new card, starting with the current one.
  yield* bank.AccountCard.watch(accountId).pipe(
    Stream.take(2),
    Stream.runForEach((card) => Effect.log(`balance ${card?.event.balance}`)),
  );

  // History from a position, then live; and a page of it.
  yield* bank.AccountCard.events(accountId, { after: 0 }).pipe(Stream.take(1), Stream.runDrain);
  const page = yield* bank.AccountCard.list(accountId, { order: "desc", take: 10 });
  return [now, page.entries.length] as const;
});

Over Domain.protocolWebSocket, watch and events are WebSockets to the Durable Object that owns the key, which pushes each new entry as it’s appended.

peek also takes atLeast, a stream and version, and waits until the stream’s writer has handled that commit. It only covers sources the writer consumes directly. AccountCards reads accounts through the AccountSummary view, so an account command’s receipt is not one its stream can wait for.

Decide who may read it

validate judges every read against the key’s latest entry, and runs again on each new entry while a client watches.

validate: Effect.fn(function* (_request, latest) {
  yield* mustOwn(latest?.event.customerId ?? null);
}),

Use it as a value

Because peek returns the latest entry, a stream whose entries are full snapshots, like the card, is a value with a history.

const now = yield* bank.AccountCard.peek(accountId);

There’s no separate “current value” to keep in sync, and no race between two writers setting it: there’s one writer, and its entries are ordered.

Test it

Story.streamed checks the entries a projection appended since the last check.

streamed(
  AccountCard,
  a1,
  new CardChanged({ customerId: c1.id, owner: "sam", balance: 7_500, frozen: false }),
),

Next, Policy covers the consumer that acts on the world rather than recording it.