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 pass give a Progress with status, units, items, and offsets.

limit has no default. A bad limit or repairIntervalMs fails with load / invalid-options.

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

const serve = Effect.scoped(
  Effect.gen(function* () {
    const fiber = yield* Projection.follow(orderTotalsProjection, {
      repairIntervalMs: 30000,
    });
    return yield* Fiber.join(fiber);
  }),
);

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 end the fiber. Close the scope to stop it. Each wait is bounded by repairIntervalMs, which defaults to 30 000 ms. Create every input before starting the follower.

TriggersCopy Anchor Link

A process-lifetime loop:

Effect.gen(function* () {
  const fiber = yield* Projection.follow(orderTotalsProjection);
});

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 });
  const workspaceFiber = yield* Projection.follow(ordersByWorkspace.at({ workspaceId: "ws-42" }), {
    limit: 5,
  });
});

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 follow, or one run at a time. 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 } 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) =>
    Projection.follow(ordersByWorkspace.at(params)),
  );
});

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.