How projections work

A projection reads one or more input streams and writes derived data. It works in units. Each unit reads a slice of each input, runs a handler, and commits a checkpoint. A restart continues from the last checkpoint.

One pass:

1. Load the checkpoint.
2. Read each input after its checkpoint offset, in declaration order.
3. Run the handler over the unit.
4. Commit the checkpoint.

ConstructorsCopy Anchor Link

Choose the guarantee with a Commit. fold defaults to Commit.checkpoint; other constructors require commit:.

Transactional outputsCopy Anchor Link

Projection.emit maps each entry to output items. With Commit.transaction, the handler runs inside a SQL transaction. Appends to streams on the same storage, and SQL through the same client, commit or roll back with the checkpoint.

import { Effect, Schema } from "effect";
import { Streamsy } from "@streamsy/core";
import { Commit, Projection } from "@streamsy/projection";

export class Numbers extends Streamsy.json<Numbers>()("Minimal/Numbers", "numbers", {
  params: {},
  schema: Schema.Finite,
}) {}
export class Doubled extends Streamsy.json<Doubled>()("Minimal/Doubled", "doubled", {
  params: {},
  schema: Schema.Finite,
}) {}

export const minimal = Effect.gen(function* () {
  const numbers = yield* Numbers;
  const doubled = yield* Doubled;
  yield* numbers.create({});
  yield* doubled.create({});
  yield* numbers.append({}, [1, 2, 3]);
  return yield* Projection.run(
    Projection.emit({
      id: "doubled",
      params: {},
      inputs: { input: Numbers },
      outputs: { output: Doubled },
      commit: Commit.transaction,
      emit: ({ item }) => ({ output: [item * 2] }),
    }).at({}),
    { limit: 10 },
  );
});

One run writes [2, 4, 6] to doubled. The source is packages/projection/src/examples/minimal.ts.

Output projectionsCopy Anchor Link

Commit.producer pins input ranges and output positions. The handler returns items for each route, and the runner appends them.

import { Schema } from "effect";
import { Streamsy } from "@streamsy/core";
import { Commit, Projection } from "@streamsy/projection";
import { Numbers } from "./minimal.ts";

export class Tripled extends Streamsy.json<Tripled>()("Minimal/Tripled", "tripled", {
  params: {},
  schema: Schema.Finite,
}) {}

export const tripled = Projection.emit({
  id: "tripled",
  params: {},
  inputs: { input: Numbers },
  outputs: { output: Tripled },
  commit: Commit.producer,
  emit: ({ item }) => ({ output: [item * 3] }),
});

Each output append has its own producer position, so a retry does not write a unit twice.

StateCopy Anchor Link

fold, mapAccum, and mapAccumBatch declare state: { schema, initial }. The value is saved with the checkpoint. Projection.read gives the value and the input offsets it is at. See State and checkpoints.

Fact streams and state streamsCopy Anchor Link

A fact stream records what happened. Each message is a domain event, appended once. Declare it with Streamsy.json.

A state stream carries the current value of keyed rows as Durable State change events. Declare it with Streamsy.state. A browser reads it with @durable-streams/state.

{
  "type": "order",
  "key": "o-17",
  "value": { "orderId": "o-17", "sku": "A1", "status": "shipped" },
  "headers": { "operation": "upsert" }
}

A projection can rebuild a state stream from the facts.

HostsCopy Anchor Link

Built-in Commits need CheckpointStore. Commit.transaction also needs Transaction, supplied only by SQL hosts. Provide the services over the storage Layer:

import { Layer } from "effect";
import { MemoryStorage, Streamsy } from "@streamsy/core";
import { CheckpointStore } from "@streamsy/projection";
import { Numbers } from "./minimal.ts";
import { Tripled } from "./tripled.ts";

export const host = CheckpointStore.layer.pipe(
  Layer.provideMerge(MemoryStorage.layer(Streamsy.group(Numbers, Tripled))),
);

The memory host runs tripled with Commit.producer. The minimal transaction example needs a SQL host. See Layers.

PagesCopy Anchor Link