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, guarded like 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.

A read model you publish entry by entry

A view’s state is shaped by its evolve, and its fold belongs to the model. A stream is for when you want to decide, in code, exactly what each client-visible entry contains and when one is written. The projection that writes it can read any source, combine a view’s state with an event, skip changes no client cares about, and publish a compact entry shaped for the screen.

The bank’s account card is an example. The card on a customer’s home screen shows the owner, the balance and whether the account is frozen. It changes when any of those change, and not when anything else does. A stream of CardChanged entries is that card, with its whole history:

EventQuery

One log per account. Its last entry is the card a client shows.

Use a view when a fold describes the read model well. Use a stream when a projection should choose each entry, and clients should follow those entries live.

Declare a stream

A stream names its key and the events its entries may be. The example 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.

A stream can also be keyed by a string Schema, like an org id, with one log per value. Its writer is then keyed by the same Schema, and reads take the value: bank.OrgFeed.peek(orgId). See Key a view by a value.

Describe it

A stream’s last argument is its annotations. The read tools, CLI subcommand and OpenAPI operations an Api builds for the stream show its description after the sentence that says what the call does.

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

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.

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(({ state: 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

The declaration lists the middleware and rejects of its reads. Its Layer’s validate judges every read against the key’s latest entry, and runs again on each new entry while a client watches.

export const AccountCardLive = AccountCard.toLayer(
  Effect.succeed(
    Effect.fn(function* (_request, latest) {
      yield* mustOwn(latest?.event.customerId ?? null);
    }),
  ),
);

A stream that declares either needs its Layer, provided with the model’s handlers.

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, Query serves a store a projection writes, across keys.