Tutorial

Build one small Streamsy application, from a typed stream to a served projection.

This page builds one small shop application. Each step links to the page with the details.

1. Declare streams

import { Schema } from "effect";
import { Streamsy } from "@streamsy/core";

export const OrderEvent = Schema.Struct({ orderId: Schema.String, amount: Schema.Finite });
export const OrderRow = Schema.Struct({ orderId: Schema.String, total: Schema.Finite });

export class Orders extends Streamsy.json<Orders>()("Shop/Orders", "orders", {
  params: {},
  schema: OrderEvent,
}) {}
export class OrderTotals extends Streamsy.json<OrderTotals>()("Shop/OrderTotals", "order-totals", {
  params: {},
  schema: OrderRow,
}) {}
export class OrderEvents extends Streamsy.json<OrderEvents>()(
  "Shop/OrderEvents",
  "orders/:orderId/events",
  { params: { orderId: Schema.String }, schema: OrderEvent },
) {}

export const shop = Streamsy.group(Orders, OrderTotals, OrderEvents);

Orders has one stream. OrderEvents has one stream for each orderId. The group lists the streams that one host keeps and serves.

See Declare streams.

2. Create and append

import { Effect } from "effect";
import { Orders } from "./streams.ts";

export const write = Effect.gen(function* () {
  const orders = yield* Orders;
  yield* orders.create({});
  const appended = yield* orders.append({}, [{ orderId: "o-17", amount: 40 }]);
  return appended.offset;
});

The first argument of each method is the stream parameters.

See Create and fork and Write.

3. Follow

import { Effect, Stream } from "effect";
import { Orders } from "./streams.ts";

const follow = Effect.flatMap(Orders, (orders) =>
  orders.follow({}).pipe(
    Stream.flatMap((batch) => Stream.fromIterable(batch.items)),
    Stream.runForEach((event) => Effect.log(`${event.orderId}: ${event.amount}`)),
  ),
);

See Read.

4. Project one stream into another

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

const projection = 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 catchUp = Projection.run(projection.at({}));

The producer position makes output retries effectively once. The checkpoint records the completed input range. Projection.run stops when the input is caught up.

See How projections work.

5. Provide a Layer and run

import { Effect } from "effect";
import { MemoryStorage } from "@streamsy/core";
import { shop } from "./streams.ts";
import { write } from "./write.ts";

await Effect.runPromise(write.pipe(Effect.provide(MemoryStorage.layer(shop))));

A projection also needs CheckpointStore. Merge its Layer over the same storage:

import { Effect, Layer } from "effect";
import { MemoryStorage } from "@streamsy/core";
import { CheckpointStore } from "@streamsy/projection";
import { catchUp } from "./projection.ts";
import { write } from "./write.ts";
import { shop } from "./streams.ts";

const host = CheckpointStore.layer.pipe(Layer.provideMerge(MemoryStorage.layer(shop)));

await Effect.runPromise(write.pipe(Effect.andThen(catchUp), Effect.provide(host)));

Use SQLite when the data must survive a restart:

import { BunStorage } from "@streamsy/storage/bun";
import { shop } from "./streams.ts";

export const storage = BunStorage.layer(shop, { client: { filename: "./streams.sqlite" } });

CheckpointStore.layer works over this Layer too, and keeps the checkpoints in the same file. See Layers and storage.

6. Serve on Bun

import { Layer } from "effect";
import { BunRuntime } from "@effect/platform-bun";
import { Http } from "@streamsy/core";
import { Serve } from "@streamsy/serve";
import { listener } from "@streamsy/serve/bun";
import { storage } from "./storage.ts";
import { shop } from "./streams.ts";

const server = Serve.layer(Http.routes(shop, { prefix: "/streams", access: "read" })).pipe(
  Layer.provide(storage),
  Layer.provide(listener({ port: 3000 })),
);

BunRuntime.runMain(Layer.launch(server), { teardown: Serve.teardown });

access: "read" allows GET and HEAD. A write answers 405. On Ctrl+C, Serve.layer answers live reads before the server stops, and Serve.teardown exits 0.

See Protocol routes and Bun.

7. Deploy to Cloudflare

The same routes run inside a Durable Object. An Alchemy stack declares the object and the Worker.

alchemy dev
alchemy deploy

See Cloudflare.

8. Read from another process

import { Layer } from "effect";
import { FetchHttpClient } from "effect/http";
import { RemoteStreams } from "@streamsy/core/remote";
import { shop } from "./streams.ts";

export const remote = RemoteStreams.layer(shop, {
  baseUrl: "https://streamsy-app.example.workers.dev/streams",
  capabilities: { expectedOffset: true, producer: true },
}).pipe(Layer.provide(FetchHttpClient.layer));

remote gives the same stream services, so follow runs over it with no change.

See Remote client.

Overview

Streamsy is an Effect implementation of the Durable Streams protocol, with typed streams, projections, and hosts for Bun and Cloudflare.

Declare streams

Declare a stream class with an id template and a codec, and list stream classes in a group.

On this page

1. Declare streams2. Create and append3. Follow4. Project one stream into another5. Provide a Layer and run6. Serve on Bun7. Deploy to Cloudflare8. Read from another process