# ReflectDB documentation Package version: 0.4.1. --- Source: https://reflectdb.dev/docs/README.md # reflectdb [![npm version](https://img.shields.io/npm/v/reflectdb.svg)](https://www.npmjs.com/package/reflectdb) [![npm downloads](https://img.shields.io/npm/dm/reflectdb.svg)](https://www.npmjs.com/package/reflectdb) [![CI](https://github.com/TimMikeladze/reflectdb/actions/workflows/ci.yml/badge.svg)](https://github.com/TimMikeladze/reflectdb/actions/workflows/ci.yml) [![types](https://img.shields.io/npm/types/reflectdb.svg)](https://www.typescriptlang.org) [![license](https://img.shields.io/npm/l/reflectdb.svg)](https://github.com/TimMikeladze/reflectdb/blob/main/LICENSE) A real-time sync engine for TypeScript. Keeps a server-side database in sync with any number of browser clients — offline-first, with optimistic local writes, automatic conflict resolution, and end-to-end type inference. You bring your own types and your own database. reflectdb handles the protocol, the op log, conflicts, reconnection, and subscriptions. ``` ┌──────────────┐ writes ┌──────────────┐ writes ┌──────────────┐ │ Browser A │ ──────────▶ │ Server │ ◀────────── │ Browser B │ │ (optimistic) │ deltas │ (authoritive)│ deltas │ (optimistic) │ │ │ ◀────────── │ │ ──────────▶ │ │ └──────────────┘ └──────────────┘ └──────────────┘ ▲ │ │ offline ▼ └────── IndexedDB ───── op log (in-memory / SQLite / Postgres / S3) ``` ## Table of Contents - [Demos](#demos) - [Why reflectdb](#why-reflectdb) - [Features](#features) - [Use Cases](#use-cases) - [Installation](#installation) - [Claude Code and Codex skill](#claude-code-and-codex-skill) - [Quick Start](#quick-start) - [Recipes](#recipes) - [WebSocket sync with SQLite + Drizzle](#websocket-sync-with-sqlite--drizzle) - [Typed params for multi-tenant queries](#typed-params-for-multi-tenant-queries) - [Authentication and room-based access control](#authentication-and-room-based-access-control) - [Per-column merge for collaborative editing](#per-column-merge-for-collaborative-editing) - [Custom conflict resolvers](#custom-conflict-resolvers) - [Validating client payloads](#validating-client-payloads) - [Ephemeral messages (cursors, presence, typing)](#ephemeral-messages-cursors-presence-typing) - [Typed presence](#typed-presence) - [Peer-to-peer (WebRTC)](#peer-to-peer-webrtc) - [Per-user query results](#per-user-query-results) - [Read-only views](#read-only-views) - [Server-driven game loops](#server-driven-game-loops) - [Transactional writes with `server.tx`](#transactional-writes-with-servertx) - [Windowed sync and pagination](#windowed-sync-and-pagination) - [Auto-generated REST API](#auto-generated-rest-api) - [High availability with Postgres](#high-availability-with-postgres) - [Sync with no database at all](#sync-with-no-database-at-all) - [Serverless sync on Vercel](#serverless-sync-on-vercel) - [htmx 4 bindings](#htmx-4-bindings) - [Whiteboard + Pictionary example](#whiteboard--pictionary-example) - [Infinite Tetris example](#infinite-tetris-example) - [Multiplayer kanban example](#multiplayer-kanban-example) - [htmx todos example](#htmx-todos-example) - [Architecture](#architecture) - [Core Concepts](#core-concepts) - [Hybrid Logical Clocks](#hybrid-logical-clocks) - [Conflict Resolution](#conflict-resolution) - [The Sync Protocol](#the-sync-protocol) - [Two stores, one sync](#two-stores-one-sync) - [The Op Log and Resume](#the-op-log-and-resume) - [API Reference](#api-reference) - [`reflectdb/core`](#reflectdbcore) - [`reflectdb/server`](#reflectdbserver) - [`reflectdb/server/storage/object`](#reflectdbserverstorageobject) - [`reflectdb/client`](#reflectdbclient) - [`reflectdb/react`](#reflectdbreact) - [`reflectdb/svelte`](#reflectdbsvelte) - [`reflectdb/vanilla`](#reflectdbvanilla) - [`reflectdb/htmx`](#reflectdbhtmx) - [`reflectdb/transport/*`](#reflectdbtransport) - [Configuration Reference](#configuration-reference) - [Query definition](#query-definition) - [View definition](#view-definition) - [Presence definition](#presence-definition) - [Server configuration](#server-configuration) - [`implement()` options](#implement-options) - [Rate limiting](#rate-limiting) - [Compaction](#compaction) - [Client configuration](#client-configuration) - [Storage adapters](#storage-adapters) - [Transport configuration](#transport-configuration) - [Development](#development) - [License](#license) ## Demos | Demo | Try it | What it demonstrates | |------|--------|----------------------| | **Huddle (WebRTC)** | [Open a room](https://reflectdb-huddle.fly.dev/) · [source](https://github.com/TimMikeladze/reflectdb/blob/main/examples/huddle/) | Sticky notes, live cursors, live ink and reactions. Cursors, drags and ink stream peer-to-peer through `` with reflectdb carrying the signaling, reactions never touch the server, and a latency race times each peer directly against the round trip through the server. Press "open a second tab" in the room. | | **Infinite multiplayer Tetris** | [Play live](https://reflectdb-tetris.fly.dev/) · [source](https://github.com/TimMikeladze/reflectdb/blob/main/examples/tetris/) | Optimistic input prediction, server reconciliation and gravity, a live leaderboard, per-player progression, and Bun SQLite persistence in one perpetual game. Open two tabs to add another player. | | **Multiplayer kanban** | [Open the board](https://reflectdb-kanban.vercel.app/) · [source](https://github.com/TimMikeladze/reflectdb/blob/main/examples/kanban/) | A board whose entire durable state is an S3 bucket — no Postgres, no SQLite, no volume — running on Vercel functions. Shows leaseless optimistic concurrency and serverless SSE. Open two tabs and drag a card. | | **htmx 4 todos** | [Open it](https://reflectdb-htmx-todos.fly.dev/) · [source](https://github.com/TimMikeladze/reflectdb/blob/main/examples/htmx-todos/) | htmx owns the DOM, reflectdb owns the data. Attributes point at `reflect:` actions instead of server routes, so the server renders no HTML at all — every fragment comes from the local store. Open two tabs, or stop typing and go offline. The list resets to its seed rows every minute. | | **Collaborative whiteboard** | [Draw live](https://reflectdb-whiteboard.fly.dev/) · [source](https://github.com/TimMikeladze/reflectdb/blob/main/examples/whiteboard/) | Freeform drawing by default, optional Pictionary rounds, guest-authenticated rooms, ephemeral cursors, chat, presence, and per-user query results. Rooms and everything in them are deleted 30 minutes after they are created. Open two tabs to draw with yourself. | Huddle, Tetris, the whiteboard and the htmx todos each run on one auto-stopping Fly Machine with no volume, so the first load after an idle period may take a moment. The kanban board has no machine to wake — it is Vercel functions and a bucket — but every board resets on a five-minute window, and the htmx todos reset every minute. All five keep their data intentionally ephemeral across deployments and Machine replacement. ## Why reflectdb Most real-time sync libraries force you to choose: CRDTs (powerful but opaque), or simple pub/sub (fast but brittle). reflectdb sits in the middle — **per-row operations** with **hybrid logical clocks** for causal ordering, validated through a server-side pipeline so your database stays authoritative. You define your schema once, and the same types flow to both sides: ```ts const { rows, insert } = useSync("todos"); // ^? Todo[] ^? (id, { title, done, createdAt? }) => void ``` No code generation. No glue layer. No second source of truth. **Bring your own stack.** reflectdb is agnostic about: - **Your database** — any TypeScript ORM, raw SQL driver, Map, or REST API works. The `query`/`mutate` callbacks hand you `db` untouched. - **Your row types** — plain TypeScript types, Drizzle `$inferSelect`, Kysely, Prisma, anything. Declare them with `t()`. - **Your HTTP server** — Bun, Node, Deno, Cloudflare Workers, anything fetch-compatible. Transports expose handler functions you wire to routes. Optional bits (use what you want): - **Drizzle ORM** — if you point `table` at a Drizzle table, row types are auto-inferred. - **Server op log storage** — SQLite (single-node), Postgres (HA), or an S3-compatible bucket (no database at all). Omit it and the op log is in-memory. - **React / Svelte bindings** — use the core client directly if you prefer. ## Features - **Real-time sync** over WebSocket, Server-Sent Events, or HTTP long-polling - **Offline-first** — optimistic local writes, queued and replayed on reconnect - **End-to-end type safety** — schema defines row types, query params, writable fields, and which columns the server owns - **Per-row and per-column conflict resolution** — `lww`, `merge`, `server`, or a custom resolver - **Causal ordering** via hybrid logical clocks (HLC) — no dependence on synchronized wall clocks - **Pluggable storage** — in-memory, SQLite, Postgres, or S3-compatible object storage for the server op log; memory or IndexedDB for the browser - **Auto-generated REST** — `server.rest()` turns your schema into CRUD endpoints that broadcast deltas - **Room-based access control** — scope clients to `org/:orgId` or arbitrary patterns - **Rate limiting** — global and per-table, fail-open - **Op log compaction** — configurable retention for old accepted ops - **High availability** — shared Postgres + optional cross-instance polling - **No database at all** — `createObjectStorage` runs a room with an S3-compatible bucket as the only durable store, group-committing one object per batch - **Runs serverless** — SSE in `serverless` mode answers each POST with the replies it produced, so sync works on Vercel, Lambda or Workers - **Framework bindings** — React hooks, Svelte stores, a vanilla-JS helper, and htmx 4 attribute bindings; the core client works anywhere - **Ephemeral channels** — presence, cursors, typing indicators that never touch the op log, with a room snapshot on join and a pluggable adapter (Redis included) so presence spans a fleet - **Typed presence** — `presence()` in the schema, `usePresence()` in the component, key derived for you - **Peer-to-peer (WebRTC)** — cursors, drags and game ticks go browser-to-browser at full rate, plus calls and screen share, all signaled over the sync connection you already have - **Read-only views** — `view()` entries that recompute on their dependencies and reject writes at both levels - **Windowed sync** — paginate large tables with `loadMore` + `useTotalCount` - **Server-side toolkit** — `tx` (transaction + auto-notify), `lock` / `tryLock`, and self-disposing `interval` / `timeout` ## Use Cases - Collaborative editing (docs, whiteboards, spreadsheets) - Multi-device note apps, todo apps, inbox-like UIs - Live dashboards where multiple clients view and edit the same state - Local-first apps that need to work offline and merge on reconnect - Admin tools that should "just update" when someone else changes a row - Field-service or retail apps on spotty networks - Games or canvases with presence indicators and live cursors - Serverless deployments with no database to attach and no machine to keep warm ## Installation ```bash bun add reflectdb # or npm install reflectdb ``` Ships both ESM and CommonJS, so `import` and `require` both work: ```ts import { defineSyncQueries, t } from "reflectdb"; // or "reflectdb/core" ``` ```js const { defineSyncQueries, t } = require("reflectdb"); ``` Everything else is a subpath — `reflectdb/server`, `reflectdb/client`, `reflectdb/react`, and so on. The bare `reflectdb` specifier is an alias for `reflectdb/core`, the surface both sides share. Peer dependencies are all optional: ```bash bun add react # for reflectdb/react bun add htmx.org@^4 # for reflectdb/htmx bun add drizzle-orm # if you want auto-inferred row types from Drizzle tables # Svelte + vanilla have no peer deps ``` ## Claude Code and Codex skill The [ReflectDB skill](https://reflectdb.dev/skills/reflectdb/SKILL.md) covers setup, typed queries, storage, transports, presence, and sync debugging. One skill folder works in both tools. This checkout links it into each tool's project skill directory. ```text # Claude Code /reflectdb add offline todo sync to this app # Codex $reflectdb add offline todo sync to this app ``` Claude Code exposes skills as slash commands; Codex uses `$reflectdb` or the `/skills` picker. See the [Claude Code skill docs](https://code.claude.com/docs/en/skills) and [Codex skill docs](https://developers.openai.com/codex/skills/). To use the skill in another project, copy `skills/reflectdb` from this checkout into `.agents/skills/reflectdb`, then link Claude Code to that same folder: ```bash mkdir -p .agents/skills .claude/skills cp -R /path/to/reflectdb/skills/reflectdb .agents/skills/reflectdb ln -s ../../.agents/skills/reflectdb .claude/skills/reflectdb ``` Run these commands in the destination project when neither `reflectdb` skill path already exists. The npm package also includes `skills/reflectdb`, so an installed copy is available at `node_modules/reflectdb/skills/reflectdb`. Restart the tool if it has not discovered the new skill. The landing page serves an [LLM documentation index](https://reflectdb.dev/llms.txt) and [complete text documentation](https://reflectdb.dev/llms-full.txt), including this skill. `/llm.txt` is an alias for `/llms.txt`. The landing dev and build commands regenerate the files from the repository README, feature docs, and skill. Edit [llms.txt](https://github.com/TimMikeladze/reflectdb/blob/main/llms.txt) to change the index, or run `bun run --cwd landing docs:llms` to regenerate the public files directly. ## Quick Start A complete sync server in ~30 lines. No ORM, no database — just plain types and an in-memory Map, handed to reflectdb as `db`. ### 1. Define your schema ```ts // schema.ts import { defineSyncQueries, t } from "reflectdb/core"; export type Todo = { id: string; title: string; done: boolean; createdAt: Date; }; export const queries = defineSyncQueries({ todos: { row: t(), conflict: "lww", serverSet: ["createdAt"], // server always sets this, clients cannot }, }); ``` ### 2. Create the server ```ts // server.ts import { serve } from "bun"; import { createSyncServer } from "reflectdb/server"; import { createWsServerTransport } from "reflectdb/transport/ws"; import { queries, type Todo } from "./schema"; const todos = new Map(); const transport = createWsServerTransport(); // `db` is whatever holds your data — an ORM handle, a pool, or a plain Map. // It's handed to every `query` and `mutate`. Optional: a query that closes over // in-memory state runs without one. const server = createSyncServer({ queries, db: todos, transport, serverId: "s1" }); server.auth(async (req) => { // validate req.headers.get("authorization") return { userId: "user-1" }; }); server.implement("todos", { query: (_ctx, db) => [...db.values()], mutate: async (op) => { if (op.type === "delete") todos.delete(op.rowId); else todos.set(op.rowId, { id: op.rowId, ...(op.payload as Partial) } as Todo); }, serverSet: { createdAt: () => new Date() }, }); // Wire WebSocket handlers to your HTTP server serve({ port: 3001, fetch(req, srv) { const url = new URL(req.url); if (url.pathname === "/sync") { const clientId = crypto.randomUUID(); if (srv.upgrade(req, { data: { clientId } })) return; } return new Response("ok"); }, websocket: { open(ws) { transport.handleOpen(ws.data.clientId, ws); }, message(ws, data) { transport.handleMessage(ws.data.clientId, String(data)); }, close(ws) { transport.handleClose(ws.data.clientId); }, pong(ws) { transport.handlePong(ws.data.clientId); }, }, }); ``` ### 3. Connect from the browser ```tsx // app.tsx import { SyncProvider, useSync, useSyncStatus } from "reflectdb/react"; import { createIndexedDBStorage } from "reflectdb/client/storage/indexeddb"; export function App() { return ( ); } function TodoList() { const { rows, insert, update, remove } = useSync("todos"); const status = useSyncStatus(); return (

Status: {status}

{rows.map((t) => ( ))}
); } ``` Open two tabs — edits in one appear in the other within a round-trip. Close the laptop, edit offline, reopen — pending ops replay automatically. ## Recipes The repo ships two end-to-end examples: [`examples/whiteboard/`](https://github.com/TimMikeladze/reflectdb/blob/main/examples/whiteboard/), a collaborative drawing app with two modes (freeform and **Pictionary**), and [`examples/tetris/`](https://github.com/TimMikeladze/reflectdb/blob/main/examples/tetris/), one perpetual Tetris game with no player cap. Between them they exercise the patterns below in one place. The snippets here are minimal, copy-paste-friendly references; see the examples for how they fit together. ### WebSocket sync with SQLite + Drizzle If you use Drizzle, point `table` at it and row types flow automatically. Swap the Map for `bun:sqlite` + Drizzle and add a persistent op log: ```ts import { Database } from "bun:sqlite"; import { drizzle } from "drizzle-orm/bun-sqlite"; import { sqliteTable, text, integer } from "drizzle-orm/sqlite-core"; import { eq } from "drizzle-orm"; import { defineSyncQueries } from "reflectdb/core"; import { createSyncServer, createSqliteStorage } from "reflectdb/server"; const todos = sqliteTable("todos", { id: text("id").primaryKey(), title: text("title").notNull(), done: integer("done", { mode: "boolean" }).notNull().default(false), createdAt: integer("created_at", { mode: "timestamp" }).notNull(), }); const queries = defineSyncQueries({ todos: { table: todos, conflict: "lww", serverSet: ["createdAt"] }, }); const db = drizzle(new Database("app.db")); const storage = createSqliteStorage({ path: "sync.db" }); const server = createSyncServer({ queries, db, transport, storage, serverId: "s1" }); server.implement("todos", { query: (_ctx, db) => db.select().from(todos), mutate: async (op, _ctx, db) => { if (op.type === "delete") { await db.delete(todos).where(eq(todos.id, op.rowId)); } else { await db.insert(todos) .values({ id: op.rowId, ...op.payload }) .onConflictDoUpdate({ target: todos.id, set: op.payload }); } }, serverSet: { createdAt: () => new Date() }, }); ``` ### Typed params for multi-tenant queries Declare query params with `t()` so the client must pass them and the server can use them to scope queries: ```ts import { defineSyncQueries, t } from "reflectdb/core"; type Post = { id: string; title: string; orgId: string }; const queries = defineSyncQueries({ posts: { row: t(), params: t<{ orgId: string }>(), tables: ["posts"], // change-detection hint for delta computation pk: "id", conflict: "lww", readonly: ["orgId"], // clients cannot write this }, }); // server server.implement("posts", { query: (ctx, kyselyDb) => kyselyDb.selectFrom("posts").where("orgId", "=", ctx.params.orgId).selectAll().execute(), mutate: async (op, ctx, kyselyDb) => { /* ... */ }, }); // client client.sync("posts", { orgId: "org-42" }); ``` Works with any ORM or raw driver. ### Authentication and room-based access control `auth()` runs on every connection. Return an `AuthContext` — anything with a `userId`. It's passed to every `query`, `mutate`, and `authorize` call. ```ts server.auth(async (req) => { const token = req.headers.get("authorization")?.replace("Bearer ", ""); const session = await validateToken(token); if (!session) throw new Error("unauthorized"); return { userId: session.userId, orgId: session.orgId }; }); ``` For multi-tenant apps, use `room()` to pin a client to a subset of data: ```ts server.room("org/:orgId", async ({ params, auth }) => { if (!auth.memberships.includes(params.orgId)) { return { ok: false, reason: "not a member of this org" }; } // return nothing (or `{ ok: true }`) to allow the subscription }); ``` Room keys are resolved from the subscription's params and fail closed: params that address a pattern only partially, or that produce a key the pattern can't match, are rejected rather than falling back to an unscoped, cross-room subscription. Set `room` in `implement()` to require a specific pattern for a query. The whiteboard example wires this up with [better-auth](https://better-auth.com) — see [`examples/whiteboard/auth.ts`](https://github.com/TimMikeladze/reflectdb/blob/main/examples/whiteboard/auth.ts). ### Per-column merge for collaborative editing When two users edit different fields of the same row, `lww` would throw one write away. `merge` keeps both: ```ts const queries = defineSyncQueries({ docs: { row: t(), conflict: "merge" }, // per-column HLCs }); ``` ``` User A writes { title: "Hello" } at HLC 100 User B writes { body: "world" } at HLC 200 → Result: { title: "Hello", body: "world" } (both accepted) ``` ### Custom conflict resolvers For domain logic — counters, highest-bid-wins, append-only lists — supply a resolver: ```ts const queries = defineSyncQueries({ auctions: { row: t(), conflict: { policy: "custom", resolve: (incoming, existing) => { const bid = (incoming.payload.bid as number) ?? 0; if (bid <= (existing.row?.bid as number ?? 0)) { throw new Error("bid too low"); // rejects the op } return { row: { ...existing.row, ...incoming.payload } }; }, }, }, }); ``` ### Validating client payloads `t()` is a **compile-time** phantom — it erases at runtime. reflectdb validates protocol structure (an op's `payload` must be a non-array object or null) but never its contents, so a client can send `{ title: 12345 }` or extra keys and they reach your `mutate` untouched. `readonly` and `serverSet` strip named fields; they don't type-check what's left. Validate in `mutate`, with whatever library you already use — reflectdb has no opinion and no dependency here: ```ts import { z } from "zod"; import { MutationError } from "reflectdb/core"; const Todo = z.object({ id: z.string(), title: z.string().min(1).max(200), done: z.boolean(), }).strict(); // reject unknown keys instead of passing them through server.implement("todos", { query: (ctx, db) => db.select().from(todos), mutate: async (op, ctx, db) => { if (op.type === "delete") { await db.delete(todos).where(eq(todos.id, op.rowId)); return; } // Updates carry a partial delta, not a whole row. const schema = op.type === "insert" ? Todo : Todo.partial(); const parsed = schema.safeParse(op.payload); if (!parsed.success) { throw new MutationError("outside_shape", parsed.error.message); } await db.insert(todos).values({ id: op.rowId, ...parsed.data }) .onConflictDoUpdate({ target: todos.id, set: parsed.data }); }, }); ``` Two details that matter: - **Throw `MutationError`, not a plain `Error`.** Only `MutationError` carries an `ErrorReason` through to the client's `onError`; anything else is reported as `server_error`. - **A refused write rolls back to what that client was shown.** The reject's `serverRow` is the row from the client's own last query result (null if it never received one), not reflectdb's stored row. That keeps a query that hides columns per user from leaking them through a refusal. - **Write the parsed value, not `op.payload`.** Writing the raw payload after validating it defeats `.strict()` and any coercion the schema applied — and it also feeds unvalidated data into reflectdb's mirror, which is what conflict resolution compares against. The same applies to `authorize`, and to writes arriving through `server.rest()` — both run the identical pipeline. ### Ephemeral messages (cursors, presence, typing) Ephemeral events are room-scoped broadcasts that bypass the op log — ideal for high-frequency signals: ```tsx import { useEphemeral } from "reflectdb/react"; const { events, broadcast } = useEphemeral({ key: "cursor", userId: currentUserId, ttlMs: 10_000, }); // on mouse move broadcast({ x: e.clientX, y: e.clientY }); // render peers Object.values(events).map((c) => ); ``` Fan-out follows the sender's **query subscriptions**: recipients are the clients subscribed to the same queries, narrowed to the sender's room when one is resolved. A client that has called no `sync()` yet has no audience, so its ephemeral messages reach nobody. The `userId` on the wire is always the authenticated one — the client-supplied value is ignored — and a client-supplied `ttlMs` is clamped server-side. Subscribing to a room also delivers a **snapshot** of that room's live ephemeral state, so a client that joins mid-session sees the peers already there instead of waiting for each one to move again. Snapshots arrive as ordinary `ephemeral` events and exclude the joiner's own entries. By default this state lives in the server process, which is correct on one node and invisible across a fleet — two clients on different instances never see each other. Point `ephemeral.adapter` at shared infrastructure to fix both halves; see [Ephemeral (presence)](#ephemeral-presence). The whiteboard renders peer cursors this way — see [`examples/whiteboard/app.tsx`](https://github.com/TimMikeladze/reflectdb/blob/main/examples/whiteboard/app.tsx). ### Typed presence `presence()` is `useEphemeral` with the shape declared in the schema instead of at the call site. The channel key is derived from the entry name plus its serialized params, so two components watching the same presence entry always agree on the key. ```ts // schema.ts import { defineSyncQueries, presence, t } from "reflectdb/core"; export const queries = defineSyncQueries({ cursor: presence({ state: t<{ x: number; y: number; name: string }>(), params: t<{ gameId: string }>(), // part of the derived key ttlMs: 10_000, }), }); ``` ```tsx // app.tsx — usePresence comes from the typed factory, not the bare import import { createSyncReact } from "reflectdb/react"; import { queries } from "./schema"; export const { SyncProvider, useSync, usePresence } = createSyncReact(queries); function Cursors({ gameId }: { gameId: string }) { const { peers, set } = usePresence("cursor", { gameId }); // ^? { userId: string; state: { x, y, name } }[] useEffect(() => { const onMove = (e: PointerEvent) => set({ x: e.clientX, y: e.clientY, name: myName }); window.addEventListener("pointermove", onMove); return () => window.removeEventListener("pointermove", onMove); }, [set]); return peers.map((p) => ); } ``` Details worth knowing: - **No server registration.** Presence entries are not queries — there is no `server.implement`/`server.view` for them. They ride the same ephemeral channel and are room-scoped by the sender's active subscriptions. - **`peers` excludes you.** Ephemeral events are only delivered to *other* clients, so render your own cursor from local state. - **Peers are keyed by connection, not by account.** Presence entries are keyed by `clientId` end to end — on the wire, in the server's store, and in `peers` — so two tabs from one login are two peers with two cursors. Put the display identity in `state` (as `name` above) if you need it; the authenticated `userId` rides along on every event for authorization and display. - **Params are required when declared**, exactly like `useSync` — `usePresence("cursor")` fails to compile if the entry declares params. - **React only.** `createSyncSvelte` / `createSyncVanilla` have no presence helper; use `sync.sendEphemeral` / `sync.onEphemeral` (or the store's `ephemeral()`) with your own key there. `derivePresenceKey(name, params)` is exported from `reflectdb/react` if you want to interoperate with the same channel by hand. ### Peer-to-peer (WebRTC) Wrap the app in `PeerMeshProvider` and ephemeral frames go browser-to-browser over WebRTC data channels. Nothing else changes: `useEphemeral` and `usePresence` keep their API, and the reflectdb connection you already have carries the signaling. ```tsx import { SyncProvider, useEphemeral } from "reflectdb/react"; import { PeerMeshProvider, useCall, usePeers } from "reflectdb/react/webrtc"; ; // Cursors: unchanged code, now peer-to-peer at full rate. const { events, broadcast } = useEphemeral({ key: "cursor", userId, ttlMs: 5000 }); // Game ticks that should never touch the server: useEphemeral({ key: "tick", userId, transport: "p2p" }); ``` Each frame's route is set by `transport`: | `transport` | Peers connected directly | Everyone else | Server / joiner snapshot | | ------------------ | ------------------------- | ------------------ | ---------------------------------------------------------------------------- | | `"auto"` (default) | every frame, peer-to-peer | through the server | throttled to one frame per 100 ms per key, always ending on the latest value | | `"server"` | through the server | through the server | every frame | | `"p2p"` | every frame, peer-to-peer | nothing | nothing | `auto` throttles the server copy only while every peer the mesh knows of is connected directly. Until then (or if a connection fails) every frame also goes through the server, so attaching a mesh never costs a peer an update. Clients without a mesh — an htmx page, an older client — see the throttled rate once all the peers that have meshes are connected directly. A frame can arrive twice, once peer-to-peer and once through the server, and out of order on the unreliable channel. Receivers keep only the newest frame per peer and key. **Calls and screen share.** `useCall()` joins the room's call: ```tsx const { call, join, leave, shareScreen, stopScreen, setMuted } = useCall(); await join(); // camera + mic; or join(stream) with your own MediaStream await shareScreen(); // adds a "screen" stream alongside the camera call?.participants.map((p) => ); ``` Media flows only between peers that have both joined. `usePeers()` lists every peer with its state (`connecting`, `connected`, `server-only`) and whether it is in the call. How it works, and its limits: - **Discovery is presence.** Each mesh keeps a `__rtc` ephemeral entry alive. The peer with the lower id opens the connection. Signaling uses a `signal` message that goes to exactly one peer, is never stored, and is delivered only when sender and recipient share a room. That holds across a fleet too: the Redis and Postgres ephemeral adapters carry `signal` messages unchanged. - **Full mesh, capped.** `maxPeers` (default 8) sets how many direct connections a client keeps. Peers past the cap stay on the server path. Video calls hit uplink limits above about 6 participants; past that, use an SFU (LiveKit, Cloudflare Calls). - **Presence cost.** Each mesh re-announces every `announceIntervalMs` (default 30 s), and every announce fans out to the whole room. Raise the interval for very large rooms. New peers are found immediately whatever the interval. - **Failure.** A peer that does not connect within `connectTimeoutMs` (default 15 s) stays on the server path and is retried after a minute. - **NAT.** With no TURN configured, the mesh uses Google's public STUN server, which connects most peers. Users behind strict NATs need TURN; see below. - **Privacy.** Peers in a room learn each other's IP addresses through ICE. Set `relayOnly` (it requires TURN) to route everything through the relay and hide them. The mesh logs a warning if `relayOnly` is set with no TURN server, and `usePeers()` reports `relayed` per peer once connected. - **Trust.** A peer-to-peer frame bypasses the server's rate limit and validation. The `clientId` and `userId` on it come from the server-relayed signaling, not from the frame, but treat the frame's `data` as untrusted input. Each peer is capped at 300 frames and 50 control messages per second; the excess is dropped. **TURN, pluggable.** No TURN provider is built in or required. The server picks one with `server.iceServers(provider)`, and each browser gets short-lived credentials over its authenticated sync connection, so the long-lived secret never leaves the server and no credential sits at an open URL: ```ts import { cloudflareTurn, sharedSecretTurn, staticIceServers } from "reflectdb/server/turn"; // Cloudflare Realtime TURN: one API call per half-TTL, shared across clients. server.iceServers(cloudflareTurn({ keyId: env.CF_TURN_KEY_ID, apiToken: env.CF_TURN_API_TOKEN })); // Self-hosted coturn (use-auth-secret) or any TURN REST-API server: no network call. server.iceServers(sharedSecretTurn({ urls: ["turn:turn.example.com:3478"], secret: env.TURN_SECRET })); // Anything else (Twilio, Metered, your own service): any function of this shape. server.iceServers(async ({ auth }) => ({ iceServers: await myTurn(auth.userId), ttlSeconds: 3600 })); ``` On the client, `iceServers` is optional. Omit it and the mesh asks the server, falling back to public STUN if the server has no provider. Pass an array to use fixed servers, or a function to fetch credentials from anywhere. Credentials that expire are refreshed a minute early and pushed into live connections, so a long call outlives its first credential. - **Durable data never goes peer-to-peer.** Ops, permissions and ordering stay on the server. Use WebRTC only for the in-between states: the drag preview, not the drop. See it running in the [huddle demo](https://reflectdb-huddle.fly.dev/) ([source](https://github.com/TimMikeladze/reflectdb/blob/main/examples/huddle/)). Without React, call `createPeerMesh(client, config)` from `reflectdb/client/webrtc`. It returns `getSnapshot()` and `subscribe()` for any store, plus `call`, `shareScreen`, `setMuted`, `leave` and `destroy`. The design is in [`docs/webrtc.md`](https://reflectdb.dev/docs/webrtc.md). ### Per-user query results A `query` callback is just a function — it can return different rows depending on the caller's `auth`. reflectdb re-runs it whenever the listed `tables` change, so each subscriber gets a personalized view that stays live. The whiteboard uses this to keep the round's secret word out of the wire for everyone except the active drawer: ```ts const queries = defineSyncQueries({ roundWord: { row: t<{ id: string; gameId: string; word: string }>(), params: t<{ gameId: string }>(), tables: ["games", "game_secrets"], // re-run on these }, }); server.implement("roundWord", { query: async (ctx, db) => { const game = await db.select().from(games).where(eq(games.id, ctx.params.gameId)).get(); if (!game || game.state !== "drawing") return []; if (game.currentDrawerId !== ctx.auth.userId) return []; // guessers see [] const secret = await db.select().from(gameSecrets) .where(eq(gameSecrets.gameId, ctx.params.gameId)).get(); return secret?.word ? [{ id: ctx.params.gameId, gameId: ctx.params.gameId, word: secret.word }] : []; }, mutate: async () => { throw new Error("read-only"); }, tables: ["games", "game_secrets"], }); ``` The `game_secrets` table isn't registered in `defineSyncQueries`, so it's never broadcast directly. Calling `server.notifyChange("game_secrets")` from the engine fans out the recomputed `roundWord` result to whichever client is now the drawer. ### Read-only views The recipe above is a query that happens to reject writes. `view()` makes that the declaration: the entry has no `mutate`, `useSync(...)` returns only `{ rows, loading }`, and a write that reaches the server anyway is rejected with `readonly_query`. ```ts // schema.ts import { defineSyncQueries, view, t } from "reflectdb/core"; export const queries = defineSyncQueries({ leaderboard: view({ row: t<{ id: string; name: string; points: number }>(), params: t<{ gameId: string }>(), deps: ["games", "scores"], // re-run when either table changes }), }); ``` ```ts // server.ts — server.view, not server.implement server.view("leaderboard", (ctx, db) => db.select().from(scores) .where(eq(scores.gameId, ctx.params.gameId)) .orderBy(desc(scores.points)) .limit(10)); ``` ```tsx // app.tsx const { rows } = useSync("leaderboard", { params: { gameId } }); // rows: { id, name, points }[] — there is no .insert / .update / .remove here ``` Notes: - **`deps` drives change detection**, falling back to `tables` and then to the entry name. A view over tables it doesn't share a name with must declare them, or it never re-broadcasts. - **`implement()` and `view()` are not interchangeable.** Calling `server.implement` on a name declared as a view throws, and so does `server.view` on a name that isn't one. - **`server.view(name, fn)` takes no options** — only the callback and the schema's dependency list. There is no `authorize`, `room`, `groupBy`, `count`/`countHints` or `pk` on a view. Do access control inside the callback (it gets `ctx.auth` and `ctx.params`), and fall back to a regular `implement()` with a throwing `mutate` when you need those knobs. - **Rows need an `id`.** The primary key isn't configurable for views, so give each row a stable `id` — that's what delta diffing keys on. Computed rows can synthesize one. - **The type-level block is React-only.** `createSyncSvelte` / `createSyncVanilla` don't narrow view entries, so a write there compiles and is refused at runtime instead. ### Server-driven game loops Some apps need state that advances on a clock, not on user input — round timers, expiring claims, scheduled rotations. Pair `server.interval` with `notifyChange` and the server stays the single source of truth. ```ts server.interval(500, () => server.lock("tick", async () => { // a tick must never outrun itself const now = Date.now(); const active = await db.select().from(games).where(eq(games.mode, "pictionary")); for (const g of active) { if (g.state === "drawing" && now >= g.roundEndsAt) { await endRound(g.id); // raw SQL writes await server.notifyChange("games"); // fan-out to subscribers } } }), ); ``` `server.interval(ms, fn)` and `server.timeout(ms, fn)` wrap the globals with three differences worth having: a throw or a rejected promise inside `fn` is caught and logged instead of taking the process down, the handle is cleared by `server.close()`, and it is disposed on `bun --hot` reload — so an edit-save loop doesn't leave a fleet of orphaned timers ticking against the same rows. Both return `{ clear() }`. `server.lock(key, fn)` serializes async work per key: calls queue and run one at a time, and a failure in one doesn't poison the queue behind it. `server.tryLock(key, fn)` is the skip-if-busy variant — it returns `null` immediately when the key is held, which is usually what you want for a tick that would otherwise pile up. ```ts const result = await server.tryLock(`game:${gameId}`, () => scoreRound(gameId)); if (result === null) return; // another call is already scoring this game ``` Both are in-process only. Across instances, keep the guard in the database (a conditional `UPDATE … WHERE state = 'drawing'` that returns rows-affected) — the lock protects a single Node/Bun process, not a cluster. The whiteboard example uses both — see [`examples/whiteboard/server.tsx`](https://github.com/TimMikeladze/reflectdb/blob/main/examples/whiteboard/server.tsx). ### Transactional writes with `server.tx` `notifyChange` per table gets tedious the moment one logical action touches three of them. `server.tx` runs the work, tracks which tables it wrote, and fires one `notifyChange` per touched table — only if the whole function succeeded. ```ts await server.tx(async (tx) => { await tx.update(games).set({ state: "scoring" }).where(eq(games.id, gameId)); await tx.insert(scores).values(rows); await tx.delete(guesses).where(eq(guesses.gameId, gameId)); }); // → games, scores and guesses each broadcast once, after COMMIT ``` - **Atomic by default.** `atomic: true` is the default: the body runs inside `BEGIN`/`COMMIT` and rolls back on throw. It resolves an adapter from `ServerConfig.txAtomic`, falling back to a bundled Drizzle adapter (lazy-loaded, so there's no top-level `drizzle-orm` dependency). With neither available it throws — pass `atomic: false` for a non-transactional group, or supply your own adapter with `server.tx({ atomic: myAdapter }, fn)`. - **Table tracking is automatic for Drizzle only.** The proxy watches `insert` / `update` / `delete` (`select` is not a write, so it doesn't count). On Kysely, Prisma or raw SQL, call `tx.touch("games")` after each write. - **Notifies never fire on a throw**, transactional or not. - **Pooled connections need care.** `BEGIN`/`COMMIT` and the writes must share one connection, so pass a single-connection handle when atomicity is load-bearing rather than a pool. For a single row there is `server.emit(table, payload)`. It generates a rowId, stamps an HLC, writes reflectdb's mirror plus the op-log entry, and broadcasts: ```ts const { rowId, hlc } = await server.emit("todos", { title: "filed by a cron", done: false }); await server.emit("todos", { done: true }, { rowId, type: "update" }); ``` It does **not** call your `implement`'s `mutate`, so it does not write your database. That makes it the right tool when reflectdb's own store is what your `query` reads, and the wrong one when your database is — a broadcast re-runs the query, so a row your database never received simply won't appear. When your write has to happen under the same stamp, use the primitive `emit` and `server.rest()` are both built on: ```ts await server.applyServerOp( { type: "insert", table: "todos", rowId, payload }, async (stamped) => { // runs before the mirror write await db.insert(todos).values({ id: stamped.rowId, ...stamped.payload }); }, { roomKey: `org/${orgId}` }, // keep the fanout inside the tenant ); ``` A throw inside `execute` aborts before anything touches the mirror or the op log. Omit `roomKey` and the broadcast reaches every subscriber of the affected query, across rooms. ### Windowed sync and pagination For large tables, sync a sliding window instead of the whole set: ```ts const queries = defineSyncQueries({ messages: { row: t(), conflict: "lww", countHints: true, // emit count_changed deltas }, }); // React const { rows } = useSync("messages", { window: 50 }); const total = useTotalCount("messages"); const loadMore = useLoadMore("messages"); // show "Load 50 more" when rows.length < total ``` Make the window a **real** limit by reading `ctx.limit` in the query and supplying a `count`: ```ts server.implement("messages", { query: ({ params, limit }, db) => db.select().from(messages) .where(eq(messages.roomId, params.roomId)) .orderBy(desc(messages.createdAt)) .limit(limit ?? 1000), count: async ({ params }, db) => (await db.select({ n: count() }).from(messages) .where(eq(messages.roomId, params.roomId)))[0].n, }); ``` Without them, the server fetches every matching row on every broadcast and slices in JS — pagination reduces bytes on the wire and nothing else. `ctx.limit` is `undefined` when the caller genuinely needs the full set, so a plain `limit ?? ` is safe. A window is an **entitlement**, not a row count: a subscriber with `window: 50` whose query matched 3 rows still receives the next 47 inserts, and `loadMore(20)` widens the entitlement by 20 regardless of how many rows actually arrived. Reconnecting restores the widened window, not the initial one. ### Auto-generated REST API `server.rest()` returns a fetch-style handler that responds to CRUD URLs derived from your schema: ```ts const rest = server.rest({ prefix: "/api" }); serve({ port: 3001, async fetch(req, srv) { const url = new URL(req.url); if (url.pathname.startsWith("/api/")) return rest(req); // ...WebSocket upgrade, etc. return new Response("ok"); }, }); ``` Endpoints generated for every `implement()`'d table: ``` GET /api/ → list (supports ?where=…&limit=&offset=) GET /api/
/:id → single row POST /api/
→ insert (body = row, or array = batch) PATCH /api/
/:id → update DELETE /api/
/:id → delete ``` REST writes go through the same pipeline as sync writes and broadcast deltas to connected clients. ### High availability with Postgres Share a Postgres op log between server instances. Clients reconnecting to a different instance resume seamlessly from their HLC watermark. ```ts import pg from "pg"; import { createPostgresStorage } from "reflectdb/server"; const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL }); const server = createSyncServer({ queries, db, transport, storage: createPostgresStorage(pool), serverId: process.env.FLY_ALLOC_ID, poll: 500, // 500ms cross-instance poll for active-active }); ``` | Mode | Config | Use case | |------|--------|----------| | **Failover only** | Shared Postgres, no `poll` | Clients resume on reconnect | | **Active-active** | Shared Postgres + `poll: 500` | Real-time cross-instance updates | Each poll tick first probes the shared op log's head HLC; an idle tick costs one `MAX(hlc)` query and broadcasts nothing. Only tables that actually changed are re-broadcast. The tick also re-merges the shared clock watermark, so an instance whose wall clock lags its peers stops stamping writes below HLCs clients have already seen. ### Sync with no database at all `createObjectStorage` runs a room with an S3-compatible bucket as the only durable store — no Postgres, no SQLite, no volume. Works with AWS S3, Cloudflare R2, Tigris, MinIO and GCS. ```ts import { createObjectStorage } from "reflectdb/server/storage/object"; const storage = createObjectStorage({ store: { provider: "tigris", // or "aws" | "r2" | "minio" | "gcs" bucket: process.env.S3_BUCKET!, credentials: { keyId: process.env.S3_ACCESS_KEY_ID!, secret: process.env.S3_SECRET_ACCESS_KEY!, }, }, roomId: "board-42", }); const server = createSyncServer({ queries, db, transport, storage }); // Flush and release the lease so the next machine takes over immediately. process.on("SIGTERM", () => storage.close().then(() => process.exit(0))); ``` Row state is authoritative **in memory** and the bucket is durability only, so reads never touch the network. A write appends to a buffer that group-commits one object per batch, then advances a compare-and-swapped manifest — the log's single linearization point. The flush loop is self-clocking, so there is no flush interval to tune, and an idle room issues no requests at all. The trade is that a room has exactly one writer, elected with a lease. Route each room to one instance: this adapter replaces shared-database HA polling with room affinity rather than layering on top of it. Where you cannot promise that routing, see [Serverless sync on Vercel](#serverless-sync-on-vercel). Every knob and its default is in [Object storage (no database)](#object-storage-no-database); the design, the failure modes and the known limits are in [`docs/object-storage.md`](https://reflectdb.dev/docs/object-storage.md). ### Serverless sync on Vercel Serverless functions break two assumptions a long-lived server gets for free. Both have a flag, and [`examples/kanban`](https://github.com/TimMikeladze/reflectdb/blob/main/examples/kanban/) is a deployed board that uses them. **Any request can land on any instance,** so there is no single writer to elect. Drop the lease and let instances race on the manifest CAS instead: ```ts const storage = createObjectStorage({ store: { /* … */ }, roomId, concurrency: "optimistic", // no lease; the loser of a CAS re-reads and retries }); ``` That rests on the same guarantee the lease mode does — the CAS is what keeps the data correct, and the lease was only ever an optimization. What it costs is that in-memory state is no longer authoritative, because another invocation may have committed since this one last looked. Call `await storage.refresh()` — one manifest GET, `true` when something moved — before a read that must be current. **A function cannot hold a WebSocket, and SSE is one-way.** The POST and the event stream are two separate invocations, so a reply to a POST can never reach a stream that process does not own. `serverless: true` returns those replies in the POST's own response instead: ```ts const transport = createSseServerTransport({ serverless: true }); const handler = new MessageHandler({ transport, serverId, db, allowAnonymous: true }); handler.setStorage(storage); // POST /api/sync/messages — the replies this message produced const messages = await transport.collectReplies(clientId, message, () => handler.whenIdle(clientId), ); return Response.json({ messages }); // GET /api/sync/events — the stream carries only OTHER clients' changes setInterval(async () => { if (await storage.refresh()) await handler.pollRemoteChanges(); }, 250); ``` `MessageHandler` is exported from `reflectdb/server`; the example drives it directly rather than through `createSyncServer`, because a serverless route needs `whenIdle` and `pollRemoteChanges`. Set the matching `serverless: true` on `createSseClientTransport`. Two more things a serverless deployment owns, both worked through in the kanban example: the client's session and subscriptions are rebuilt per invocation, because the `hello` and `sync_declare` went to a different process, and the stream instance must `bootstrap` so its result cache holds what the client is actually holding — the broadcast engine sends a diff against that cache, and an empty one makes every existing row look new. ### htmx 4 bindings `reflectdb/htmx` lets htmx drive the DOM while reflectdb owns the data. Bindings point at a `reflect:` action instead of a server route, so reads and writes resolve against the local store — optimistic, offline-capable, and re-rendered whenever a peer's change arrives. htmx 4 core ships no SSE or WebSocket support of its own; sync stays reflectdb's job. ```ts import htmx from "htmx.org"; import { createHtmxSync } from "reflectdb/htmx"; const reflect = createHtmxSync({ htmx, url: "ws://localhost:3001/sync", token, tables: ["todos"], }); reflect.view("todos", ({ rows }) => rows .map( (todo) => `
  • ${escapeHtml(todo.text)}
    • `, ) .join(""), ); // Form bodies arrive as strings — coerce before they reach the op log. reflect.parse("todos", (payload) => ({ ...payload, done: payload.done === "true", })); await reflect.connect(); ``` ```html
      ``` The action grammar is REST-shaped, so the attributes read like ordinary htmx: ``` GET reflect:
      → render the collection view GET reflect:
      /:id → render one row POST reflect:
      → insert (row id from the body's `id`, else generated) a body `id` names the row; it is not stored as a column PUT reflect:
      /:id → update PATCH reflect:
      /:id → update DELETE reflect:
      /:id → delete ``` Writes answer `204 No Content`, so htmx swaps nothing where the write happened. The store change then re-renders every bound element a beat later — one render path whether the edit came from this tab, another tab, or the server. Worth knowing: - An element binds by making its first `reflect:` read, so give collection bindings `hx-trigger="load"`. - Use an inner swap (`innerHTML`, `innerMorph`) on collection bindings. `outerHTML` replaces the bound element itself, and a replacement still carrying `hx-trigger="load"` would re-request forever. - `hx-swap="innerMorph"` is usually what you want: htmx 4 morphs in place, so focus and caret position survive a re-render. - Query params reach the view for filtering (`reflect:todos?done=false`); they do not change the server subscription. Declare that with `tables`, or with `reflect.sync.sync(table, { params })`. - A row read whose row is missing answers `204`, leaving existing markup alone instead of blanking it. - A view returns a raw HTML string — escape interpolated values yourself. - A bad action or an unregistered view answers 4xx/5xx, which htmx swaps nowhere by default, and a `reflect:` request never reaches the network tab. The adapter `console.error`s every one of them so the failure is not silent. - Checkboxes, radios and `
      -.parquet materialized rows (phase 3, DuckDB) ``` `` is the fencing epoch under `"single-writer"` and the instance's `writerId` under `"optimistic"`. The distinction is load-bearing, not cosmetic: without a lease every instance reads the same epoch, so an epoch-named key is the same key on every instance. Two of them then write one object, and the commit that supersedes it lists the key it is still pointing at as garbage — an hour later GC deletes the live snapshot and the room never boots again. Only `_lease` and `_manifest` are ever overwritten, and both only via CAS. Everything else is write-once, which is what makes concurrent readers safe. ## Writer election and fencing ```ts // acquire const cur = await driver.get("_lease"); // { owner, epoch, expiresAt } if (cur && cur.body.expiresAt > now) return NOT_LEADER; await driver.put("_lease", { owner: me, epoch: cur.body.epoch + 1, expiresAt: now + ttl }, { ifMatch: cur.etag }); // loser gets 412 ``` Every WAL segment name and every manifest write carries `epoch`. **The manifest CAS is the real guard, not lease expiry.** A zombie writer that lost its lease still holds a stale manifest etag, so its manifest CAS fails with 412. It can only write orphan WAL segments, which readers skip by epoch. This sidesteps the classic "paused process still holds a valid lease" hole — the lease is an optimization to avoid wasted work, not the safety mechanism. The writer must **self-fence**: a failed renewal stops write acceptance immediately. ### Lease TTL is long on purpose TTL only bounds *unclean* failover, because [`close()`](#shutdown) releases the lease explicitly. Long TTL is what makes idle rooms cheap: | TTL / renew | Unclean failover | PUTs/day/room | Cost/month/room | |---|---|---|---| | 30s / 10s | 30s | 8,640 | ~$1.30 | | 300s / 120s | 5min | 720 | ~$0.11 | Default is **300s / 120s**, with `lease.mode: "on-write"` so a room holding connected clients but taking no writes spends nothing at all. ## Concurrency modes The lease above assumes the deployment can route a room to one instance. Some cannot — Vercel functions, Lambda, anything where any request lands on any instance — so the adapter takes a `concurrency` mode. | Mode | Lease | In-memory state | Route a room to one instance | |---|---|---|---| | `"single-writer"` (default) | elected, renewed, fenced | authoritative | required | | `"optimistic"` | none | may be stale | not required | `"optimistic"` rests on exactly the same guarantee `"single-writer"` does: **the manifest CAS is what keeps the data correct, and the lease was only ever an optimization** to stop two servers doing redundant work. Concurrent writers race on the CAS, the loser re-reads and retries, and nobody is fenced. What it costs is that in-memory state is no longer authoritative, because another instance may have committed since this one last looked. Hence: ```ts await storage.refresh(); // one GET; true when the room actually moved ``` `refresh()` re-reads the manifest and applies any segments — or another instance's snapshot — that landed since. It does nothing further when `commitSeq` has not changed, which is what makes a serverless poll loop affordable: the steady state is a single small read, and queries re-run only when it returns true. Under `"single-writer"` it resolves `false` without a request, because this instance *is* the writer and there is nothing to catch up on. ### `commitSeq` closes an ABA hole The manifest carries a counter that increments on every commit, and it exists solely to make the manifest bytes differ on every write. S3 derives an etag from object content, so writing identical bytes leaves the etag unchanged. Without the counter a writer could read etag E, have another writer commit a manifest that happens to serialize identically, and still win its `ifMatch: E` — a lost update no 412 ever reports. In practice `oplogHead` and `walSegs` almost always differ, but "almost always" is not a property to rest a linearization point on. `lastWriter` rides along for a related reason. Together with `commitSeq` it identifies a commit uniquely, which is what lets a writer that took a 412 tell "my own write, acknowledged late" from "someone else got there first". `epoch` cannot do that job under `"optimistic"`: with no lease every instance shares the manifest's epoch, and two of them will attempt the same `commitSeq` — so an epoch-based check would adopt a rival's commit as its own and silently drop the segment it just wrote. ### What the platform still owns `"optimistic"` removes the routing requirement, not the two things a serverless runtime does differently. Both are worked through in [`examples/kanban`](https://github.com/TimMikeladze/reflectdb/blob/main/examples/kanban/): - **The session is rebuilt per invocation.** `hello` and `sync_declare` went to a different process, so an instance holding the event stream has no subscriber to push to until it replays them — and must `bootstrap` too, because the broadcast engine emits a diff against its result cache, and an empty cache makes every existing row look new. - **A reply cannot be streamed by a process that does not own the stream.** The SSE transport's `serverless: true` returns each POST's replies in that POST's own response; the stream is left pushing other clients' changes. ## Group commit No `flushMs` knob. The flush loop is **self-clocking**, the same trick Postgres group commit and Kafka's linger use: ```ts while (running) { await buffer.waitNonEmpty(); // idle blocks here: no timer, no PUT if (minLingerMs) await delay(minLingerMs); // coalesce same-tick ops const batch = buffer.drain(maxBytes); await driver.put(`wal/${epoch}-${seq++}.jsonl`, encode(batch)); await casManifest({ walSegs: [...prev, seg], oplogHead: batch.maxHlc }); emitDurable(batch.maxHlc); } ``` Exactly one flush is in flight at a time. That single constraint makes it adapt with no configuration: - **Low write rate** — buffer holds one op, flush fires at once. Latency is one store round trip. Optimal. - **High write rate** — ops accumulate *during* the in-flight flush, so the next batch is naturally as large as the round trip allows. Throughput is `maxBytes / rtt`. Optimal. - **Idle** — the loop blocks. Zero PUTs, zero timers. Batch size auto-tracks store latency, so a slow provider batches harder and a fast one (S3 Express, Tigris colocated with Fly) batches less. Exposing a `flushMs` would invite someone to set 10ms and quadruple their bill for no latency gain, so it is deliberately not a knob. `minLingerMs` defaults to 5ms purely to coalesce ops arriving in the same event-loop tick. Pure win, no latency cost that matters. ## Durability model ### The failure this avoids Naively acking a write as soon as it hits memory is unsafe: ``` client writes op → server applies to memory → server acks client retires op from its pending queue server crashes before flush client reconnects → resume ships server snapshot → the row reverts ``` The client's optimistic write silently disappears. Op idempotency does not save you, because there is nothing left to replay. ### Resolution Two modes, and **`"durable"` is the default**: | Mode | Ack when | Loss window | Correct today | |---|---|---|---| | `"durable"` (default) | after manifest CAS | none | yes | | `"buffered"` | after memory apply | one flush interval | **no** — see below | `"durable"` costs one store round trip per write batch (~50-150ms on S3, ~10ms on S3 Express or a colocated Tigris bucket). Correct with **zero protocol change**, which is why it is the default. Shipping a fast-but-lossy default would be trading correctness for a benchmark. `"buffered"` is only safe once the client retires pending ops on a **durable watermark** rather than on the apply ack: - `applied` ack — immediate, confirms ordering and conflict outcome, unblocks UI - `durableHlc` — broadcast after manifest CAS; client retires ops `<= durableHlc` That is a small protocol change (HLC watermarks are already everywhere in the codebase) but it is a protocol change, so it lands in phase 2. The adapter already emits `onDurable(hlc)` today so phase 2 is purely additive. Until then `"buffered"` is opt-in and documented as lossy. ## Replay protection must be durable too `reserveOp` is atomic for free — one writer per room, no `await` inside — but atomicity is not the whole requirement. It also has to survive a restart. `src/server/op-processor.ts` uses the reservation result to gate the entire pipeline for an op, **including the query's `mutate` callback**. That callback is arbitrary user code: incrementing a counter, charging something, sending mail. So the tempting argument — "a duplicate is harmless, conflict resolution discards it because its HLC is not newer" — covers only the row write, never the side effect. Losing the reservation window on every restart would re-run those callbacks for any client that reconnects and resends, which on a platform that replaces machines constantly is the common case, not a rare one. Reservations are therefore written to the WAL as their own record kind: ```ts { k: "reserve", opIds: [...], at: } ``` Appended but **not awaited**. `reserveOp` runs before the op it guards, and that op's own `applyOp` awaits the very batch the record lands in — so under `durability: "durable"` the reservation is durable by the time the write it protects is acknowledged, at zero extra round trips. Two details that matter: - A `reserve` record carries no HLC, so it must not advance `oplogHead`. Doing so would tell a resuming client the room changed when nothing did. - Replay restores the record's **original** timestamp, not `now()`. Re-stamping on every boot would keep a long-dead op id alive forever in a room that restarts often. ## Backpressure and health An undefined backpressure policy means "OOM during a store outage". Explicit: ``` buffered bytes > batch.maxBufferBytes → "reject" (default) writes throw ObjectStorageBackpressureError; the client retries. Backpressure propagates to the source, which is correct. "degrade" keep accepting, stop promising durability, flip health to "degraded" ``` Related: during a store outage under `"durable"`, every client's pending queue grows without bound. The adapter surfaces health so the app can show "reconnecting / not saved" rather than lying with a happy UI: ```ts storage.health // "healthy" | "degraded" | "unavailable" storage.onHealthChange(cb) ``` The difference between "degraded" and "corrupted" is designed here, not discovered in an incident. ## Memory budget State is authoritative in memory, so the ceiling is a cliff, not a slope. The budget is **global across rooms**, because a per-room cap lets one whale room starve five hundred small ones. ``` memory.maxTotalBytes global budget across all rooms in the process memory.maxRoomBytes per-room cap memory.onExceeded "reject" (phase 1) | "evict" | "spill" (phase 3) memory.idleEvictMs zero-client rooms flush, release lease, drop state ``` Phase 1 implements accounting plus `"reject"`, so the cliff surfaces as a typed error instead of an OOM. `"spill"` (fall back to DuckDB-over-Parquet reads) is the escape hatch that stops in-memory from being a hard product limit, and it is the strongest reason DuckDB is in this design at all — but it needs the Parquet layer, so it waits for phase 3. ## Shutdown Fly replaces machines constantly. A `SIGTERM` without a flush drops the buffer, which under `"buffered"` is silent loss on **every deploy**. ```ts async close() { stopAcceptingWrites(); await flush(); // bounded by shutdownFlushMs, default 5s await stopFlushLoop(); // bounded await releaseLease(); // bounded; failover becomes ~0, not leaseTtl } ``` **Every step is bounded, not just the drain.** An in-flight request cannot be cancelled — `fetch` is on the wire and the store decides when it answers — so an unbounded wait anywhere here lets a hung store hold the process open through a SIGTERM until the platform's kill timer fires. That is strictly worse than abandoning a loop that is already unable to make progress: nothing is left half-written, because a segment PUT is a single write and the manifest only advances after it lands. Lease release on clean shutdown is what lets the TTL be long, which is what makes idle rooms cost nothing. The three decisions are load-bearing on each other. ## Compaction and GC Activity-gated flushing has one sharp edge: under sustained load it produces many tiny segments. An hour of active drawing at one flush per 200ms is ~18,000 segments, and boot then needs 18,000 GETs. Boot time degrades linearly with write volume. So activity-gated flush **requires** a compaction companion, triggered on segment count rather than on a clock: ``` when walSegs.length >= compaction.afterSegments (200) or walBytes >= compaction.afterBytes (64MB): write snap/- full materialized row state CAS manifest { snapHlc, walSegs: [] } schedule delayed GC of the superseded segments (never of a key the manifest still names) ``` Still activity-gated: segment count only grows from writes, so idle stays free. **GC must be delayed.** A reader that fetched the old manifest may be mid-GET of a segment the new manifest dropped. Deleting immediately gives it a 404. `compaction.gcGraceMs` defaults to 1 hour. ### Compaction must advance the resume cutoff The sharpest edge in the design, because it fails silently. A snapshot carries rows and reservations but **no ops**, and the compaction CAS clears `walSegs`. So after the next restart the op ring holds only the segments written since that compaction — everything older is gone. `handleResume` in `src/server/handler.ts` rejects a client whose watermark predates `getMeta("compactionCutoff")`. With no cutoff recorded it takes the other branch instead: it asks `getChangedTablesSince`, gets `[]` from the truncated ring, and tells the client nothing changed. The client then sits on arbitrarily stale rows forever, and nothing raises an error anywhere. `CompactionManager` states the contract for the SQL adapters — *commit the cutoff first, then lazily delete ops*. This adapter's own compaction is bound by the same rule: whatever makes ops unavailable owns advancing the cutoff. It is written inside the same CAS that clears `walSegs`, so there is no window where the ops are gone but the cutoff still says otherwise, and it only ever moves forward — a cutoff that went backwards would start admitting resumes it had already correctly rejected. ## Provider compatibility DuckDB compatibility is uniform — `TYPE s3` with `ENDPOINT` and `URL_STYLE` covers every S3-compatible store. **Conditional-write support is where providers actually differ**, and that is what this design depends on: | Provider | `If-None-Match: *` | `If-Match: ` | Notes | |---|---|---|---| | AWS S3 | yes (Aug 2024) | yes (Nov 2024) | S3 Express One Zone also, ~10ms | | Cloudflare R2 | yes | yes | zero egress — best for phase 4 client bootstrap | | Tigris | yes | yes | Fly-native, colocated, no egress | | GCS | yes | yes | native uses `x-goog-if-generation-match`; verify via S3 interop | | MinIO | **no wildcard** | yes | exact etag only; needs `init()`, see below | | Backblaze B2 | no | no | not viable as primary store | ### The MinIO gotcha MinIO shipped conditional writes in February 2023, before AWS, but never accepted the `*` wildcard — it requires an exact etag. `create-if-absent` therefore does not work. Resolution: the driver declares `caps.casWildcard`. When false, the store requires a one-time **`init()`** that unconditionally writes an unowned `_lease` and an empty `_manifest`. Every subsequent write is `If-Match: `, which MinIO supports fine. `init()` is a **deploy step, not something N racing servers run**. Concurrent first-init on a non-wildcard driver is unsafe and the adapter says so. `caps.casWildcard` is a capability the driver reports, not a user setting. ## The driver interface Four methods. Everything above sits on top and never learns which provider it is talking to. ```ts interface ObjectDriver { get(key: string): Promise<{ body: Uint8Array; etag: string } | null>; put(key: string, body: Uint8Array, opts?: { ifMatch?: string; ifNoneMatch?: "*" }): Promise; // → etag, throws PreconditionFailed list(prefix: string): Promise<{ key: string; size: number }[]>; delete(keys: string[]): Promise; readonly caps: { casWildcard: boolean }; } ``` Three implementations: - **memory** — with fault injection (412 storms, 500s, latency). Ships in phase 1, not after: CAS races are the hard part of this design and they do not reproduce naturally. - **filesystem** — same semantics via atomic rename; makes the whole suite runnable with no network. - **s3** — SigV4 over `fetch`. Must use WebCrypto (`crypto.subtle`), never `node:crypto`: `bunup.config.ts` builds `src/` with `target: "browser"` and a `node:` import anywhere in `src/` breaks every consumer's bundle. See `src/server/node-require.ts` for the one sanctioned escape hatch. ### Two clients, two roles DuckDB's `httpfs` **cannot do conditional writes** — it reads and bulk-writes files, and exposes no `If-Match` surface. It therefore cannot be the CAS primitive. | Plane | Client | Job | |---|---|---| | Control | raw S3 REST (this driver) | lease CAS, manifest CAS, WAL PUT, GC | | Data | DuckDB `httpfs` | Parquet snapshot read/write, compaction, boot, time travel | One user-facing config maps to both. DuckDB stays fully optional, and phase 1 ships without it. DuckDB config, when it lands, uses one code path for every provider: ```sql CREATE OR REPLACE SECRET reflect ( TYPE s3, PROVIDER config, KEY_ID '...', SECRET '...', REGION 'auto', ENDPOINT 't3.storage.dev', URL_STYLE 'vhost', SCOPE 's3://my-bucket/rooms/' -- scope it; an unscoped secret applies to every s3:// path ); ``` Skip DuckDB's dedicated `TYPE r2` / `TYPE gcs` secrets. They are convenience wrappers (`r2` derives the endpoint from `ACCOUNT_ID`; `gcs` hardcodes an endpoint and accepts HMAC keys only) that would fork the config into three shapes for no gain. ## Configuration ```ts interface ObjectStorageConfig { // ── must expose: no sane default exists ────────────────────── store?: { provider?: "aws" | "r2" | "tigris" | "minio" | "gcs"; // preset fills endpoint/urlStyle bucket: string; prefix?: string; endpoint?: string; region?: string; // "auto" urlStyle?: "vhost" | "path"; credentials: { keyId: string; secret: string; sessionToken?: string } | "credential_chain"; accountId?: string; // R2: derives the endpoint }; driver?: ObjectDriver; // instead of `store`: filesystem, memory, your own roomId: string; writerId?: string; // random; set it to make lease ownership legible durability?: "durable" | "buffered"; // "durable" — see Durability model retentionMs?: number; // Infinity concurrency?: "single-writer" | "optimistic"; // see Concurrency modes // ── should expose: workload-dependent ──────────────────────── batch?: { maxBytes?: number; // 4 MiB minLingerMs?: number; // 5 maxBufferBytes?: number; // 64 MiB onBackpressure?: "reject" | "degrade"; // "reject" }; compaction?: { afterSegments?: number; // 200 afterBytes?: number; // 64 MiB gcGraceMs?: number; // 3_600_000 }; lease?: { ttlMs?: number; // 300_000 renewMs?: number; // 120_000 mode?: "always" | "on-write"; // "on-write" — zero PUTs when idle }; memory?: { maxTotalBytes?: number; maxRoomBytes?: number; onExceeded?: "reject" | "evict" | "spill"; // "reject" idleEvictMs?: number; // 300_000 }; shutdownFlushMs?: number; // 5000 duckdb?: { enabled?: boolean }; // false in phase 1 onDurable?: (hlc: string) => void; // per batch; phase 2 broadcasts it onHealthChange?: (health: StorageHealth) => void; } ``` ### Knobs deliberately not exposed - **`flushMs`** — self-clocking group commit removes the need, and exposing it invites a 10ms setting that quadruples cost for no latency gain. - **DuckDB internals** (threads, `memory_limit`, extension paths) — derive from `memory.maxTotalBytes`. - **Per-provider settings beyond the preset** — `caps.casWildcard` is detected, not configured. - **Separate read/write endpoints, custom retry curves** — until someone asks. Roughly six knobs anyone touches; the rest exist for the one person who needs them. Phase 1 with only `store` and `roomId` set is the intended starting surface. ## Known limits - **No cross-room transactions.** `src/server/tx-atomic.ts` scope is room-local. - **Read fanout caps at one machine.** Single writer per room means one machine's connection limit. A ten-thousand-viewer room needs read-only followers tailing the manifest (lag = flush + poll). Deferred. - **Reader replicas are eventually consistent** — lag is flush plus poll. Does not matter while real-time fanout is WS from the writer. - **Hard delete versus immutable segments.** Object storage makes infinite retention nearly free, which makes GDPR-style hard delete expensive: `retentionMs` needs a compaction pass that actually drops tombstoned rows rather than marking them. Deferred to the Parquet phase. - **Boot thundering herd.** Five hundred rooms waking at once is a GET storm; needs a concurrent-boot cap once one process hosts many rooms. - **A fenced writer keeps serving reads.** Once the manifest CAS is lost, buffered writes are rejected — but the mutations they carried are already in memory (`submit` mutates before appending), so the room goes on answering `getRow` / `getRows` from a view that diverges from the store and from the new writer's. `health` flips to `"unavailable"`, and nothing currently acts on it. Whether reads should fail closed once fenced is an open decision, not an oversight; failing closed trades a stale read for an outage, and which is worse depends on the app. - **`batch.maxBytes` counts UTF-16 units, not bytes.** `JSON.stringify(...).length` under-counts non-Latin text by up to 3x, so a 4 MiB cap can emit a ~12 MiB object. Internally consistent — the same estimate is added and subtracted, so nothing desyncs — but the cap does not mean quite what it says. - **`pendingGc.deletableAt` is stamped with the compacting writer's clock** and compared against a possibly different writer's clock after a failover. Immaterial at the 1 hour default; it would matter if `gcGraceMs` were lowered to seconds. ## Deferred work Recorded so a later phase cannot silently default the wrong way. ### Direct-from-bucket client bootstrap breaks the ACL model Phase 4 (client GETs a snapshot straight from the bucket via presigned URL) collides with `src/server/enforcement.ts` and per-user query results. **A snapshot object has one ACL.** It cannot be filtered per user. Handing a client a presigned URL gives them every row in that table regardless of what their query would have returned. Direct bootstrap is therefore legal **only when room membership is the entire access check** for that table. It must be opt-in and explicit: ```ts query("strokes", { bootstrap: "direct" }) // room ACL is the whole ACL query("dms", { bootstrap: "server" }) // default ``` Default `"server"`. A silent default to `"direct"` is a data breach, not a performance regression. ### Row encoding Rows are `Record` plus a `colClocks` map, both dynamic. Two options for the Parquet phase: - **JSON blob columns** — `tbl, row_id, data, col_clocks, hlc`, mirroring the SQLite table. Trivial, no schema registry. **Chosen for phase 1.** - **Typed columns** derived from `src/core/schema.ts` — columnar, unlocks analytics and time-travel queries. Opt-in per table when the Parquet layer lands. Do not try to auto-infer. ### Cold op-log tier The op log is append-only and HLC-ordered, which is exactly Parquet's shape. Once `snap/` is Parquet, `getOpsSince` for a long-offline client can union hot in-memory ops with cold Parquet segments in one DuckDB query, and DuckDB can `ATTACH` SQLite and Postgres directly so the same trick works for the existing adapters. ## Phases 1. **Adapter on object storage.** *Shipped.* Driver interface; memory + filesystem + S3 drivers; lease, manifest, group commit, WAL replay, JSON snapshots; backpressure, health, memory accounting, shutdown flush; plus `concurrency: "optimistic"` and `refresh()` for platforms that cannot route a room to one instance. No DuckDB, no Parquet. Passes `test/server/storage/storage-adapter-suite.ts`. 2. **Durable watermark protocol.** `durableHlc` broadcast, client retires pending ops against it. Unlocks a safe `"buffered"`. Room-affinity routing; retire the shared-storage HA polling path for this adapter. 3. **DuckDB data plane.** Parquet snapshots, compaction as `COPY ... TO 's3://...'`, boot-from-Parquet, `memory.onExceeded: "spill"`, cold op-log tier. 4. **Direct-from-bucket bootstrap.** Presigned snapshot plus server delta, gated on the per-table `bootstrap` opt-in above. --- Source: https://reflectdb.dev/docs/webrtc.md # WebRTC for ephemeral traffic Status: built — phases 1–3 and the bindings. A landing-page demo is not built yet. ## What and why High-frequency ephemeral traffic (cursors, drags, game ticks) and media (calls, screen share) go peer-to-peer over WebRTC. reflectdb's own connection does the signaling, so apps need no extra infrastructure beyond an optional TURN server. Durable data never goes P2P. Ops, permissions, and ordering stay on the server. WebRTC changes only the route an ephemeral frame takes; the server path remains the source of truth and the fallback. Use cases, in build order: | # | Use case | Rides on | | --- | ---------------------------------- | ------------------------------------------- | | 1 | Live cursors / selections | data channel, unreliable | | 3 | Drag previews, in-progress strokes | data channel, unreliable; commit via ops | | 4 | Multiplayer game state | data channel, unreliable, P2P-only option | | 5a | Follow mode (viewport) | data channel — same as #1, a `viewport` key | | 2 | Voice/video huddle | media tracks | | 5b | Screen share | media track (`getDisplayMedia`) | ## Today's gap `ephemeral` messages are broadcast to every room subscriber and written to the presence store. Signaling needs the opposite: one recipient, never stored (an SDP offer must not appear in a joiner's presence snapshot). So the server gains one message type, and the client gains one entrypoint. ## Server: `signal` message ```ts // Client → Server interface SignalMessage { type: "signal"; to: string; data: Record; } // Server → Client interface SignalEvent { type: "signal"; from: string; userId: string; data: Record; } ``` - **Targeted, not stored.** Delivered to the client `to` only. Never touches `EphemeralAdapter.set`. - **Authorization: shared room.** Delivered only if sender and recipient share at least one subscription with the same room, matched exactly. A sender whose subscription has no room reaches only recipients whose subscription to that query also has none, never a client inside a room. The sender's rooms are resolved on the origin instance (same as `targets` today); the recipient check runs on the instance holding the recipient's socket. No shared room means a silent drop, so a client cannot probe for the existence of another client ID. - **`from` and `userId` come from the session**, never from the client. - **Metered** in the existing ephemeral bucket (`ephemeralPerSecond`). Signaling is a few dozen messages per connection, far below the limit. - **Size cap 4 KB** per signal; larger frames are dropped. The cap stays under the Postgres NOTIFY limit (8000 bytes) with room for the envelope. A video SDP can run several KB, so the mesh splits descriptions into 1500-character `part` frames and reassembles them. - **Across a fleet:** `EphemeralBroadcast` gains optional `to?: string` and `kind?: "signal"`. Existing adapters carry it unchanged (they serialize the whole event), so Redis, Postgres, and memory need no changes. `deliverRemoteEphemeral` routes `kind: "signal"` to the one socket if it is local and shares a target room. - **No protocol version bump.** An old server rejects `signal` as an unknown type, so no peer ever connects. The mesh gives up after `connectTimeoutMs` (15 s) and stays on the server path. - **`hello_ack` carries `clientId`**, the server's id for the connection. Peers see each other by that id, and the transport chooses it, so it can differ from the client's configured `clientId`. `SyncClient.getClientId()` returns it once connected. A mesh that sees it change after a reconnect drops its peers and starts over. ## Client: `reflectdb/client/webrtc` Lives in its own entrypoint and does not touch the core bundle. Imports nothing from the rest of the library. ```ts import { createPeerMesh } from "reflectdb/client/webrtc"; const mesh = createPeerMesh(client, { iceServers: [{ urls: "stun:stun.l.google.com:19302" } /* TURN */], maxPeers: 8, // full-mesh ceiling; peers past it stay on the server path relayOnly: false, // true = iceTransportPolicy "relay": hides peer IPs, needs TURN }); mesh.getSnapshot(); // { peers: PeerInfo[], call: CallState | null } mesh.subscribe(listener); // for useSyncExternalStore or any store mesh.destroy(); ``` **Peer discovery.** Discovery uses existing presence: each mesh announces itself on the reserved ephemeral key `__rtc` (`{ v: 1 }`, refreshed every `announceIntervalMs`, default 30 s, with a TTL of three intervals). Each refresh fans out to the whole room, so the cost is O(room²) messages per interval; discovery never waits on a refresh (see below). A peer appears via the presence snapshot or a live `__rtc` event, and leaves when that entry expires or the connection closes. **Negotiation.** This is the standard "perfect negotiation" pattern. The lower `clientId` opens the connection and is impolite; the higher one is polite and yields on an offer collision. Both sides agree without a round-trip. **Failure.** Two cases, handled differently: - _Never connected_ within `connectTimeoutMs`: the peer is marked `server-only` and stays on the server path. After a 60 s cool-down, its next announce or offer starts a fresh attempt. If it falls silent for three announce intervals, it is dropped. - _Connected, then lost_ (`disconnected` or `failed`): the peer leaves the direct path at once (state `connecting`, so `auto` stops throttling for it). ICE is restarted on `failed`, at most twice. If the connection has not recovered within `connectTimeoutMs`, the peer is treated as gone and dropped until it announces again. A clean close from the far side (including a closing tab, which closes its connections on `pagehide`) drops the peer immediately. **Signals are handled one at a time, in arrival order**, so an answer is never built while the offer before it is still being applied. **`call()` and `shareScreen()` are idempotent while a capture is pending.** A second call waits on the first rather than opening the device again. `leave()` stops every local track, including those of a stream the app passed in. **Channels per peer.** - `eph` — `{ ordered: false, maxRetransmits: 0 }`. Ephemeral frames. - `ctl` — reliable, JSON. The whole control vocabulary: - `{ t: "call", on: boolean }`: joined or left the call. - `{ t: "stream", id, kind: "camera" | "screen" }`: label for a stream's tracks, sent before they are added. - `{ t: "unstream", id }`: that stream was removed. - `{ t: "mute", audio: boolean, video: boolean }`: current mute state. ## Routing ephemeral frames `sendEphemeral` / `useEphemeral` gain `transport`: | `transport` | P2P-connected peers | Everyone else | Server store / joiner snapshot | | ----------------------------------- | ------------------- | ------------- | ----------------------------------------------------------------- | | `"server"` (default without a mesh) | server | server | every frame | | `"auto"` (default with a mesh) | P2P every frame | server | throttled while every known peer is direct; every frame otherwise | | `"p2p"` | P2P every frame | nothing | nothing | - **`auto` throttle.** While every peer the mesh knows of is connected directly, the server copy goes out at most every 100 ms per key (leading and trailing edge, so the final position always lands). Joiner snapshots and clients without a mesh then see 10 Hz while connected peers get 60 Hz. As soon as any known peer is not direct, every frame also goes through the server, and a parked trailing frame is dropped, so the server never ends on a stale value. - **Dedupe.** Every frame carries `seq` (monotonic per client and key). A receiver keeps the highest `seq` per (clientId, key) and drops older frames. That covers the same frame arriving over both paths and out-of-order data channel delivery. `seq` sits in a reserved `__seq` field of `data`, so the wire protocol is unchanged. - **`p2p`** is for game ticks (#4). Never stored, never seen by peers past `maxPeers`. The game must tolerate this; authoritative state belongs in ops. - Drag previews (#3) use `auto`. The drop is an ordinary mutation, and the preview key is cleared on drop. - Follow mode (#5a) is the `viewport` key on `auto`. It needs no new API, only a documented recipe. ## Media (#2, #5b) ```ts await mesh.call(); // camera + mic; or mesh.call(stream) with your own MediaStream await mesh.shareScreen(); // adds a "screen" stream alongside the camera mesh.getSnapshot().call?.participants; // [{ clientId, userId, camera, screen, audioMuted, videoMuted }] mesh.setMuted({ audio: true }); mesh.leave(); // also cancels a join still waiting on the camera prompt ``` - A call is opt-in per peer. Joining sets the `__rtc` presence entry to `{ v: 1, call: true }`. Tracks flow only between peers that have both joined, and are added via renegotiation on the existing connection. - `ctl` carries `{ t: "stream", id, kind: "camera" | "screen" }` before the tracks are added, so the receiver can tell a screen apart from a camera. - Full mesh only. Above ~6 video participants, uplink bandwidth becomes the limit. The doc says so and points to an SFU (LiveKit, Cloudflare Calls); building an SFU is out of scope. ## ICE servers (STUN/TURN): pluggable No provider is built in or hardcoded. The server picks one, the browser gets short-lived credentials over the connection it already has, and an app can bypass both. - **Server:** `server.iceServers(provider)`, where a provider is `(ctx: { auth }) => { iceServers, ttlSeconds? }` (sync or async). Built-ins live in `reflectdb/server/turn`: - `cloudflareTurn({ keyId, apiToken, ttlSeconds? })`: calls Cloudflare Realtime `POST /v1/turn/keys/{keyId}/credentials/generate-ice-servers`. It caches one credential for half its TTL, so a busy room makes one API call, not one per peer. It drops `:53` URLs, which browsers block. - `sharedSecretTurn({ urls, secret, ttlSeconds? })`: the TURN REST-API scheme (coturn `use-auth-secret`, and most self-hosted servers). The username is `expiry:userId` and the credential is HMAC-SHA1 of it under the shared secret. No network call. - `staticIceServers(servers)`: a fixed list, such as STUN only. - Any function you write, for Twilio, Metered or your own service. - **Protocol:** the client sends `{ type: "ice_request" }` and the server answers `{ type: "ice_servers", iceServers, expiresAt }`, with an empty list when no provider is set. Requests are metered in the ephemeral bucket and need an authenticated session, so credentials never sit at an open URL. An old server ignores the request, and the client times out after 5 s. - **Client:** `createPeerMesh` / `` take `iceServers` as: - omitted (default): ask the server, and fall back to Google STUN if it has no provider; - an array: use it as is; - a function: called for servers, returning an array or `{ iceServers, expiresAt }`, so an app can fetch credentials from anywhere. The mesh resolves ICE servers before it announces or answers anyone, refreshes them a minute before `expiresAt`, and pushes fresh credentials into live connections with `setConfiguration`, so a long call outlives its first credential. - **`relayOnly` without TURN** logs a warning: every peer would time out to `server-only`. - **Relayed or direct:** `PeerInfo.relayed` reports whether the selected candidate pair goes through TURN, read from `getStats()` once connected. ## Bindings - React: ``, `usePeers()`, `useCall()`; `useEphemeral` accepts `transport`. Svelte and vanilla ephemeral bindings accept `transport`; the mesh itself is framework-agnostic (`getSnapshot` / `subscribe`). htmx is out: there is no client JS to own a mesh. ## Security notes (for the README) - Peers in a room see each other's IP addresses through ICE candidates. Use `relayOnly: true` with TURN to hide them. - A malicious peer can send anything on the data channel. Validate P2P frames like untrusted input; the server rate limit no longer applies to them. - The `userId` on a P2P frame is whatever the peer claims. The mesh stamps the `userId` learned through server-relayed signaling, not the frame's own value. **Discovery without a snapshot.** Rooms that have no presence snapshot (queries with no room) only learn of a peer when it announces. When a peer with a lower id announces, the higher-id peer replies with a targeted `hello` signal, so the lower-id peer can open the connection immediately instead of waiting for the next announce. ## Testing - **Server (bun test):** signal routing on one node; delivery across instances through an in-process bus that round-trips events through JSON, as Redis and Postgres do (both carry the event opaquely, so they need no signal-specific code); the no-shared-room drop; the size cap. - **Client (bun test):** real data channels through `node-datachannel`'s polyfill (a dev dependency): connect, `p2p` delivery, `auto` throttle, fallback, connection-id identity, and discovery without a snapshot. The polyfill never fires `negotiationneeded`, so the test restores that one event. `seq` dedupe and routing without a mesh are unit-tested on `SyncClient`. - **Real WebRTC:** two tabs on `test/ui` checked with the browser agent: cursors over P2P (the server frame rate drops to 10 Hz), then a call with a screen share. ## Phases 1. `signal` message, server and fleet routing, and tests. 2. `createPeerMesh` with data channels, `transport` routing and dedupe, and the React hooks. Covers #1, #3, #4, #5a. 3. Media: `call`, `shareScreen`, `useCall`. Covers #2 and #5b. 4. README section and a `transport` option on the Svelte and vanilla ephemeral bindings (both done). The mesh itself is framework-agnostic (`getSnapshot` / `subscribe`), so it needs no per-framework wrapper. A landing-page demo is still to do. ## Decided - `auto` is the default whenever a mesh exists. `transport: "server"` opts a key out. - `call()` and `shareScreen()` accept a ready `MediaStream` as well as capture constraints, for processed video, canvas sources, and tests that avoid a permission prompt. - `iceServers` defaults to Google's public STUN (`stun:stun.l.google.com:19302`). No TURN ships by default; apps supply their own. --- Source: https://reflectdb.dev/skills/reflectdb/SKILL.md --- name: reflectdb description: Use when adding ReflectDB to a TypeScript app, defining synced queries, connecting clients, configuring persistence or transports, adding presence or WebRTC, or debugging ReflectDB sync, offline replay, conflicts, and authorization. --- # ReflectDB Build or fix the ReflectDB integration requested in the current conversation. Use the request accompanying the skill invocation as the task. If invoked without a task, inspect the current integration and ask which behavior to add or fix. This skill works in Claude Code and Codex using their available file, shell, and browsing tools. ## Find the matching API Inspect the app's package manifest and lockfile to identify the ReflectDB version, runtime, framework, database, and transport. Preserve the app's stack. Install `reflectdb` with its existing package manager when needed; add optional peers only for the integrations in use. Use the installed package's exports and declarations to verify signatures. In the ReflectDB repository, inspect `src/` and the relevant tests instead. Read [integration.md](https://reflectdb.dev/skills/reflectdb/references/integration.md) for the typed schema, server, and client pattern. The reference targets ReflectDB 0.4.1; adapt it to the installed declarations when versions differ. For recipes beyond the reference, read the matching section in the package's README or the repository README. Online fallback: . Check version-specific source when current online docs differ from the installed package. Resolve reference paths relative to this skill directory, not the app's working directory. ## Implement sync 1. Define shared queries with `defineSyncQueries` and `t()`, or infer rows from existing Drizzle tables. Declare typed params for scoped subscriptions. Keep server database handles and secrets out of browser imports. 2. Create the server with `createSyncServer`. Implement each writable query's reads and mutations against the app's database. Authenticate connections with the app's existing session verification, and authorize scoped reads and writes from the authenticated context. 3. Wire a transport supported by the deployment. Use WebSocket for a process that can hold connections; use SSE or polling where HTTP fits. Connect lifecycle handlers. In 0.4.1, `server.auth` receives a synthesized request whose Bearer header contains the client's `token`, not upgrade cookies. Bridge an existing cookie session to a verified sync token when needed. 4. Connect through the matching typed client or framework factory. Use IndexedDB when offline data and queued writes must survive a reload. Namespace browser storage by authenticated user, keep instances stable, and remount the provider when identity changes. Unmount it on logout and apply the app's local-cache deletion policy. Supply `onReauth` for token refresh; changing the token prop alone does not reauthenticate a client. 5. Verify the changed path with the app's type check and a focused test or running example. For sync changes, check two clients receiving the same accepted write; for offline changes, check a queued write after reconnect. Report the behavior actually checked and any verification left undone. ## Preserve these contracts - **Two stores:** `db` holds application rows; server `storage` holds sync rows, clocks, and the op log. Persisting the op log does not persist an in-memory application Map. Configure both according to the durability requirement. `createSqliteStorage` requires Bun; use a compatible adapter on Node. - **Mutations:** the row ID is a separate argument. Updates are partial patches, so merge them into the existing row or use an SQL update. Implement deletion explicitly. `t()` provides no runtime validation: validate incoming fields and distinguish insert requirements from update patches. - **Ownership:** omit `id`, `readonly`, and `serverSet` fields from client payloads. Array-form `serverSet` needs values in `implement`; object-form declarations can supply values in the schema. Use authenticated context for ownership fields, and authorize against stored rows for writes. - **Isolation:** params and room names come from clients. Verify membership, scope database reads and mutations, and set `room` in `implement` when a query requires that room pattern. Only share results with `groupBy` when every subscriber in the group is allowed the identical rows and fields. - **Delivery:** `lww` compares row timestamps; `merge` compares per-column timestamps for concurrent field edits. Both accept partial update patches. Use schema `presence()` or ephemeral channels for cursors and typing. Views use `view()` and `server.view()` and reject writes. Keep persisted data on the durable sync path. ## Choose the integration | Need | Entry point or reference | | ---------------------------- | --------------------------------------------------------------------------------------------------------------------------- | | Shared schema | `reflectdb/core` (also exported by `reflectdb`) | | Server and op log adapters | `reflectdb/server`; Drizzle helpers at `reflectdb/server/drizzle` | | Framework-independent client | `createSyncClient` from `reflectdb/client` | | React | `createSyncReact` from `reflectdb/react`; its provider uses WebSocket | | Svelte, vanilla JS, htmx | `createSyncSvelte`, `createSyncVanilla`, `createSyncHtmx` from their corresponding subpaths; inspect each factory's options | For serverless SSE, read the README's “Serverless sync on Vercel” recipe: both transport ends need `serverless: true` and POST responses must return `collectReplies`. Use a client accepting a custom transport; React's standard `SyncProvider` takes a WebSocket URL, not a transport prop. For object storage, read `docs/object-storage.md` in a ReflectDB checkout or . For peer meshes, calls, screen share, or TURN, read `docs/webrtc.md` or and verify the installed `reflectdb/client/webrtc` and `reflectdb/react/webrtc` exports. ## Diagnose sync failures Trace one failing write through the transport, authentication, subscription, authorization, mutation, op acknowledgment, and recipient query result. Capture the actual rejection reason before changing config. | Symptom | Inspect first | | --------------------------------------- | ----------------------------------------------------------------------------------------- | | Connected but no rows | Declared subscription, required params, bootstrap/resume, query result, room access | | Local write disappears | Rejected op, runtime validation, authorization, conflict policy, partial-update handling | | Rows disappear on restart | Application database persistence separately from op log storage | | Offline edits disappear on reload | IndexedDB adapter, storage hydration, client identity, pending-op replay | | Server-side write never reaches clients | ReflectDB write/transaction API or explicit notification after an external database write | Use the relevant tests as protocol examples when working in this repository. Keep fixes scoped to the failing path; finish with the observed before/after behavior and verification result. --- Source: https://reflectdb.dev/skills/reflectdb/references/integration.md # Typed integration This example targets ReflectDB 0.4.1. Use the app's runtime and database; the Bun transport and in-memory Map below are an example for local development. The Map loses application rows on process exit. ## Shared schema ```ts // schema.ts import { defineSyncQueries, t } from "reflectdb/core"; export type Todo = { id: string; title: string; done: boolean }; export const queries = defineSyncQueries({ todos: { row: t(), conflict: "merge" }, }); ``` ## Server ```ts // server.ts import { createSyncServer } from "reflectdb/server"; import { createBunWsServerTransport } from "reflectdb/transport/bun-ws"; import { queries, type Todo } from "./schema"; const db = new Map(); const { transport, websocket } = createBunWsServerTransport(); const server = createSyncServer({ queries, db, transport, serverId: "local-dev", }); // Local demo identity. Replace with the app's session verification before // exposing this endpoint. Scope queries and authorize writes for private data. server.auth(async () => ({ userId: "local-demo" })); server.implement("todos", { query: (_ctx, db) => [...db.values()], mutate: async (op, _ctx, db) => { if (op.type === "delete") { db.delete(op.rowId); return; } const patch = op.payload ?? {}; if (Object.keys(patch).some((key) => key !== "title" && key !== "done")) { throw new Error("Unknown todo field"); } if ("title" in patch && typeof patch.title !== "string") { throw new Error("title must be a string"); } if ("done" in patch && typeof patch.done !== "boolean") { throw new Error("done must be a boolean"); } const existing = db.get(op.rowId); if (op.type === "update" && !existing) { throw new Error("Todo not found"); } const row = { ...existing, ...patch, id: op.rowId }; if (typeof row.title !== "string" || typeof row.done !== "boolean") { throw new Error("title and done are required"); } db.set(op.rowId, { id: row.id, title: row.title, done: row.done }); }, }); const http = Bun.serve({ port: Number(process.env.PORT ?? 0), // 0 chooses an available port. websocket, fetch(req, srv) { if (new URL(req.url).pathname === "/sync") { if (srv.upgrade(req, { data: { id: crypto.randomUUID(), req } })) return; return new Response("WebSocket upgrade required", { status: 426 }); } return new Response("Not found", { status: 404 }); }, }); console.log(`Sync endpoint: ws://localhost:${http.port}/sync`); ``` In 0.4.1, `server.auth` receives a synthesized request with `Authorization: Bearer `. The upgrade request's cookies and headers are not passed through to that callback, even though the transport receives `req`. Use the app's token verifier, or exchange an authenticated cookie session for a sync token through the app's existing HTTP auth flow. In an existing app, mount the transport on its HTTP server instead of creating a second listener. Use `wss:` on HTTPS pages. ## React client ```tsx // sync.ts import { createSyncReact } from "reflectdb/react"; import { queries } from "./schema"; export const { SyncProvider, useSync, useSyncStatus } = createSyncReact(queries); ``` ```tsx // app.tsx (browser entry) import { useMemo } from "react"; import { createIndexedDBStorage } from "reflectdb/client/storage/indexeddb"; import { SyncProvider, useSync } from "./sync"; export function App({ url, token, userId, refreshToken, }: { url: string; token: string; userId: string; refreshToken: () => Promise; }) { const storage = useMemo( () => createIndexedDBStorage({ dbName: `todos-demo:${userId}` }), [userId], ); return ( ); } function Todos() { const { rows, insert, update, remove } = useSync("todos"); return (
      {rows.map((todo) => ( ))}
      ); } ``` Use the server's printed URL and the app's session token. For this local demo, the server accepts any token. The provider owns initialization, connection, subscriptions listed in `tables`, resume/bootstrap, and teardown. Mount `App` after the authenticated identity is known. User-specific storage prevents one account from rendering another account's cached rows before the connection authenticates. Unmount on logout and delete that account's cache when the app's logout policy requires it. The provider key changes on account switch; `onReauth` obtains refreshed tokens from the app's auth flow. For required params, declare `params: t<{ projectId: string }>()` in the query and call `useSync("todos", { params: { projectId } })`. Leave parameterized queries out of the provider's `tables`; the hook declares that subscription. Authorize the project on the server and filter reads and writes by it. ## Other clients For a framework-independent client, inspect `createSyncClient` from `reflectdb/client`. It accepts `queries`, `clientId`, `transport`, `token`, and optional `storage`. Hydrate with `init()`, connect, declare subscriptions with `sync(name, params)`, and call `resume()` or `bootstrap()` as appropriate. Close the client on teardown. `autoSync: true` declares parameterless queries on connect; it does not replace the rest of the lifecycle. React's typed `useSync` uses `remove(id)`; the standalone client uses `delete(name, id)`. Both take IDs separately from mutation payloads.