Define and run

DeclareCopy Anchor Link

import { Commit, Projection } from "@streamsy/projection";
import { Orders, OrderTotals } from "./streams.ts";

export const orderTotals = Projection.emit({
  id: "order-totals",
  params: {},
  inputs: { input: Orders },
  outputs: { totals: OrderTotals },
  commit: Commit.producer,
  emit: ({ item }) => ({ totals: [{ orderId: item.orderId, total: item.amount }] }),
});
export const orderTotalsProjection = orderTotals.at({});

A declaration is inert. declaration.at(params) binds it. A runner executes the bound projection.

With one input, use inputs: { input: Orders }. An entry handler reads entry.item; a batch handler reads batch.input.items. With more inputs, give each one a name, for example inputs: { orders: Orders, refunds: Refunds }.

HandlerCopy Anchor Link

Each constructor names its handler after itself. process(batch, unit, prepared) returns an Effect; the other handlers can return a value or an Effect. Handler services must be supplied by the host or the Commit.

A runner needs the handler services, each input stream service, and the Commit's required services.

CommitsCopy Anchor Link

Only fold has a default Commit. A wrong constructor/Commit pair fails at the commit: field.

Commit.transaction provides the host's SqlClient. SQL on that client, declared outputs, and the checkpoint commit or roll back together. Declared outputs must share the transaction owner. Hand-written stream appends keep the owner check. External calls are at least once.

Memory hosts provide no Transaction. Use Commit.checkpoint for folds, Commit.producer for outputs, and Commit.atLeastOnce for processing on memory.

Commit.atLeastOnce saves the input range before running the handler. Retries keep the same unit.key. Use an outbox for durable delivery of retried external effects.

Commit.withCompensation(f) wraps a Commit, directly or through .pipe. It calls f(result, cause, unit) if the handler succeeded but the commit failed. Compensation does not make external effects atomic.

PrepareCopy Anchor Link

Projection.process with Commit.transaction accepts prepare(batch, unit). It runs outside the transaction, at least once. Its result is the third argument of process(batch, unit, prepared).

A transaction retry reads the inputs again. With unchanged ranges, unit.key is unchanged. If new input extends the range, the key changes. Do not use it as a stable external idempotency key across all retries.

A Commit plus its host Layer plays a similar role to Effect Workflow plus WorkflowEngine: the declaration chooses behavior and the host supplies persistence. Commit.atLeastOnce has the retry behavior of an Activity; it does not provide workflow orchestration.

Output projectionsCopy Anchor Link

Projection.emit declares routes as outputs and returns items for each entry. A JSON route takes its item type. A state route takes DurableState.Entry for its collections: a row change or a State control such as StreamAddress.reset().

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

export class Amounts extends Streamsy.json<Amounts>()("Shop/Amounts", "amounts", {
  params: {},
  schema: Schema.Finite,
}) {}

export const amounts = Projection.emit({
  id: "amounts",
  params: {},
  inputs: { input: Orders },
  outputs: { output: Amounts },
  commit: Commit.producer,
  emit: ({ item }) => ({ output: [item.amount] }),
});

The runner pins the unit input ranges in the checkpoint, then appends each output with its own producer position. A retry sends the same positions, so the server drops duplicates. The handler must be deterministic on its inputs.

Several outputs, with row changes and state:

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

class Numbers extends Streamsy.json<Numbers>()("Tracker/Numbers", "numbers", {
  params: {},
  schema: Schema.Finite,
}) {}
const Card = Schema.Struct({ issueId: Schema.String, total: Schema.Finite });
class Board extends Streamsy.state<Board>()("Tracker/Board", "board", {
  params: {},
  collections: { card: { schema: Card, key: "issueId" } },
}) {}
class Transitions extends Streamsy.json<Transitions>()("Tracker/Transitions", "transitions", {
  params: {},
  schema: Schema.Finite,
}) {}

const tracker = Projection.mapAccumBatch({
  id: "tracker",
  params: {},
  inputs: { numbers: Numbers },
  outputs: { board: Board, transitions: Transitions },
  commit: Commit.producer,
  state: { schema: Schema.Finite, initial: 0 },
  mapAccumBatch: (state, batch) => {
    const next = state + batch.numbers.items.reduce((sum, n) => sum + n, 0);
    return [
      next,
      {
        board: [
          DurableState.upsert("card", { issueId: "total", total: next }),
          DurableState.remove("card", "old"),
        ],
        transitions: [next],
      },
    ] as const;
  },
});

Output rules:

• Routes and fixed addresses carry the output codecs. Their parameters must be a subset of the projection parameters.
• DurableState.upsert("card", row) and DurableState.remove("card", key) make row changes. Each change names its collection.
• StreamAddress.reset(), StreamAddress.snapshotStart() and StreamAddress.snapshotEnd() make State controls. The output writes them unchanged, in order with the row changes.
• A missing output key emits nothing. An empty output uses no producer sequence.
• Two outputs cannot name the same stream or an input stream.
• Commit.producer can write across owners, but foreign output streams must already exist. It can create absent outputs on its checkpoint owner.
• Producer outputs do not commit atomically with each other. A gone or closed output fails with pin / invalid-output.

One Commit controls one projection. To combine SQL writes and an independently committed output or external sink, chain projections through a stream. A single producer projection can declare several output routes.

RunningCopy Anchor Link

import { Effect, Layer } from "effect";
import { MemoryStorage } from "@streamsy/core";
import { CheckpointStore, Projection } from "@streamsy/projection";
import { orderTotalsProjection } from "./order-totals.ts";
import { shop } from "./streams.ts";

const host = CheckpointStore.layer.pipe(Layer.provideMerge(MemoryStorage.layer(shop)));
const program = Projection.run(orderTotalsProjection, { limit: 10 });

await Effect.runPromise(program.pipe(Effect.provide(host)));

run and each follow element give a Progress. It has the state and offsets that the last pass of the cycle left, and the status, units, and items of the cycle. state is undefined when the projection has no state:. pass gives status, units, items, and offsets.

limit has no default. A bad limit or repairIntervalMs fails with load / invalid-options. For follow, the stream fails at its first pull.

import { Effect, Stream } from "effect";
import { Projection } from "@streamsy/projection";
import { orderTotalsProjection } from "./order-totals.ts";

const serve = Projection.follow(orderTotalsProjection, { repairIntervalMs: 30000 }).pipe(
  Stream.runForEach(({ state, offsets }) => Effect.log("committed", state, offsets)),
);

The stream emits after the first cycle, also when it commits nothing. After that it emits after each cycle that commits a unit or reaches limit, and after the last cycle when every input is closed. Then it ends. A wake that finds no input emits nothing.

The stream is pull based. The next cycle runs only when the consumer pulls, so a slow consumer gets one element for all the input that arrived while it worked. The consumer's pace sets the follower's pace. Add Stream.buffer({ capacity: 1, strategy: "sliding" }) to decouple them. A consumer failure stops the follower, so catch render errors in the consumer.

follow retries storage-failure. It uses options.retry, then the ProjectionRetry context reference, then its default jittered exponential schedule from 200 ms, capped at 30 s. Both overrides take a Schedule. Other faults fail the stream. Interrupt the consumer to stop the follower. Each wait is bounded by repairIntervalMs, which defaults to 30 000 ms. Create every input before starting the follower.

Terminal inputs and wake streamsCopy Anchor Link

A closed input with no new items still makes one unit with slice.closed. The checkpoint records that close, so another run does not repeat the empty unit. Batch handlers can react to it; per-item handlers have no entries. A confirmed gone input with a checkpoint makes one final unit with slice.gone: true, then the kernel forgets the binding by default. The last run result or follow element holds the state and offsets that the final unit committed. Projection.read gives the initial state after that, because the record is gone. A missing input without a checkpoint still fails with history-unavailable. Missing history on a live input and storage failures remain faults.

A pending unit replays its pinned ranges. If an input is removed between the pin and the settle, the replay fails with range-unreproducible and the pin stays. The final gone unit does not run for that binding. Retain history until pending work settles, or forget the binding after you clean up its outputs.

Projection.follow ends when all inputs close. A later removal needs an attached projection or another explicit run.

Streamsy.wake is the normal State service with fixed Streamsy.wakeCollections. Projection.wake({ id, params, input, output }) emits one keyed row { stream, offset, closed? } per source and deletes it when the source is gone. The producer creates a missing wake stream on its own owner. A wake stream on another owner must exist before the first append, so its owner calls the State service's create at start.

On Bun or Node, build Projection.attach(filesWake) beside the API and over the same storage and checkpoint layers, before writes begin. Each input id must supply every projection parameter. Its local Storage.committed subscription selects bindings; remote inputs do not trigger it. By default the pending set lives in memory, so a crash before processing waits for another source commit. With durable: true, each commit also writes a pending mark in its transaction, and the layer runs marked bindings when it starts. Retry takes a Schedule; the default is capped, jittered exponential backoff. Use one writer per binding. Cloudflare uses scoped ProjectionWakeup.attach.

TriggersCopy Anchor Link

A process-lifetime loop:

Effect.gen(function* () {
  const fiber = yield* Projection.follow(orderTotalsProjection).pipe(
    Stream.runDrain,
    Effect.forkScoped,
  );
});

Bounded work before a request answers:

Effect.gen(function* () {
  yield* Projection.run(orderTotalsProjection, { limit: 5 });
  return yield* readOrderTotals();
});

Follow input changes. Changes during a run join into one more run:

Effect.gen(function* () {
  const projectionFiber = yield* Projection.follow(orderTotalsProjection, { limit: 5 }).pipe(
    Stream.runDrain,
    Effect.forkScoped,
  );
  const workspaceFiber = yield* Projection.follow(ordersByWorkspace.at({ workspaceId: "ws-42" }), {
    limit: 5,
  }).pipe(Stream.runDrain, Effect.forkScoped);
});

A Durable Object alarm:

const alarm = Effect.gen(function* () {
  return yield* Projection.run(orderTotalsProjection, { limit: 10 });
});

A CLI:

const hand = Effect.gen(function* () {
  return yield* Projection.run(orderTotalsProjection);
});

await Effect.runPromise(hand.pipe(Effect.provide(host)));

Use one writer for each projection key: one running follow stream, or one run at a time. Each run of a follow stream value is a new writer. Stream.share gives one follower to several consumers, and stops it after the last consumer exits. To keep a projection up with no consumer, drain it and publish each element to a SubscriptionRef. A schedule must not overlap itself. A second writer is a deployment fault detected by the checkpoint token when it moves. Concurrent current and decide calls fold in memory and do not write checkpoints.

The Bugsy demo uses request and change triggers on one key.

ParametersCopy Anchor Link

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

class Facts extends Streamsy.json<Facts>()("Shop/Facts", "facts/:workspaceId", {
  params: { workspaceId: Schema.String },
  schema: orderSchema,
}) {}
class Totals extends Streamsy.json<Totals>()("Shop/Totals", "totals/:workspaceId", {
  params: { workspaceId: Schema.String },
  schema: totalSchema,
}) {}

const ordersByWorkspace = Projection.emit({
  id: "order-totals",
  params: { workspaceId: Schema.String },
  inputs: { facts: Facts },
  outputs: { output: Totals },
  commit: Commit.producer,
  emit: ({ item }) => ({ output: toTotals([item]) }),
});

const bound = ordersByWorkspace.at({ workspaceId: "ws-42" });
const params = ordersByWorkspace.parse("facts/ws-42");
const follow = Effect.gen(function* () {
  const fibers = yield* Effect.forEach([{ workspaceId: "ws-42" }], (params) =>
    Stream.runDrain(Projection.follow(ordersByWorkspace.at(params))).pipe(Effect.forkScoped),
  );
});

Parameter rules:

• Each input and output stream declares a subset of the projection parameter names, with the same types.
• A stream with fewer parameters is shared by every binding.
• parse tries the input streams in order and gives the decoded subset from the first match.
• The package keeps no list of bound parameters. Bind and trigger each one.

To keep a dynamic set, keep a registry stream of parameter records. A projection over that stream can start and stop triggers.

ConstructorsCopy Anchor Link

See the Commit fit table for compatible guarantees.

foldCopy Anchor Link

One entry updates a value. The default Commit.checkpoint saves the encoded value and offsets in one compare-and-set. Keep the step free of external effects.

import { Schema } from "effect";
import { Projection } from "@streamsy/projection";
import { Orders } from "./streams.ts";

const total = Projection.fold({
  id: "total",
  params: {},
  inputs: { orders: Orders },
  state: { schema: Schema.Finite, initial: 0 },
  fold: (sum, { item }) => sum + item.amount,
});

processCopy Anchor Link

One batch runs effects. Projection.each adapts an entry handler and preserves input declaration order.

import { Effect } from "effect";
import { SqlClient } from "effect/sql";
import { Commit, Projection } from "@streamsy/projection";
import { Orders } from "./streams.ts";

const rows = Projection.process({
  id: "order-rows",
  params: {},
  inputs: { orders: Orders },
  commit: Commit.transaction,
  process: Projection.each(({ item }) =>
    Effect.gen(function* () {
      const sql = yield* SqlClient.SqlClient;
      yield* sql`INSERT INTO order_totals (order_id, total) VALUES (${item.orderId}, ${item.amount})`;
    }),
  ),
});

The application creates and migrates order_totals.

emitCopy Anchor Link

One entry returns output items. See Output projections. The step returns a value or an Effect. It does not carry state.

mapAccumCopy Anchor Link

One entry updates state and returns output items as [state, emits]. It is like Stream.mapAccum, with a checkpoint and declared output routes.

import { Schema } from "effect";
import { Commit, Projection } from "@streamsy/projection";
import { Orders, OrderTotals } from "./streams.ts";

const running = Projection.mapAccum({
  id: "running",
  params: {},
  inputs: { orders: Orders },
  outputs: { totals: OrderTotals },
  commit: Commit.producer,
  state: { schema: Schema.Finite, initial: 0 },
  mapAccum: (sum, { item }) => {
    const next = sum + item.amount;
    return [next, { totals: [{ orderId: item.orderId, total: next }] }] as const;
  },
});

mapAccumBatchCopy Anchor Link

One batch updates state and returns [state, emits]. See the tracker above. It can coalesce row updates within a unit. Batch handlers must produce the same final rows for any unit size; emitted history can differ.

fold, emit, mapAccum, and mapAccumBatch return a value or an Effect. In per-entry steps, use unit for parameters and offsets only; its ranges describe the whole unit.