Write a storage adapter

Storage is one Effect service. An adapter that implements it and passes the contract tests is a backend for every Streamsy program.

The protocol, HTTP, forks, expiry, producer fencing, and projection checkpoints are above Storage. An adapter stores records and messages and applies mutations.

The contractCopy Anchor Link

import { Effect, Layer, Option, Stream } from "effect";
import { Storage, type StorageShape } from "@streamsy/core";

const adapter: StorageShape = {
  capabilities: { fork: "none", atomicScope: "stream", wake: "poll", expiryIndex: "lazy" },
  record: (id) => Effect.succeed(Option.none()),
  messages: (id, window) => Effect.succeed([]),
  producer: (id, producerId) => Effect.succeed(Option.none()),
  mutate: (mutation) => Effect.die("not implemented"),
  changes: (id) => Stream.empty,
  nextExpiry: Effect.succeed(Option.none()),
};

export const layer = Layer.succeed(Storage, adapter);

Each read must return new values, so a caller cannot change the store through a returned object.

Bind streamsCopy Anchor Link

import { Layer } from "effect";
import { Protocol } from "@streamsy/core/protocol";
import { layer } from "./adapter.ts";
import { shop } from "./streams.ts";

export const shopStorage = Protocol.bind(Protocol.layer().pipe(Layer.provideMerge(layer)), shop);

shopStorage works like MemoryStorage.layer.

CapabilitiesCopy Anchor Link

MutationsCopy Anchor Link

A mutation is a non-empty ordered list of Create, Append, and Delete operations.

Check every precondition before the mutation, in the same transaction. On the first failure, write nothing and answer MutationRejected { index, reason, record }.

The protocol maps each reason to a typed error. An I/O failure is StorageFault, with retryable.

WakesCopy Anchor Link

Each wake carries a ChangeSnapshot with present, currentOffset, closed, and softDeleted. A wake with no change is valid.

• Register the subscriber before the first snapshot.
• A push adapter keeps one pending wake for each subscriber.
• Interrupting the subscriber, or closing the owner, releases the subscription.

Forks and expiryCopy Anchor Link

A chain adapter stores the source id and fork offset, so a child read can join the source prefix. A removed source with live children is soft-deleted until the last child goes.

An indexed adapter stores each deadline. An expiry delete carries the deadline it saw. After expiry-mismatch, the next nextExpiry must move past that deadline.

Commit boundaryCopy Anchor Link

To offer CommitBoundary.withTransaction, an adapter needs one transaction owner, a way for mutate to join it, and wakes that publish after the outer commit. CommitBoundary comes from @streamsy/core.

CheckpointStore.layer from @streamsy/projection needs CommitBoundary and OwnerRecords from @streamsy/core/internal. OwnerRecords reads, writes and removes encoded strings by key. Calls inside withTransaction join that transaction. An adapter that wants projections provides both services from its storage Layer.

Contract testsCopy Anchor Link

import { StorageContract } from "@streamsy/core/testing";
import { layer } from "./adapter.ts";

StorageContract.run({
  name: "my adapter",
  layer,
  expected: { fork: "none", atomicScope: "stream", wake: "poll", expiryIndex: "lazy" },
});

Run it under Bun. The kit builds a fresh Layer for each case, gives a TestClock, and skips cases that expected rules out. Set ownerRecords: true when the Layer also provides CommitBoundary and OwnerRecords. The records cases then check that records and stream writes commit and roll back together.

Then run the upstream server conformance suite against Http.routes(Streamsy.group(Streamsy.Wildcard), { access: "full" }) over the adapter.

Shipped adaptersCopy Anchor Link

Projection storageCopy Anchor Link

A custom checkpoint Layer can implement EncodedStore and call fromStore from @streamsy/projection/checkpoint. The record contains encoded state and offsets. Commit.transaction also needs a SQL Transaction; a storage capability flag alone does not provide it. See Custom Commits.