Cloudflare

A Worker router sends each protocol path to one Durable Object. The object owns its SQLite storage, its alarm, and its HttpRouter app.

import type { DurableObjectNamespace } from "@cloudflare/workers-types";
import { Http, Streamsy } from "@streamsy/core";
import { Placement, StreamsyObject, router } from "@streamsy/serve/cloudflare";
import { DurableObjectStorage } from "@streamsy/storage/durable-object";
import { Events } from "./streams.ts";

const game = Streamsy.group(Events);

interface Env {
  readonly STREAMS: DurableObjectNamespace;
}
export class StreamsObject extends StreamsyObject.make<Env, Streamsy.Services<typeof game>>({
  app: Http.routes(game, { prefix: "/streams", access: "append" }),
  layer: (state) => DurableObjectStorage.layer(game, { client: { storage: state.storage } }),
}) {}
export default router<Env>({
  prefix: "/streams",
  namespace: (env) => env.STREAMS,
  placement: Placement.byKey(() => "streams"),
});

The second type parameter of StreamsyObject.make is the services that the app needs. Streamsy.Services<typeof game> is the stream services of the group.

StorageCopy Anchor Link

import { DurableObjectStorage } from "@streamsy/storage/durable-object";
import { shop } from "./streams.ts";

const storage = DurableObjectStorage.layer(shop, { client: { storage: ctx.storage } });

Pass the full ctx.storage handle. The driver needs its transaction API. The long-poll timeout is 25 s.

PlacementCopy Anchor Link

Rules:

• A fork source and its child must be on the same object.
• Keep placement stable while data lives on an object.
• A transaction cannot span objects. Use producer outputs for delivery across owners.
• Paths whose first segment starts with __ are reserved for internal mounts such as object links. The router answers 404 for them and calls no object.
• The router understands protocol paths only. Route Serve paths to their object in your own Worker code.
• Authenticate before the router forwards a request.

Object lifecycleCopy Anchor Link

Request logging is off. No URL or request header can start an alarm turn.

WakeupsCopy Anchor Link

Pass wakeups to StreamsyObject.make or objectHandlers. Each wakeup reads its next deadline from committed state and runs bounded, idempotent work when due. Return Option.none() when there is no work. The host always runs stream expiry first. Names must be unique within the object, including streamsy/expiry; duplicates fail at acquisition.

import { Wakeup } from "@streamsy/serve/cloudflare";

const catchUp = Wakeup.make({
  name: "app/catch-up",
  next: nextDeadline,
  run: runDueWork,
});
// Add wakeups: [catchUp] to the object's configuration.

The earliest deadline owns the object's one alarm slot. Due work arms an alarm at now + 1 after a write; it runs in an alarm turn. A GET that queues work calls Wakeups.use((wakeups) => wakeups.touch) before the handler returns, and the host reconciles when the request ends. A streamed body that queues work after the handler returns calls Wakeups.use((wakeups) => wakeups.reconcile). Work outside a request does the same.

Failures are logged and do not stop the other wakeups. This applies to next and to run. A failing next arms its own retry, and the other deadlines still arm. A failed run backs off from one second to one hour. A successful run backs off in the same way when work that was due before it started is still due. Override the policy with retry: Schedule.Schedule<unknown>. A finite schedule stops retries on exhaustion. A run that stops at its bound returns { more: true }. The alarm then fires again at once and the failure count stays at zero. Progress to a future deadline, or no work, resets the failure count. Wakeups.status shows each retryAt floor. Backoff lives in memory and resets on eviction. The alarm turn sets a 30-second recovery guard before dispatch. Delivery is at least once. A wakeup's permit is not reentrant, so a run must not call its own wakeup's run.

For an action outbox, import OutboxWakeup from @streamsy/serve/action/runtime and register OutboxWakeup.make(sink, handler, options). The wakeup reads the outbox deadline and drains it in the same acquired context as requests. A pass that fills limit reports more work, so a large backlog drains in back-to-back turns. A delivery handler must not drain its own sink. Set ordered: true to stop behind a retrying entry. With partitionId, order applies to that lane; otherwise it applies to the whole sink. The default lets later due entries pass a retry. Request drains and alarm drains share a permit for the sink.

Reconciliation at acquisition restores deadlines left by a crash between a commit and arming the alarm. An object must wake again for that recovery to happen.

ProjectionsCopy Anchor Link

Run projections in request middleware or register a wakeup to run them in an alarm turn, over the same storage Layer as their checkpoints. See Triggers. A long-lived follow fiber does not suit an object.

ProjectionWakeup.attach from @streamsy/projection acquires that wakeup in the object layer scope, before request writes. The object's own commits pick the bindings. Each commit that writes an input of the projection adds projection.at(params) to a set in memory. Its next answers "now" while the set has a binding, and its run is Projection.run(binding, { limit }) for each one:

import { Layer } from "effect";
import { Streamsy } from "@streamsy/core";
import { CheckpointStore, ProjectionWakeup } from "@streamsy/projection";
import { StreamsyObject } from "@streamsy/serve/cloudflare";
import { DurableObjectStorage } from "@streamsy/storage/durable-object";
import { Files, UserWake, filesWake } from "./wake-streams.ts";

export class FilesObject extends StreamsyObject.make({
  app: Layer.empty,
  layer: (state) =>
    CheckpointStore.layer.pipe(
      Layer.provideMerge(
        DurableObjectStorage.layer(Streamsy.group(Files, UserWake), {
          client: { storage: state.storage },
        }),
      ),
    ),
  wakeups: [ProjectionWakeup.attach(filesWake)],
}) {}

The set is lost when the object stops, so the wakeup is durable by default. Each commit also writes a pending mark for each binding it touches, in the same transaction, to streamsy_projection_pending. After an eviction or a deploy, acquisition reads the marks and arms the alarm, and the alarm runs those bindings. A binding that catches up deletes its mark. A binding that fails waits alone, from 1 s doubling to 1 hour, while its neighbours run, and Wakeups.status counts it as deferred. Pass durable: false to keep only the in-memory set. Then a binding whose run did not happen before the object stopped is late until its source gets its next write.

The object reconciles its wakeups after each write request, after each alarm pass and when it is acquired, so a source write arms the alarm at once. A Projection.wake binding can write a wake stream on another object through an object link.

Streams on other objectsCopy Anchor Link

An object can use a stream that another object owns. The request goes through the owner's object stub, and never to the network.

ObjectLink.make declares both sides of an internal mount with one value:

import { ObjectLink, Placement, rule, StreamsyObject } from "@streamsy/serve/cloudflare";

const wakeLink = ObjectLink.make({
  group: Streamsy.group(UserWake),
  prefix: "/__wake",
  placement: Placement.byRoute([rule({ stream: UserWake, owner: ({ user }) => user })]),
  access: "full",
});

// The owner serves the mount.
export class UserObject extends StreamsyObject.make<Env, UserWake>({
  app: Layer.mergeAll(wakeLink.routes, Serve.mount(UserWake, "/users/:user/wake")),
  layer: userLayer,
}) {}

// The caller gets `UserWake` from the link.
export class FolderObject extends StreamsyObject.make<Env, Files | CheckpointStore, UserWake>({
  app,
  layer: folderLayer,
  links: (env) => [wakeLink.client(env.USERS)],
  wakeups: [ProjectionWakeup.attach(filesWake)],
}) {}

The caller client sends a request only for a stream in the group, under the prefix. It answers other paths with 404 and calls no object. It declares producer and expected-offset support, because the owner is always a Streamsy mount.

Use full for a wake stream with params, because the producer creates it with PUT on its first 404. Use append when the owner creates each stream at start.

The host builds the links under the object's layer and beside it. A service in layer can use a linked stream at start. If layer also provides a linked stream, for example because the storage group holds it, the object does not start. Each request answers 503, and the log names the stream. Without this check, rows would go to a local stream that no client reads.

To read a stream that another object already serves on a public mount, use ObjectStreams.layer without a link:

const folderFiles = (env: Env) =>
  ObjectStreams.layer(Streamsy.group(Files), {
    objects: env.FOLDERS,
    placement,
    prefix: "/streams",
  });

The link trusts every caller that holds a stub. Only code in the same Worker script holds one.

AlchemyCopy Anchor Link

@streamsy/serve/alchemy builds the same object for an Alchemy stack.

import * as Cloudflare from "alchemy/Cloudflare";
import { Effect, Layer } from "effect";
import { Streamsy, Http } from "@streamsy/core";
import * as Host from "@streamsy/serve/alchemy";
import { DurableObjectStorage } from "@streamsy/storage/durable-object";

const streams = Streamsy.group(Streamsy.Wildcard);

export class Streams extends Cloudflare.DurableObject<Streams>()(
  "Streams",
  Effect.gen(function* () {
    const state = yield* Cloudflare.DurableObjectState;
    return Effect.suspend(() =>
      Host.objectHandlers({
        app: Http.routes(streams, { prefix: "/streams", access: "full" }),
        layer: Layer.mergeAll(
          DurableObjectStorage.layer(streams, { client: { storage: state.raw.storage } }),
          Host.alarmSlotLayer(state.raw.storage),
        ),
      }),
    );
  }),
) {}

The outer generator declares resources. The inner Effect acquires storage for each instance. objectHandlers shares one acquisition across fetch and alarm, and releases it when the instance scope closes.

The Worker gets the Streams namespace during construction. Its fetch uses Host.router({ objects, prefix: "/streams", placement }). The router is an HttpEffect, so authentication can run before it.

The pinned Alchemy version is 2.0.0-beta.83.

objectHandlers takes links too, as a plain array: links: [link.client(Host.objectFetch(objects))]. Alchemy stubs answer with an Effect, so wrap the binding with Host.objectFetch(objects) before you give it to link.client or ObjectStreams.layer. Both hosts use the same ObjectLink and LinkClient.

ObjectFetch returns Effect<Response, HttpClientError>. ObjectStreams uses HttpClient.make so caller context, tracing and cancellation reach the stub. Native Cloudflare clients can take a namespace or this Effect transport.

A remote client uses RemoteStreams.layer with the Worker protocol prefix as baseUrl.

Projection servicesCopy Anchor Link

Compose CheckpointStore.layer and Transaction.layer from @streamsy/projection over this storage Layer with Layer.provideMerge. They provide CheckpointStore and Transaction on the same SQL owner. Commit.transaction commits SQL, output appends, and offsets together. See Checkpoints and generations.