Durable State

@streamsy/state is a typed layer on top of JSON mode. Instead of arbitrary JSON values, a Durable State stream carries change messages (insert / update / delete over named collections) and control messages (snapshot boundaries, reset). The message shape follows the Durable State synchronization convention, so application code can replay the stream into local collections. This package provides the typed wire vocabulary over Streamsy; it is not a complete browser sync runtime.

bun add @streamsy/state

Define a schema

A schema maps each collection name to a value codec/Standard Schema, an optional wire type, and a primaryKey:

import { createMemoryStorageAdapter, createStreamProtocol } from "@streamsy/core";
import { createDurableStateProtocol, type JsonCodec } from "@streamsy/state";

type Issue = { id: string; title: string; status: "open" | "closed" };

const issueCodec: JsonCodec<Issue> = {
  encode: (value) => value,
  decode(value) {
    const v = value as Partial<Issue>;
    if (typeof v?.id !== "string" || typeof v?.title !== "string") {
      throw new Error("invalid issue");
    }
    return { id: v.id, title: v.title, status: v.status ?? "open" };
  },
};

const protocol = createStreamProtocol({ storage: { adapter: createMemoryStorageAdapter() } });
const state = createDurableStateProtocol(protocol, {
  issues: { type: "issue", schema: issueCodec, primaryKey: "id" },
});

• schema is any JsonSchema<T> — a JsonCodec<T> or a Standard Schema (Zod, Valibot, …).
• type is the on-the-wire type tag; it defaults to the collection name.
• primaryKey is a property name ("id") or a function (value) => string for derived keys (e.g. (v) => `issue:${v.id}`).

Apply changes

create / get return a DurableStateStream<S> whose .state exposes the typed mutators:

const created = await state.create("workspace-42");
if (created.status !== "created" && created.status !== "exists") {
  throw new Error(`create failed: ${created.status}`);
}
const stream = created.stream; // DurableStateStream<S>

await stream.state.insert("issues", { id: "i1", title: "Bug", status: "open" });
await stream.state.update(
  "issues",
  { id: "i1", title: "Bug", status: "closed" },
  { oldValue: { id: "i1", title: "Bug", status: "open" } },
);
await stream.state.delete("issues", "i1");

DurableState<S> methods:

• insert(type, value, { key?, headers? })
• update(type, value, { key?, oldValue?, headers? })
• delete(type, key, { oldValue?, headers? })
• snapshotStart({ offset?, headers? }), snapshotEnd(...), reset(...) — wire-level control messages
• append(message) — append a pre-built DurableStateMessage (validated)

Keys are derived from the value via primaryKey unless you pass an explicit key. Values are validated through the codec before append, so an invalid value throws rather than writing a bad message. The optional headers carry sync metadata (txid, timestamp, from).

The snapshot/reset methods only encode and append the Durable State control vocabulary. Streamsy does not currently build snapshots, compact retained history, bootstrap a client from a snapshot, or recover a stale cursor automatically; applications adopting those controls must define that lifecycle themselves.

Message shape

Change messages serialize to the standard durable-state envelope; control messages omit type/key:

// insert
{ type: "issue", key: "i1", value: { id: "i1", title: "Bug", status: "open" },
  headers: { operation: "insert" } }

// update (with optional old_value)
{ type: "issue", key: "i1", value: { /* ... */ }, old_value: { /* ... */ },
  headers: { operation: "update" } }

// delete
{ type: "issue", key: "i1", headers: { operation: "delete" } }

// control
{ headers: { control: "snapshot-start", offset: "2_0" } }

Read and materialize

Reads decode back into typed change/control messages, so a client can fold them into local state:

const read = await stream.read();
if (read.status === "ok") {
  for (const { value } of read.messages) {
    if ("type" in value) {
      // change message: value.type, value.key, value.value, value.headers.operation
    } else {
      // control message: value.headers.control
    }
  }
}

DurableStateStream<S> also exposes readLive(options), metadata(), delete(), and the underlying json / stream handles for byte- or JSON-level access. Application code can fold these messages into its own collections. Streamsy does not currently ship a generic TanStack DB adapter in @streamsy/state.

For the design behind that client fold, read Super Simple Sync with Durable State and TanStack DB. The repository's issue-tracker demo shows the end-to-end multi-user shape using Streamsy as transport and the official @durable-streams/state/db package for its browser StreamDB/TanStack DB adapter.