Serve handlers

Serve handlers return response Effects. Mount them alongside app routes with HttpRouter.add, Serve.route, or Serve.mount.

import { Layer, Schema } from "effect";
import { Http, Streamsy } from "@streamsy/core";
import { Serve } from "@streamsy/serve";

class Events extends Streamsy.json<Events>()("Game/Events", "events/:seat", {
  params: { seat: Schema.String },
  schema: Schema.String,
}) {}
class Cards extends Streamsy.state<Cards>()("Game/Cards", "cards/:seat", {
  params: { seat: Schema.String },
  collections: { cards: { schema: Schema.Struct({ id: Schema.String }), key: "id" } },
}) {}
export const game = Streamsy.group(Events, Cards);

export const App = Layer.mergeAll(
  Http.routes(game, { prefix: "/streams", access: "read" }),
  Serve.mount(Events, "/feed/:seat"),
  Serve.mount(Cards, "/state/:seat"),
);

Route	Serves
`Serve.stream(address)`	The items of a stream
`Serve.state(address)`	The Durable State change stream
`Serve.value(schema, effect)`	One JSON value, with an ETag

Serve.mount(Stream, path) decodes path params and chooses the stream or state handler. Serve.route(path, handler) owns GET and HEAD and rejects other methods. Join routes with Layer.mergeAll, then serve them with Serve.layer on a long-running host, or with HttpRouter.toWebHandler.

Request	Response
GET, HEAD	The data
Other methods	405 with `allow: GET, HEAD`
Parameter does not decode	400
Path does not match	404 from the router
Absent stream	404
Unsaved projection value	The declared initial value

The stream codecs decode the path parameters. Two routes with the same shape fail the Layer build. A Serve route does not create a stream, checkpoint, or state row.

State routes

Serve.state serves the change stream with no snapshot. A new client reads from -1. On replay-from-start, clear local rows and replay from -1.

Stored control messages (snapshot-start, snapshot-end and reset) are served as they are, in order with the changes. A client should clear its rows at a reset and keep reading forward. Serve.state does not make snapshots, so it refuses a request with the x-streamsy-state-reset: snapshot header.

Projection outputs

Pass resolved addresses to Serve.state or Serve.stream. Compute projection values with Projection.read and Effect.map.

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

const params = { seat: Schema.FiniteFromString };
class Numbers extends Streamsy.json<Numbers>()("Seats/Numbers", "numbers/:seat", {
  params,
  schema: Schema.Finite,
}) {}
const Card = Schema.Struct({ id: Schema.String, total: Schema.FiniteFromString });
class Board extends Streamsy.state<Board>()("Seats/Board", "board/:seat", {
  params,
  collections: { card: { schema: Card, key: "id" } },
}) {}
class Transitions extends Streamsy.json<Transitions>()("Seats/Transitions", "transitions/:seat", {
  params,
  schema: Schema.FiniteFromString,
}) {}
const tracker = Projection.mapAccumBatch({
  commit: Commit.producer,
  id: "tracker",
  params,
  inputs: { numbers: Numbers },
  outputs: {
    board: Board,
    transitions: Transitions,
  },
  state: { schema: Schema.Struct({ total: Schema.FiniteFromString }), initial: { total: 0 } },
  mapAccumBatch: (state, batch) =>
    Effect.sync(() => {
      const total = state.total + batch.numbers.items.reduce((sum, n) => sum + n, 0);
      return [
        { total },
        {
          board: [
            DurableState.upsert("card", { id: "total", total }),
            DurableState.remove("card", "old"),
          ],
          transitions: [total],
        },
      ] as const;
    }),
});
const OutputRoutes = Layer.mergeAll(
  Serve.mount(Board, "/board/:seat"),
  Serve.mount(Transitions, "/transitions/:seat"),
  Serve.route(
    "/summary/:seat",
    Effect.flatMap(Serve.params(Board), (p) =>
      Serve.value(
        Schema.Struct({ total: Schema.FiniteFromString }),
        Projection.read(tracker.at(p)).pipe(Effect.map(({ state }) => state)),
      ),
    ),
  ),
);

Serve does not run the projection. Projection.read returns the declared initial value until state is saved.

The route app needs the output streams and the Commit's required services for the value. A value schema must encode to JSON.

Auth and request parameters

const Mine = Serve.route(
  "/me",
  Effect.flatMap(CurrentSeat, (seat) => Serve.stream(Events.at({ seat: seat.id }))),
).pipe(Layer.provide(SeatAuth.layer));

SeatAuth is HttpRouter.middleware that checks the token. It answers 401, or provides CurrentSeat. The handler builds the address from that service.

Protect the protocol routes too, if they expose private data.

Values

Serve.value(schema, effect) accepts any computation. Its schema must encode to JSON. Errors remain catchable in the Effect error channel.

Behaviour	Value
Body	Canonical JSON
ETag	From the body bytes. Strong and weak validators
Cache-Control	`private, max-age=0, must-revalidate`

| Contract | The x-streamsy-value-contract header |

HttpApiBuilder

handleRaw accepts the same handler. Convert respondable errors to responses to retain Serve's headers; declared HttpApi errors use HttpApi's encoder.

import { Effect, Schema } from "effect";
import { HttpServerRespondable } from "effect/http";
import { HttpApi, HttpApiBuilder, HttpApiEndpoint, HttpApiGroup } from "effect/http-api";

const Api = HttpApi.make("Game").add(
  HttpApiGroup.make("feeds").add(
    HttpApiEndpoint.get("feed", "/api/feed/:seat", { params: { seat: Schema.String } }),
  ),
);
const Feeds = HttpApiBuilder.group(Api, "feeds", (handlers) =>
  handlers.handleRaw("feed", ({ params }) =>
    Serve.stream(Events.at(params)).pipe(Effect.catch(HttpServerRespondable.toResponse)),
  ),
);

Browser clients and CORS

@streamsy/serve/contract exports the header names and decodePublicError. It does not load Effect.

const Cors = HttpRouter.cors({
  allowedOrigins: ["https://app.example.com"],
  allowedMethods: ["GET", "HEAD", "OPTIONS"],
  allowedHeaders: [
    "authorization",
    "x-streamsy-state-version",
    "x-streamsy-state-contract",
    "x-streamsy-stream-version",
    "x-streamsy-stream-contract",
    "x-streamsy-value-contract",
  ],
  exposedHeaders: [
    "stream-next-offset",
    "stream-cursor",
    "stream-up-to-date",
    "etag",
    "x-streamsy-state-version",
    "x-streamsy-state-contract",
    "x-streamsy-stream-version",
    "x-streamsy-stream-contract",
    "x-streamsy-value-contract",
  ],
});
const BrowserApp = Layer.mergeAll(App, Cors);

Error kind	Client recovery
Stream, state	Replay from the start
Value	Fetch again

A contract fingerprint covers the stream address and the state collection names and keys. Set options.contract when the Schema or computation changes and the metadata does not.

Use URL-safe stream segments. Serve decodes parameters, and the protocol routes use raw encoded ids.

Long-running hosts

Serve.layer(routes, options) is HttpRouter.serve with a drain for live reads. Serve.teardown sets the exit code. Both come from @streamsy/serve and work on any runtime. The host adds its listener and its runtime's runMain.

import { Layer } from "effect";
import { BunRuntime } from "@effect/platform-bun";
import { MemoryStorage } from "@streamsy/core";
import { Serve } from "@streamsy/serve";
import { listener } from "@streamsy/serve/bun";
import { App, game } from "./app.ts";

const Server = Serve.layer(App, { drainTimeout: "2 seconds" }).pipe(
  Layer.provide(MemoryStorage.layer(game)),
  Layer.provide(listener({ port: 3000 })),
);

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

options takes the options of HttpRouter.serve, plus drainTimeout (default 2 s).

At shutdown:

The drain signals every live read. A parked long poll answers 204 with its offset and cursor, as after a long-poll timeout. An SSE body ends after its last batch.
The drain waits until no live read is running, or until drainTimeout passes. A live request in this window gets the same reply at once.
The serve Layer closes and interrupts the handlers that still run.
Storage closes, then the listener stops.

Exit	Code with `Serve.teardown`
Interrupted by `SIGINT` or `SIGTERM`	0
Success	0
Failure, such as a port in use	Not 0 (1 by default)

Without Serve.teardown, runMain exits 130 after a signal. Serve.teardown maps every interrupt-only exit to 0. Use it for a main that ends only by failure or signal, such as Layer.launch.

A Durable Object host gets no stop signal. It uses HttpRouter.toWebHandler, and neither Serve.layer nor Serve.teardown.

Serve handlers

State routesCopy Anchor Link

Projection outputsCopy Anchor Link

Auth and request parametersCopy Anchor Link

ValuesCopy Anchor Link

HttpApiBuilderCopy Anchor Link

Browser clients and CORSCopy Anchor Link

Long-running hostsCopy Anchor Link