Layers and storage

A Streamsy program is an Effect that needs stream services. A Layer provides them and decides where the data is kept.

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

const runtime = ManagedRuntime.make(MemoryStorage.layer(shop));
try {
  await runtime.runPromise(write);
} finally {
  await runtime.dispose();
}

LayersCopy Anchor Link

Streamsy.Host<typeof group> names the stream services, so one type fits any host Layer. Each host module names its options Options, for example BunStorage.Options.

MemoryCopy Anchor Link

MemoryStorage.layer(group, options) keeps streams in the process. The data is lost when the Layer is disposed.

Each mutation copies the whole store, so an append takes linear time. Use memory for development and tests. Use constrained: true in tests to check that a program does not depend on the default capabilities.

Several backendsCopy Anchor Link

Merge one Layer for each backend:

import { Layer } from "effect";
import { MemoryStorage, Streamsy } from "@streamsy/core";
import { BunStorage } from "@streamsy/storage/bun";
import { Drafts, Journal } from "./streams.ts";

export const layer = Layer.merge(
  MemoryStorage.layer(Streamsy.group(Drafts)),
  BunStorage.layer(Streamsy.group(Journal), { client: { filename: "./journal.sqlite" } }),
);

A missing stream service is a type error.

Rules:

• A group checks overlap only among its own streams. Put all served streams in one group.
• A fork stays inside one storage Layer.
• A remote Layer has its own capabilities.
• Commit.transaction keeps SQL, output appends, and the checkpoint on one SQL owner. Commit.producer can write to existing outputs on other owners. See Several backends.

Commit with your own tablesCopy Anchor Link

Every storage Layer gives CommitBoundary. On the SQLite Layers, withTransaction runs application SQL and a stream mutation in one transaction.

import { Effect } from "effect";
import { CommitBoundary, Storage, type Mutation } from "@streamsy/core";
import * as SqlClient from "effect/sql/SqlClient";

declare const mutation: Mutation;

const program = Effect.gen(function* () {
  const boundary = yield* CommitBoundary;
  const storage = yield* Storage;
  const sql = yield* SqlClient.SqlClient;
  return yield* boundary
    .withTransaction(
      Effect.gen(function* () {
        yield* sql.unsafe("INSERT INTO application_rows(id) VALUES (?)", ["example"]);
        return yield* storage.mutate(mutation);
      }),
    )
    .pipe(Effect.catchTag("MutationRejected", Effect.succeed));
});

A MutationRejected that leaves the boundary rolls back both writes. Catch it outside the boundary. Nested calls join the outer transaction and make no savepoints.

TestingCopy Anchor Link

Application tests provide MemoryStorage.layer(group) and run the same program.

Bind streams to a test graph with Protocol.bind(layerTest(), group).

Protocol tierCopy Anchor Link

Storage adapters, protocol tests, and HTTP internals use these.