diff --git a/CHANGELOG.md b/CHANGELOG.md index b7399b2..8a231c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,32 @@ # Changelog +## 0.2.0-next.24 + +- **Private realtime: live updates now arrive with the ROW on a private + channel, in one socket hop.** The client half of the platform's private + doorbell (codehs/bool#570; measured there: commit → observer ding ≈ 0–3ms, + before the writer's own insert response returns). + + `subscribeToChanges` now mints a short-TTL "wristband" from the gateway + (`POST /_bool/v1/realtime/token` — where liveAccess/session are re-checked on + EVERY mint), presents it via `realtime.setAuth`, and joins the app's private + topics: the app room, plus the personal room when signed in. It re-mints at + ~75% of the TTL; a refused re-mint (access revoked, gateway down) silences + the private rooms and degrades to the legacy public row-data-free ping — so + does a missing wristband desk (older platform) or a refused join. Live-ness + degrades, never dies. One doorbell is shared by all subscribers (previously + every `entities..subscribe` opened its own channel). + +- **Re-lands the live layer reverted in `0.2.0-next.23`** — `useEntity` + (`bool-sdk/react`), `LiveEntityStore`, `matchesFilter`, `compareBySort`, the + `id`/`row` payload fields, and the globalThis client registry — now on the + architecture it was built for: a ding carrying `row` applies with ZERO + fetches; only id-only dings (RLS-on tables without `owner_id`, legacy pings) + keyed-fetch through the gateway. + +- **Coalescing is now leading-edge.** A lone change reconciles immediately + (the trailing window taxed every single change 50ms just in case a burst was + coming); bursts still collapse into one pass per window. ## 0.2.0-next.20 - **Reverts the reload-hold behavior added in `0.2.0-next.19`.** That release diff --git a/package.json b/package.json index 0f3e510..001d023 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "bool-sdk", - "version": "0.2.0-next.23", + "version": "0.2.0-next.24", "description": "Client SDK for apps built on Bool — gateway data access, end-user auth, the AI battery, the React auth layer, and the local-dev CLI (link, types, deploy).", "type": "module", "main": "./dist/index.js", diff --git a/src/client.test.ts b/src/client.test.ts index ee5ada8..d69433b 100644 --- a/src/client.test.ts +++ b/src/client.test.ts @@ -2,6 +2,7 @@ import { beforeEach, describe, expect, test } from "bun:test"; import { createBoolClient, getDefaultBoolClient, + hasDefaultBoolClient, isDeploymentSubdomain, BoolAiError, type BoolClientConfig, @@ -414,6 +415,11 @@ describe("bool.ai battery", () => { }); }); +// The registry has to be a singleton across module INSTANCES, not just within +// one. Regression coverage for a real break: a generated app imported +// `createBoolClient` from "bool-sdk" and `useEntity` from "bool-sdk/react", +// nothing imported the bootstrap module at all, and every hook threw "No Bool +// client exists yet" at first render. describe("default client registry", () => { test("the last-created client is the default (hot reload re-registers)", () => { const first = createBoolClient(CONFIG); @@ -421,6 +427,64 @@ describe("default client registry", () => { const second = createBoolClient(CONFIG); expect(getDefaultBoolClient()).toBe(second); }); + + test("lives on a globalThis symbol, so a duplicated client.js shares it", () => { + // Simulates the second bundle: another copy of this module would read the + // same well-known symbol rather than its own module-scoped variable. + const client = createBoolClient(CONFIG); + const shared = (globalThis as any)[Symbol.for("bool-sdk.defaultClient")]; + expect(shared).toBe(client); + }); + + test("hasDefaultBoolClient reports registration without throwing", () => { + createBoolClient(CONFIG); + expect(hasDefaultBoolClient()).toBe(true); + }); + + test("the unregistered error names the exact fix (the import to add)", () => { + const saved = (globalThis as any)[Symbol.for("bool-sdk.defaultClient")]; + (globalThis as any)[Symbol.for("bool-sdk.defaultClient")] = null; + try { + expect(hasDefaultBoolClient()).toBe(false); + // The old message said "call createBoolClient() first", which is useless + // to someone whose app already calls it in a module nothing imports. + expect(() => getDefaultBoolClient()).toThrow(/import ".\/lib\/supabase"/); + expect(() => getDefaultBoolClient()).toThrow(/src\/main\.tsx/); + } finally { + (globalThis as any)[Symbol.for("bool-sdk.defaultClient")] = saved; + } + }); +}); + +// The doorbell's lifecycle is fully covered with fake deps in realtime.test.ts; +// here we pin the WIRING: subscribing asks the gateway's wristband desk at the +// right URL with credentials, and tearing down while the mint is in flight +// orphans it (no channel ever joins). +describe("subscribeToChanges wiring (private doorbell)", () => { + test("POSTs the mint URL with credentials; immediate unsubscribe orphans the start", async () => { + let mintCalls = 0; + respond = (url) => { + if (url.includes("/_bool/v1/realtime/token")) { + mintCalls++; + return Response.json({ + token: "t", + expiresIn: 900, + topics: { app: "bool:app_x:app", user: null }, + }); + } + return new Response("[]", { headers: { "content-type": "application/json" } }); + }; + const client = createBoolClient({ ...CONFIG, viewerToken: "vt-9" }); + const off = client.subscribeToChanges(() => {}); + off(); // torn down before the mint resolves + await tick(); + expect(mintCalls).toBe(1); + const call = calls.find((c) => c.url.includes("/_bool/v1/realtime/token"))!; + expect(call.url).toBe("https://bool.test/served/my-app/_bool/v1/realtime/token"); + expect(call.init?.method).toBe("POST"); + expect(call.init?.credentials).toBe("include"); + expect(headersOf(call).get("x-bool-viewer")).toBe("vt-9"); + }); }); describe("isDeploymentSubdomain", () => { diff --git a/src/client.ts b/src/client.ts index a6e822b..2cc6989 100644 --- a/src/client.ts +++ b/src/client.ts @@ -7,9 +7,12 @@ // requires it) but is powerless: a v2 schema has no anon/authenticated // grants, so a direct REST call 403s. // - Realtime connects directly to Supabase (the WS can't be proxied) and -// subscribes to the app's PUBLIC, row-data-free Broadcast "doorbell" -// channel `bool:` — no token needed (the payload is just -// {table, op}; the data itself stays behind the gateway). +// joins the app's PRIVATE doorbell topics using a short-TTL token minted +// by the gateway (/_bool/v1/realtime/token — where liveAccess/session are +// re-checked on every mint). Those dings carry the changed ROW when the +// table's audience is nameable, so live views update in one socket hop. +// Falls back to the legacy public row-data-free ping when the mint desk +// is unavailable. See realtime.ts. // - `auth` (end-user auth) routes to the gateway's users plane // (/_bool/v1/users) so the app can offer its own signup/login without ever // handling a credential server-side. @@ -18,6 +21,7 @@ // (/_bool/v1/users) in the Bool platform repo. import { createClient, type SupabaseClient } from "@supabase/supabase-js"; import { createEntitiesModule, type EntitiesModule } from "./entities.js"; +import { createDoorbell, type RealtimeMint } from "./realtime.js"; /** Matches the server's append-only gateway path version. */ const GATEWAY_API = "v1"; @@ -125,9 +129,21 @@ export type BoolAuth = { rotateApiKey(): Promise<{ data: { apiKey: string | null }; error: unknown }>; }; -/** A row-data-free change notification: some row in `table` saw `op`. Refetch - * whatever you derive from that table — the ping never carries the data. */ -export type BoolChangePayload = { table?: string; op?: string }; +/** A change notification: some row in `table` saw `op`. */ +export type BoolChangePayload = { + table?: string; + op?: string; + /** The changed row's id (older platform triggers ping without it — the live + * layer then degrades to a full reload). */ + id?: string | null; + /** The full row, present on PRIVATE doorbell dings for tables whose audience + * is nameable at write time (shared tables → the app room, owned rows → the + * owner's room). Absent on the legacy public ping — that channel is + * anon-key-reachable and must stay row-data-free — and for RLS-on tables + * without owner_id, where subscribers keyed-fetch through the gateway + * instead (RLS answers per viewer). */ + row?: Record; +}; /** A JSON Schema describing the shape `bool.ai.generate` should return. Passed * straight to the gateway, which validates the model's output against it. e.g. @@ -192,20 +208,44 @@ export type BoolClient = { // The last-created client, used by the React layer (bool-sdk/react) so app // components don't have to thread the client through props. Last-created wins // so a hot-reloaded `src/lib/supabase.ts` re-registers its fresh client. -let defaultClient: BoolClient | null = null; +// +// Held on `globalThis`, NOT in a module-scoped `let`, because the registry must +// be a true singleton across module *instances*. An app imports +// `createBoolClient` from "bool-sdk" and `useEntity` from "bool-sdk/react" — +// two separate entry points, which Vite's dep optimizer pre-bundles into two +// chunks (`bool-sdk.js`, `bool-sdk_react.js`). If client.js gets inlined into +// both, a module-scoped variable gives each chunk its OWN registry: the app +// registers its client in one and the hook reads `null` from the other, so +// every hook throws even though the app did everything right. A symbol on +// globalThis is shared by construction, whatever the bundler does with the +// module graph. +const REGISTRY = Symbol.for("bool-sdk.defaultClient"); +type Registry = { [REGISTRY]?: BoolClient | null }; + +function registry(): Registry { + return globalThis as unknown as Registry; +} export function getDefaultBoolClient(): BoolClient { - if (!defaultClient) { + const client = registry()[REGISTRY]; + if (!client) { throw new Error( - "No Bool client exists yet — call createBoolClient() first. " + - "(Bool apps do this in src/lib/supabase.ts; import from there.)", + "No Bool client exists yet. Add `import \"./lib/supabase\";` to " + + "src/main.tsx — that module calls createBoolClient() and registers it, " + + "and hooks like useEntity read it from there. (A file that only imports " + + "from \"bool-sdk/react\" never loads it on its own.)", ); } - return defaultClient; + return client; } export function setDefaultBoolClient(client: BoolClient): void { - defaultClient = client; + registry()[REGISTRY] = client; +} + +/** Is a client registered? Lets callers branch instead of catching a throw. */ +export function hasDefaultBoolClient(): boolean { + return Boolean(registry()[REGISTRY]); } export function createBoolClient(config: BoolClientConfig): BoolClient { @@ -590,24 +630,68 @@ export function createBoolClient(config: BoolClientConfig): BoolClient { }, }; - // Realtime "doorbell": the app schema's grants are revoked, so Supabase - // `postgres_changes` never fires. Instead the server broadcasts a - // row-data-free ping on the PUBLIC channel "bool:" + schema whenever any - // row changes. Subscribe with the anon key (no token needed) and REFETCH - // on each ping. + // Realtime doorbell — the private-channel design. Change payloads (with the + // ROW on them for tables whose audience is nameable) arrive on PRIVATE + // topics; the gateway mints the short-TTL wristband that lets this socket + // join them (/_bool/v1/realtime/token — it re-runs liveAccess/session on + // every mint, so revoked access silences the private rooms within the TTL). + // When the desk is unavailable (older platform, 503, offline) the doorbell + // falls back to the legacy PUBLIC row-data-free ping. All lifecycle logic + // lives in realtime.ts (unit-tested with fake deps); this is just the wiring + // of gateway fetch + supabase-js primitives into it. + // + // NOTE on sign-in: the wristband is minted with whatever identity the + // CURRENT session has. A user who signs in mid-session gains their personal + // room at the next TTL refresh (≤15 min) or page navigation — acceptable + // because the app room already carries all shared-table changes. + const mintRealtimeToken = async (): Promise => { + try { + const headers = new Headers(); + if (viewerToken) headers.set("x-bool-viewer", viewerToken); + if (euSessionToken) headers.set("x-bool-eu-session", euSessionToken); + if (apiKey) headers.set("api_key", apiKey); + const res = await fetch(`${GATEWAY}/_bool/${GATEWAY_API}/realtime/token`, { + method: "POST", + headers, + credentials: "include", + }); + if (!res.ok) return null; + return (await res.json()) as RealtimeMint; + } catch { + return null; + } + }; + + const doorbell = createDoorbell({ + mint: mintRealtimeToken, + setAuth: async (token) => { + await db.realtime.setAuth(token); + }, + channel: (topic, opts) => { + // Adapt one supabase RealtimeChannel to the doorbell's minimal shape. + // The channel is created lazily-configured and joined in join(), so + // onBroadcast registration always precedes the subscribe. + const ch = db.channel(topic, { config: { private: opts.private } }); + return { + onBroadcast(cb) { + ch.on("broadcast", { event: "*" }, (msg) => + cb((msg as { payload?: BoolChangePayload }).payload ?? {}), + ); + }, + join(status) { + ch.subscribe((state) => status(state)); + }, + leave() { + void db.removeChannel(ch); + }, + }; + }, + legacyTopic: "bool:" + schema, + }); + const subscribeToChanges = ( listener: (payload: BoolChangePayload) => void, - ): (() => void) => { - const channel = db - .channel("bool:" + schema) - .on("broadcast", { event: "*" }, (msg) => - listener((msg as { payload?: BoolChangePayload }).payload ?? {}), - ) - .subscribe(); - return () => { - void db.removeChannel(channel); - }; - }; + ): (() => void) => doorbell.subscribe(listener); const client: BoolClient = { db, diff --git a/src/index.ts b/src/index.ts index fe37e82..bc06389 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,6 +2,7 @@ export { createBoolClient, getDefaultBoolClient, setDefaultBoolClient, + hasDefaultBoolClient, type BoolClient, type BoolClientConfig, type BoolAuth, @@ -28,3 +29,11 @@ export { type UpdateManyResult, type ImportResult, } from "./entities.js"; +export { + LiveEntityStore, + matchesFilter, + compareBySort, + type EntityRow, + type LiveQueryOptions, + type LiveSnapshot, +} from "./live.js"; diff --git a/src/live.test.ts b/src/live.test.ts new file mode 100644 index 0000000..6deaac6 --- /dev/null +++ b/src/live.test.ts @@ -0,0 +1,509 @@ +import { describe, expect, test } from "bun:test"; +import type { BoolChangePayload } from "./client.js"; +import type { EntityHandler, FilterQuery, SortSpec } from "./entities.js"; +import { + compareBySort, + LiveEntityStore, + matchesFilter, + type EntityRow, +} from "./live.js"; + +// --------------------------------------------------------------------------- +// Harness: a fake entity handler with a controllable server-side row set and +// manual promise resolution, so response ordering (the racy part) is exact. +// --------------------------------------------------------------------------- + +type Row = EntityRow & { title?: string; done?: boolean; rank?: number | null }; + +function deferred() { + let resolve!: (v: V) => void; + let reject!: (e: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function makeHarness(initial: Row[] = []) { + const rows = new Map(initial.map((r) => [r.id, { ...r }])); + const listeners = new Set<(p: BoolChangePayload) => void>(); + const calls: { list: number; filter: FilterQuery[]; creates: Partial[] } = { + list: 0, + filter: [], + creates: [], + }; + // When set, the next list/filter call parks on this deferred instead of + // resolving immediately (then clears, so later calls auto-resolve). + let gate: ReturnType> | null = null; + + const visible = () => [...rows.values()].map((r) => ({ ...r })); + const handler = { + async list() { + calls.list++; + if (gate) { + const g = gate; + gate = null; + return g.promise; + } + return visible(); + }, + async filter(q: FilterQuery) { + calls.filter.push(q); + if (gate) { + const g = gate; + gate = null; + return g.promise; + } + // Only the by-id shape the store issues needs to work here. + const ids = q.id as string[] | undefined; + return visible().filter((r) => !ids || ids.includes(r.id)); + }, + async get(id: string) { + const r = rows.get(id); + if (!r) throw new Error("not found"); + return { ...r }; + }, + async create(v: Partial) { + calls.creates.push(v); + const row = { ...v, id: v.id as string } as Row; + rows.set(row.id, row); + return { ...row }; + }, + async update(id: string, v: Partial) { + const row = { ...rows.get(id)!, ...v, id }; + rows.set(id, row); + return { ...row }; + }, + async delete(id: string) { + rows.delete(id); + return { success: true }; + }, + subscribe(cb: (p: BoolChangePayload) => void) { + listeners.add(cb); + return () => listeners.delete(cb); + }, + } as unknown as EntityHandler; + + return { + handler, + rows, + calls, + ding(p: BoolChangePayload) { + for (const l of listeners) l(p); + }, + gateNextLoad() { + gate = deferred(); + return gate; + }, + }; +} + +const tick = (ms = 0) => new Promise((r) => setTimeout(r, ms)); +// Past the 50ms coalesce window, plus a beat for the keyed fetch to settle. +const settle = () => tick(70); + +// --------------------------------------------------------------------------- +// matchesFilter — mirrors the server translation, incl. SQL null semantics +// --------------------------------------------------------------------------- + +describe("matchesFilter", () => { + const row = { id: "1", status: "active", count: 10, tags: ["a", "b"], archived_at: null }; + + test("scalar equality, null, and array (IN) shorthands", () => { + expect(matchesFilter(row, { status: "active" })).toBe(true); + expect(matchesFilter(row, { status: "done" })).toBe(false); + expect(matchesFilter(row, { archived_at: null })).toBe(true); + expect(matchesFilter(row, { status: null })).toBe(false); + expect(matchesFilter(row, { status: ["active", "paused"] })).toBe(true); + expect(matchesFilter(row, { status: ["done"] })).toBe(false); + }); + + test("comparison operators", () => { + expect(matchesFilter(row, { count: { $gte: 10 } })).toBe(true); + expect(matchesFilter(row, { count: { $gt: 10 } })).toBe(false); + expect(matchesFilter(row, { count: { $lt: 11, $gte: 10 } })).toBe(true); + expect(matchesFilter(row, { count: { $ne: 5 } })).toBe(true); + expect(matchesFilter(row, { count: { $in: [1, 10] } })).toBe(true); + expect(matchesFilter(row, { count: { $nin: [1, 10] } })).toBe(false); + }); + + test("SQL three-valued logic: a NULL column fails comparisons, like PostgREST", () => { + expect(matchesFilter(row, { archived_at: { $ne: "x" } })).toBe(false); + expect(matchesFilter(row, { archived_at: { $gt: "2020" } })).toBe(false); + expect(matchesFilter(row, { archived_at: { $exists: false } })).toBe(true); + expect(matchesFilter(row, { count: { $exists: true } })).toBe(true); + // absent key behaves like NULL + expect(matchesFilter(row, { missing: { $eq: null } })).toBe(true); + expect(matchesFilter(row, { missing: { $ne: 1 } })).toBe(false); + }); + + test("$regex, $all, $not", () => { + expect(matchesFilter(row, { status: { $regex: "^act" } })).toBe(true); + expect(matchesFilter(row, { status: { $regex: "^x" } })).toBe(false); + expect(matchesFilter(row, { tags: { $all: ["a", "b"] } })).toBe(true); + expect(matchesFilter(row, { tags: { $all: ["a", "z"] } })).toBe(false); + expect(matchesFilter(row, { count: { $not: { $eq: 5 } } })).toBe(true); + expect(matchesFilter(row, { count: { $not: { $gte: 5 } } })).toBe(false); + }); + + test("$and / $or / $nor", () => { + expect(matchesFilter(row, { $and: [{ status: "active" }, { count: { $gt: 5 } }] })).toBe(true); + expect(matchesFilter(row, { $or: [{ status: "done" }, { count: 10 }] })).toBe(true); + expect(matchesFilter(row, { $or: [{ status: "done" }, { count: 11 }] })).toBe(false); + expect(matchesFilter(row, { $nor: [{ status: "done" }] })).toBe(true); + expect(matchesFilter(row, { $nor: [{ status: "active" }] })).toBe(false); + }); +}); + +// --------------------------------------------------------------------------- +// compareBySort — postgres default order, stable ties +// --------------------------------------------------------------------------- + +describe("compareBySort", () => { + const rows: Row[] = [ + { id: "b", rank: 2 }, + { id: "a", rank: 1 }, + { id: "c", rank: null }, + { id: "d", rank: 2 }, + ]; + + test("ascending puts nulls last; descending puts them first", () => { + expect([...rows].sort(compareBySort("rank")).map((r) => r.id)).toEqual(["a", "b", "d", "c"]); + expect([...rows].sort(compareBySort("-rank")).map((r) => r.id)).toEqual(["c", "b", "d", "a"]); + }); + + test("ties break on id for stability", () => { + const sorted = [...rows].sort(compareBySort("rank")); + expect(sorted[1]!.id).toBe("b"); + expect(sorted[2]!.id).toBe("d"); + }); +}); + +// --------------------------------------------------------------------------- +// LiveEntityStore — the state machine +// --------------------------------------------------------------------------- + +describe("LiveEntityStore: loading + dings", () => { + test("initial load populates the snapshot and clears loading", async () => { + const h = makeHarness([{ id: "1", title: "one" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + expect(store.getSnapshot().loading).toBe(false); + expect(store.getSnapshot().data.map((r) => r.id)).toEqual(["1"]); + stop(); + }); + + test("a burst of dings coalesces into ONE keyed fetch, not N reloads", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + expect(h.calls.list).toBe(1); + + for (let i = 2; i <= 20; i++) h.rows.set(String(i), { id: String(i) }); + for (let i = 2; i <= 20; i++) h.ding({ table: "t", op: "INSERT", id: String(i) }); + await settle(); + + expect(h.calls.list).toBe(1); // never re-listed + expect(h.calls.filter.length).toBe(1); // ONE by-id fetch for the whole burst + expect((h.calls.filter[0]!.id as string[]).length).toBe(19); + expect(store.getSnapshot().data.length).toBe(20); + stop(); + }); + + test("LEADING EDGE: a lone ding reconciles immediately, not after the window", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + h.rows.set("2", { id: "2" }); + h.ding({ table: "t", op: "INSERT", id: "2" }); + await tick(15); // well under the 50ms window + expect(store.getSnapshot().data.length).toBe(2); // already applied + stop(); + }); + + test("LEADING EDGE: a burst right after a pass still collapses into one more fetch", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + // First ding fires a pass immediately... + h.rows.set("2", { id: "2" }); + h.ding({ table: "t", op: "INSERT", id: "2" }); + await tick(10); + const after1 = h.calls.filter.length; + expect(after1).toBe(1); + // ...then a burst inside the window batches into exactly one more pass. + for (let i = 3; i <= 12; i++) { + h.rows.set(String(i), { id: String(i) }); + h.ding({ table: "t", op: "INSERT", id: String(i) }); + } + await settle(); + expect(h.calls.filter.length).toBe(2); + expect(store.getSnapshot().data.length).toBe(12); + stop(); + }); + + test("DELETE dings apply locally with zero fetches", async () => { + const h = makeHarness([{ id: "1" }, { id: "2" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + h.ding({ table: "t", op: "DELETE", id: "2" }); + await settle(); + + expect(store.getSnapshot().data.map((r) => r.id)).toEqual(["1"]); + expect(h.calls.filter.length).toBe(0); + expect(h.calls.list).toBe(1); + stop(); + }); + + test("a ding with no id (old trigger) degrades to one coalesced full reload", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + h.rows.set("2", { id: "2" }); + h.ding({ table: "t", op: "INSERT" }); + h.ding({ table: "t", op: "INSERT" }); + h.ding({ table: "t", op: "INSERT" }); + await settle(); + + expect(h.calls.list).toBe(2); // initial + ONE reload for the burst + expect(store.getSnapshot().data.length).toBe(2); + stop(); + }); + + test("a row carried ON the ding applies with zero fetches (private-channel forward-compat)", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + h.ding({ table: "t", op: "INSERT", id: "2", row: { id: "2", title: "pushed" } }); + await settle(); + + expect(store.getSnapshot().data.find((r) => r.id === "2")?.title).toBe("pushed"); + expect(h.calls.filter.length).toBe(0); + expect(h.calls.list).toBe(1); + stop(); + }); + + test("an id fetched but not returned (deleted / RLS-hidden) leaves the view", async () => { + const h = makeHarness([{ id: "1" }, { id: "2" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + h.rows.delete("2"); // row vanishes server-side before the keyed fetch lands + h.ding({ table: "t", op: "UPDATE", id: "2" }); + await settle(); + + expect(store.getSnapshot().data.map((r) => r.id)).toEqual(["1"]); + stop(); + }); + + test("filtered view: a changed row that stops matching drops out; one that starts matching drops in", async () => { + const h = makeHarness([ + { id: "1", done: false }, + { id: "2", done: true }, + ]); + const store = new LiveEntityStore(h.handler, { filter: { done: false } }); + const stop = store.start(); + await tick(); + + h.rows.set("1", { id: "1", done: true }); // leaves the filter + h.rows.set("2", { id: "2", done: false }); // enters it + h.ding({ table: "t", op: "UPDATE", id: "1" }); + h.ding({ table: "t", op: "UPDATE", id: "2" }); + await settle(); + + expect(store.getSnapshot().data.map((r) => r.id)).toEqual(["2"]); + stop(); + }); +}); + +describe("LiveEntityStore: response ordering (the anti-rewind guard)", () => { + test("a stale full-load response landing after a newer one is DROPPED", async () => { + const h = makeHarness([{ id: "1", title: "fresh" }]); + const store = new LiveEntityStore(h.handler); + + const stale = h.gateNextLoad(); // first load parks + const stop = store.start(); + await tick(); + await store.refetch(); // second load resolves immediately with "fresh" + expect(store.getSnapshot().data[0]!.title).toBe("fresh"); + + stale.resolve([{ id: "1", title: "stale" }]); // now the FIRST response lands + await tick(); + + expect(store.getSnapshot().data[0]!.title).toBe("fresh"); // not rewound + stop(); + }); +}); + +describe("LiveEntityStore: optimistic mutations", () => { + test("create renders immediately, settles to the committed row", async () => { + const h = makeHarness(); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + const p = store.create({ title: "new" }); + expect(store.getSnapshot().data.length).toBe(1); // before the await + const created = await p; + expect(created?.title).toBe("new"); + expect(store.getSnapshot().data.length).toBe(1); // still one row (same id) + expect(store.getSnapshot().error).toBeNull(); + stop(); + }); + + test("an optimistic row SURVIVES a full refetch that doesn't include it yet", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + // Park the create so its overlay stays in-flight... + let releaseCreate!: () => void; + const origCreate = h.handler.create.bind(h.handler); + h.handler.create = (async (v: Partial) => { + await new Promise((r) => (releaseCreate = r)); + return origCreate(v); + }) as typeof h.handler.create; + + const p = store.create({ id: "opt-1", title: "mine" }); + await tick(); + // ...and refetch while it's pending: the server list lacks opt-1. + await store.refetch(); + const ids = store.getSnapshot().data.map((r) => r.id); + expect(ids).toContain("opt-1"); // did NOT flicker away + + releaseCreate(); + await p; + expect(store.getSnapshot().data.map((r) => r.id)).toContain("opt-1"); + stop(); + }); + + test("create failure rolls back and surfaces the error without throwing", async () => { + const h = makeHarness(); + h.handler.create = (async () => { + throw new Error("insert failed"); + }) as typeof h.handler.create; + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + const created = await store.create({ title: "doomed" }); + expect(created).toBeNull(); + expect(store.getSnapshot().data.length).toBe(0); // rolled back + expect((store.getSnapshot().error as Error).message).toBe("insert failed"); + stop(); + }); + + test("update overlays instantly and rolls back on failure", async () => { + const h = makeHarness([{ id: "1", title: "old" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + // success path + await store.update("1", { title: "new" }); + expect(store.getSnapshot().data[0]!.title).toBe("new"); + + // failure path + h.handler.update = (async () => { + throw new Error("update failed"); + }) as typeof h.handler.update; + const updated = await store.update("1", { title: "doomed" }); + expect(updated).toBeNull(); + expect(store.getSnapshot().data[0]!.title).toBe("new"); // restored + stop(); + }); + + test("remove hides instantly and restores on failure", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + h.handler.delete = (async () => { + throw new Error("delete failed"); + }) as typeof h.handler.delete; + const p = store.remove("1"); + expect(store.getSnapshot().data.length).toBe(0); // hidden immediately + expect(await p).toBe(false); + expect(store.getSnapshot().data.length).toBe(1); // restored + stop(); + }); + + test("the doorbell echo of your own write reconciles to a no-op (no duplicates)", async () => { + const h = makeHarness(); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + + const created = await store.create({ title: "mine" }); + h.ding({ table: "t", op: "INSERT", id: created!.id }); // the echo + await settle(); + + expect(store.getSnapshot().data.length).toBe(1); + stop(); + }); +}); + +describe("LiveEntityStore: view shaping", () => { + test("sort order is maintained across delta applies", async () => { + const h = makeHarness([ + { id: "1", rank: 1 }, + { id: "3", rank: 3 }, + ]); + const store = new LiveEntityStore(h.handler, { sort: "rank" as SortSpec }); + const stop = store.start(); + await tick(); + + h.rows.set("2", { id: "2", rank: 2 }); + h.ding({ table: "t", op: "INSERT", id: "2" }); + await settle(); + + expect(store.getSnapshot().data.map((r) => r.id)).toEqual(["1", "2", "3"]); + stop(); + }); + + test("an explicit limit trims the view after sorting", async () => { + const h = makeHarness([ + { id: "1", rank: 1 }, + { id: "2", rank: 2 }, + ]); + const store = new LiveEntityStore(h.handler, { sort: "rank", limit: 2 }); + const stop = store.start(); + await tick(); + + h.rows.set("0", { id: "0", rank: 0 }); + h.ding({ table: "t", op: "INSERT", id: "0" }); + await settle(); + + expect(store.getSnapshot().data.map((r) => r.id)).toEqual(["0", "1"]); + stop(); + }); + + test("stop() detaches: dings after teardown do nothing", async () => { + const h = makeHarness([{ id: "1" }]); + const store = new LiveEntityStore(h.handler); + const stop = store.start(); + await tick(); + stop(); + + h.ding({ table: "t", op: "DELETE", id: "1" }); + await settle(); + expect(store.getSnapshot().data.length).toBe(1); // untouched + expect(h.calls.filter.length).toBe(0); + }); +}); diff --git a/src/live.ts b/src/live.ts new file mode 100644 index 0000000..4a69c9f --- /dev/null +++ b/src/live.ts @@ -0,0 +1,423 @@ +// The Bool live layer: a framework-agnostic sync engine over an entity table. +// +// This is the machinery `useEntity` (bool-sdk/react) sits on, and it exists +// because the hand-rolled version of this state machine — wire load(), refetch +// on every doorbell ping, layer optimistic writes — is exactly what generated +// app code kept getting subtly wrong: unguarded refetches raced each other +// (stale response lands last and rewinds the screen), a burst of pings caused +// a burst of full-list refetches, and a full-list replacement wiped optimistic +// rows whose writes were still in flight (the "items pop in and out" bug). +// Same reasoning as the auth React layer: own the state machine in one tested +// place; app code just renders. +// +// What it does, per change ding from the doorbell (see BoolChangePayload): +// - COALESCE (leading-edge): a lone ping reconciles immediately; followers +// within a short window batch into ONE pass, so a 50-row bulk write costs +// one round trip, not fifty — without taxing single changes. +// - DELTA-APPLY: `{op: DELETE, id}` removes locally with no fetch at all; +// a ding carrying the full `row` (the private doorbell's normal case) +// applies directly with ZERO fetches; only id-only dings (RLS-on tables +// without owner_id, legacy public pings) fetch the changed rows by id +// through the gateway (full auth + RLS) and upsert them. +// - ORDER: full reloads carry a monotonic sequence number; a stale response +// that lands after a newer one is dropped, never applied. +// - OPTIMISTIC LAYER: create/update/remove render immediately as an overlay +// ON TOP of committed server state — a concurrent refetch can't wipe an +// in-flight write, and a failed write rolls back by simply dropping its +// overlay. Committed rows and the overlay share the same client-generated +// id, so the doorbell echo of your own write reconciles to a no-op. +// +// Fallbacks are graceful: a ding with no id (an old-trigger fleet, or a table +// without an `id` column) degrades to one coalesced full reload — still +// strictly better than today's ping-per-refetch. +import type { BoolChangePayload } from "./client.js"; +import type { EntityHandler, FilterQuery, SortSpec } from "./entities.js"; + +/** Rows the live layer manages. Entity tables always have a string `id`. */ +export type EntityRow = { id: string } & Record; + +export type LiveQueryOptions = { + /** Same filter DSL as `bool.entities.
.filter(...)`. */ + filter?: FilterQuery; + /** `-col` descending, `col` ascending. Defaults to `-created_at`. */ + sort?: SortSpec; + /** Max rows to keep in view. When set, the view is trimmed after sorting. */ + limit?: number; +}; + +export type LiveSnapshot = { + data: T[]; + /** True until the first load settles (success or error). */ + loading: boolean; + /** The most recent load/mutation error, cleared by the next success. */ + error: unknown; +}; + +// How long to gather dings before reconciling. Long enough to swallow a bulk +// write's per-row pings, short enough to be imperceptible next to the network +// hop the reconcile itself costs. +const COALESCE_MS = 50; + +// A reconcile pass that would need to fetch more than this many distinct rows +// is cheaper as one full reload. +const MAX_KEYED_FETCH = 100; + +/** Postgres-flavored comparator for a SortSpec: ascending puts NULLs last, + * descending puts them first (matching the server's default order, so a + * delta-applied row lands where the next full load would put it). Ties break + * on id so the view is stable across reconciles. */ +export function compareBySort(sort: SortSpec = "-created_at") { + const desc = sort.startsWith("-"); + const col = sort.replace(/^[-+]/, ""); + return (a: T, b: T): number => { + const av = a[col] as string | number | null | undefined; + const bv = b[col] as string | number | null | undefined; + if (av != null || bv != null) { + if (av == null) return desc ? -1 : 1; + if (bv == null) return desc ? 1 : -1; + if (av < bv) return desc ? 1 : -1; + if (av > bv) return desc ? -1 : 1; + } + return a.id < b.id ? -1 : a.id > b.id ? 1 : 0; + }; +} + +/** Does one Mongo-style operator hold for a row value? Mirrors the SQL the + * server-side translation produces — including three-valued logic: a NULL + * column fails every comparison except an explicit null/exists check, exactly + * as PostgREST excludes those rows. */ +function operatorMatches(v: unknown, op: string, operand: unknown): boolean { + switch (op) { + case "$eq": + return operand === null ? v == null : v === operand; + case "$exists": + return operand ? v != null : v == null; + default: + break; + } + // SQL: NULL anything → NULL → row excluded. + if (v == null) return false; + switch (op) { + case "$ne": + return operand === null ? true : v !== operand; + case "$gt": + return (v as never) > (operand as never); + case "$gte": + return (v as never) >= (operand as never); + case "$lt": + return (v as never) < (operand as never); + case "$lte": + return (v as never) <= (operand as never); + case "$in": + return Array.isArray(operand) && operand.includes(v); + case "$nin": + return Array.isArray(operand) && !operand.includes(v); + case "$regex": + // POSIX ~ vs JS RegExp — identical for the common patterns app code + // writes; exotic POSIX classes may diverge until the next full load. + try { + return new RegExp(String(operand)).test(String(v)); + } catch { + return false; + } + case "$all": + return ( + Array.isArray(v) && Array.isArray(operand) && operand.every((x) => (v as unknown[]).includes(x)) + ); + case "$not": { + const [innerOp, innerVal] = Object.entries((operand ?? {}) as Record)[0] ?? []; + if (!innerOp) return true; + return !operatorMatches(v, innerOp, innerVal); + } + default: + return true; // unsupported operator — server ignores it too + } +} + +/** Client-side evaluation of the entities filter DSL, used to decide whether a + * changed row still belongs in a filtered live view without re-running the + * whole query. Mirrors the server translation in entities.ts; on any + * divergence the next full load is the source of truth. */ +export function matchesFilter(row: Record, query: FilterQuery): boolean { + for (const [column, value] of Object.entries(query)) { + if (value === undefined) continue; + if (column === "$and") { + if (!(value as FilterQuery[]).every((sub) => matchesFilter(row, sub))) return false; + } else if (column === "$or") { + if (!(value as FilterQuery[]).some((sub) => matchesFilter(row, sub))) return false; + } else if (column === "$nor") { + if ((value as FilterQuery[]).some((sub) => matchesFilter(row, sub))) return false; + } else { + const v = row[column]; + if (value === null) { + if (v != null) return false; + } else if (Array.isArray(value)) { + if (v == null || !(value as unknown[]).includes(v)) return false; + } else if (typeof value === "object") { + for (const [op, operand] of Object.entries(value as Record)) { + if (operand === undefined) continue; + if (!operatorMatches(v, op, operand)) return false; + } + } else if (v !== value) { + return false; + } + } + } + return true; +} + +type PendingOp = + | { kind: "create"; id: string; row: Partial } + | { kind: "update"; id: string; patch: Partial } + | { kind: "remove"; id: string }; + +/** Generate a client-side row id (the optimistic-UI cornerstone: ONE id shared + * by the optimistic row, the insert, and the doorbell echo). */ +function newRowId(): string { + if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") { + return crypto.randomUUID(); + } + // Non-crypto fallback for exotic runtimes; collision odds are irrelevant at + // per-user-session row counts. + return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 10)}`; +} + +export class LiveEntityStore { + /** Committed (server-confirmed) rows by id. */ + private server = new Map(); + /** Optimistic overlay, in creation order, applied over `server`. */ + private pending: PendingOp[] = []; + private listeners = new Set<() => void>(); + private snapshot: LiveSnapshot = { data: [], loading: true, error: null }; + private queue: BoolChangePayload[] = []; + private flushTimer: ReturnType | null = null; + /** When the last reconcile pass ran — drives leading-edge coalescing. */ + private lastFlushAt = -Infinity; + /** Monotonic full-load counter — the anti-rewind guard. Only the response to + * the NEWEST load may apply; anything else is a stale answer arriving late. */ + private loadSeq = 0; + private unsubscribe: (() => void) | null = null; + + constructor( + private handler: EntityHandler, + private opts: LiveQueryOptions = {}, + ) {} + + /** Subscribe to the doorbell and run the initial load. Returns a stop + * function; the store is restartable (React StrictMode mounts twice). */ + start(): () => void { + this.unsubscribe?.(); + this.unsubscribe = this.handler.subscribe((p) => this.onDing(p)); + void this.load(); + return () => { + this.unsubscribe?.(); + this.unsubscribe = null; + if (this.flushTimer !== null) { + clearTimeout(this.flushTimer); + this.flushTimer = null; + } + this.queue = []; + }; + } + + /** Register a snapshot listener (useSyncExternalStore-shaped). */ + onSnapshot(listener: () => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + /** The current immutable snapshot. Reference-stable between changes. */ + getSnapshot(): LiveSnapshot { + return this.snapshot; + } + + /** Force a full reload (sequenced like any other). */ + async refetch(): Promise { + await this.load(); + } + + // ---- optimistic mutations ------------------------------------------------- + // Each renders instantly via the overlay, settles into `server` on success, + // and rolls back by dropping its overlay on failure (error surfaced on the + // snapshot, never thrown — the view must keep rendering). + + async create(fields: Partial): Promise { + const id = typeof fields.id === "string" ? fields.id : newRowId(); + const op: PendingOp = { kind: "create", id, row: { ...fields, id } }; + this.pending.push(op); + this.emit(); + try { + const created = await this.handler.create({ ...fields, id } as Partial); + this.server.set(id, created); + this.snapshot = { ...this.snapshot, error: null }; + return created; + } catch (error) { + this.snapshot = { ...this.snapshot, error }; + return null; + } finally { + this.pending = this.pending.filter((p) => p !== op); + this.emit(); + } + } + + async update(id: string, fields: Partial): Promise { + const op: PendingOp = { kind: "update", id, patch: fields }; + this.pending.push(op); + this.emit(); + try { + const updated = await this.handler.update(id, fields); + this.server.set(id, updated); + this.snapshot = { ...this.snapshot, error: null }; + return updated; + } catch (error) { + this.snapshot = { ...this.snapshot, error }; + return null; + } finally { + this.pending = this.pending.filter((p) => p !== op); + this.emit(); + } + } + + async remove(id: string): Promise { + const op: PendingOp = { kind: "remove", id }; + this.pending.push(op); + this.emit(); + try { + await this.handler.delete(id); + this.server.delete(id); + this.snapshot = { ...this.snapshot, error: null }; + return true; + } catch (error) { + this.snapshot = { ...this.snapshot, error }; + return false; + } finally { + this.pending = this.pending.filter((p) => p !== op); + this.emit(); + } + } + + // ---- loading & reconciling ------------------------------------------------ + + private async load(): Promise { + const seq = ++this.loadSeq; + try { + const rows = this.opts.filter + ? await this.handler.filter(this.opts.filter, this.opts.sort, this.opts.limit) + : await this.handler.list(this.opts.sort, this.opts.limit); + if (seq !== this.loadSeq) return; // a newer load superseded this one + this.server = new Map(rows.map((r) => [r.id, r])); + this.snapshot = { ...this.snapshot, loading: false, error: null }; + this.emit(); + } catch (error) { + if (seq !== this.loadSeq) return; + this.snapshot = { ...this.snapshot, loading: false, error }; + this.emit(); + } + } + + private onDing(p: BoolChangePayload): void { + this.queue.push(p); + if (this.flushTimer === null) { + // LEADING-EDGE coalescing: a lone change (the common case) reconciles on + // the next macrotask — no felt delay — while a burst still collapses: + // once a pass has run, followers within the window batch into one more + // pass. A pure trailing window taxed every single change 50ms just in + // case a burst was coming. + const sinceLast = Date.now() - this.lastFlushAt; + const delay = sinceLast >= COALESCE_MS ? 0 : COALESCE_MS - sinceLast; + this.flushTimer = setTimeout(() => { + this.flushTimer = null; + void this.flush(); + }, delay); + } + } + + private async flush(): Promise { + this.lastFlushAt = Date.now(); + const batch = this.queue.splice(0); + if (batch.length === 0) return; + + // Collapse the batch to one final op per id (a row inserted then deleted + // within the window nets out to its LAST op). Any ding without an id means + // we can't reconcile precisely — degrade the whole pass to one full load. + const finalOps = new Map(); + for (const p of batch) { + if (!p.id) return void this.load(); + finalOps.set(p.id, p); + } + + let changed = false; + const toFetch: string[] = []; + for (const [id, p] of finalOps) { + if (p.op === "DELETE") { + if (this.server.delete(id)) changed = true; + } else if (p.row) { + // Row rode the ding (private-channel doorbell) — apply directly. + changed = this.applyRow(id, p.row as T) || changed; + } else { + toFetch.push(id); + } + } + if (changed) this.emit(); + + if (toFetch.length === 0) return; + if (toFetch.length > MAX_KEYED_FETCH) return void this.load(); + + try { + const rows = await this.handler.filter( + { id: toFetch } as FilterQuery, + this.opts.sort, + toFetch.length, + ); + const returned = new Set(rows.map((r) => r.id)); + let applied = false; + for (const row of rows) applied = this.applyRow(row.id, row) || applied; + // Requested but absent: deleted meanwhile, or not visible to this viewer + // (RLS) — either way it doesn't belong in the view. + for (const id of toFetch) { + if (!returned.has(id) && this.server.delete(id)) applied = true; + } + if (applied) this.emit(); + } catch { + // Keyed reconcile failed (offline blip, gateway hiccup) — fall back to a + // sequenced full load rather than leaving the view stale. + void this.load(); + } + } + + /** Upsert one committed row, honoring the view filter: a row that no longer + * matches drops out, one that now matches drops in. Returns whether the + * committed state changed. */ + private applyRow(id: string, row: T): boolean { + const belongs = !this.opts.filter || matchesFilter(row, this.opts.filter); + if (!belongs) return this.server.delete(id); + this.server.set(id, row); + return true; + } + + // ---- view computation ----------------------------------------------------- + + private emit(): void { + const byId = new Map(this.server); + // Overlay optimistic ops in creation order. Creates and updates stay + // visible even if they wouldn't match the view filter — hiding the row the + // user JUST touched reads as data loss; the next committed reconcile + // settles membership. + for (const op of this.pending) { + if (op.kind === "create") { + byId.set(op.id, { ...(byId.get(op.id) ?? {}), ...op.row } as T); + } else if (op.kind === "update") { + const base = byId.get(op.id); + if (base) byId.set(op.id, { ...base, ...op.patch }); + } else { + byId.delete(op.id); + } + } + let data = [...byId.values()].sort(compareBySort(this.opts.sort)); + if (this.opts.limit !== undefined && data.length > this.opts.limit) { + data = data.slice(0, this.opts.limit); + } + this.snapshot = { ...this.snapshot, data }; + for (const l of this.listeners) l(); + } +} diff --git a/src/react.test.tsx b/src/react.test.tsx index 9973bec..265b0c8 100644 --- a/src/react.test.tsx +++ b/src/react.test.tsx @@ -1,7 +1,13 @@ import { beforeEach, describe, expect, test } from "bun:test"; import { renderToString } from "react-dom/server"; import { createBoolClient } from "./client"; -import { AuthGate, BoolAuthProvider, takeResetTokenFromSearch, useBoolAuth } from "./react"; +import { + AuthGate, + BoolAuthProvider, + takeResetTokenFromSearch, + useBoolAuth, + useEntity, +} from "./react"; // SSR smoke tests: effects don't run in renderToString, so the provider is in // its initial loading state — enough to pin the gate/hook contract without a @@ -97,3 +103,32 @@ describe("takeResetTokenFromSearch", () => { expect(takeResetTokenFromSearch("")).toEqual({ token: null, rest: "" }); }); }); + +// SSR smoke for useEntity: effects don't run in renderToString, so this pins +// the initial contract (loading, empty data, callable shape) without a +// browser. The live state machine itself is covered headlessly in live.test.ts. +describe("useEntity", () => { + test("renders the initial loading snapshot on the server", () => { + createBoolClient(CONFIG); + function List() { + const todos = useEntity("todos"); + return ( +
+ {todos.loading ? "loading" : "ready"}:{todos.data.length} +
+ ); + } + const html = renderToString(); + expect(html).toContain("loading"); + expect(html).toContain("0"); + }); + + test("accepts an explicit client and returns mutation handles", () => { + const client = createBoolClient(CONFIG); + function Probe() { + const t = useEntity("todos", { client, sort: "-created_at", limit: 5 }); + return {typeof t.create === "function" ? "ok" : "bad"}; + } + expect(renderToString()).toContain("ok"); + }); +}); diff --git a/src/react.tsx b/src/react.tsx index 6bbd2bc..3583173 100644 --- a/src/react.tsx +++ b/src/react.tsx @@ -12,11 +12,19 @@ import { createContext, useContext, useEffect, + useRef, useState, + useSyncExternalStore, type FormEvent, type ReactNode, } from "react"; import { getDefaultBoolClient, type BoolClient, type BoolUser } from "./client.js"; +import { + LiveEntityStore, + type EntityRow, + type LiveQueryOptions, +} from "./live.js"; +import type { EntityHandler } from "./entities.js"; type AuthActionResult = { error: unknown }; @@ -250,3 +258,94 @@ export function useSignInForm() { signInWithGoogle: startGoogleSignIn, }; } + +// --------------------------------------------------------------------------- +// useEntity — a live view of one entity table. +// +// The data-sync state machine (initial load, doorbell subscription, coalesced +// delta reconciling, stale-response ordering, optimistic create/update/remove +// with rollback) is exactly what generated app code kept getting wrong when it +// hand-wired load() + subscribe(() => load()) — races made items pop in and +// out. Same philosophy as the auth layer above: the SDK owns the machine +// (LiveEntityStore in live.ts, where it's unit-tested headlessly); app code +// just renders. +// +// const todos = useEntity("todos", { sort: "-created_at" }); +// todos.data // live rows — updates on every change +// todos.create({ title: "hi" }) // appears instantly, rolls back on error +// todos.update(id, { done: true }) +// todos.remove(id) +// --------------------------------------------------------------------------- + +export type UseEntityResult = { + /** Live rows: committed server state with in-flight optimistic writes + * layered on top, filtered/sorted/limited per the hook options. */ + data: T[]; + /** True until the first load settles. */ + loading: boolean; + /** Latest load/mutation error (cleared by the next success). Mutations never + * throw — check this (or a mutation's return value) to surface failures. */ + error: unknown; + /** Optimistic insert. Resolves to the committed row, or null on failure + * (already rolled back). Pass an `id` to control it; one is generated + * otherwise. */ + create: (fields: Partial) => Promise; + /** Optimistic patch. Resolves to the committed row, or null on failure. */ + update: (id: string, fields: Partial) => Promise; + /** Optimistic delete. Resolves false on failure (row restored). */ + remove: (id: string) => Promise; + /** Force a full reload (rarely needed — changes arrive on their own). */ + refetch: () => Promise; +}; + +export function useEntity( + table: string, + opts?: LiveQueryOptions & { + /** Defaults to the client created by createBoolClient() — in a Bool app + * you never pass this. */ + client?: BoolClient; + }, +): UseEntityResult { + const bool = opts?.client ?? getDefaultBoolClient(); + // A new table/filter/sort/limit is a different query → fresh store. JSON + // keying means an inline `{ filter: {...} }` object literal is fine (no + // useMemo required of app code). + const key = + table + + " " + + bool.schema + + " " + + JSON.stringify([opts?.filter ?? null, opts?.sort ?? null, opts?.limit ?? null]); + const ref = useRef<{ key: string; store: LiveEntityStore } | null>(null); + if (ref.current === null || ref.current.key !== key) { + ref.current = { + key, + store: new LiveEntityStore(bool.entities[table] as unknown as EntityHandler, { + filter: opts?.filter, + sort: opts?.sort, + limit: opts?.limit, + }), + }; + } + const store = ref.current.store; + + // Subscribe + initial load, torn down on unmount or query change. The store + // is restartable, so StrictMode's mount-unmount-mount is safe. + useEffect(() => store.start(), [store]); + + const snap = useSyncExternalStore( + (cb) => store.onSnapshot(cb), + () => store.getSnapshot(), + () => store.getSnapshot(), + ); + + return { + data: snap.data, + loading: snap.loading, + error: snap.error, + create: (fields) => store.create(fields), + update: (id, fields) => store.update(id, fields), + remove: (id) => store.remove(id), + refetch: () => store.refetch(), + }; +} diff --git a/src/realtime.test.ts b/src/realtime.test.ts new file mode 100644 index 0000000..0b9324d --- /dev/null +++ b/src/realtime.test.ts @@ -0,0 +1,223 @@ +import { describe, expect, test } from "bun:test"; +import { createDoorbell, type DoorbellDeps, type RealtimeMint } from "./realtime"; +import type { BoolChangePayload } from "./client"; + +// The doorbell lifecycle drives everything through injected deps, so these +// tests exercise the full machine — mint, join, refresh, revocation, fallback, +// teardown — with zero sockets. + +type FakeChannel = { + topic: string; + priv: boolean; + joined: boolean; + left: boolean; + cb: ((p: BoolChangePayload) => void) | null; + status: ((s: string) => void) | null; +}; + +function makeHarness(opts: { mints?: (RealtimeMint | null)[] } = {}) { + const mints = [...(opts.mints ?? [])]; + const h = { + channels: [] as FakeChannel[], + authed: [] as string[], + mintCalls: 0, + scheduled: [] as { fn: () => void; ms: number; cancelled: boolean }[], + /** Run the next pending refresh timer. */ + async fire() { + const t = h.scheduled.find((s) => !s.cancelled && !(s as any).fired); + if (!t) throw new Error("nothing scheduled"); + (t as any).fired = true; + await t.fn(); + await tick(); + }, + ding(topic: string, payload: BoolChangePayload) { + for (const ch of h.channels) { + if (ch.topic === topic && ch.joined && !ch.left) ch.cb?.(payload); + } + }, + live: () => h.channels.filter((c) => c.joined && !c.left), + }; + const deps: DoorbellDeps = { + async mint() { + h.mintCalls++; + return mints.length ? mints.shift()! : null; + }, + async setAuth(token) { + h.authed.push(token); + }, + channel(topic, { private: priv }) { + const ch: FakeChannel = { topic, priv, joined: false, left: false, cb: null, status: null }; + h.channels.push(ch); + return { + onBroadcast(cb) { + ch.cb = cb; + }, + join(status) { + ch.joined = true; + ch.status = status; + }, + leave() { + ch.left = true; + }, + }; + }, + legacyTopic: "bool:app_x", + schedule(fn, ms) { + const t = { fn: fn as () => void, ms, cancelled: false }; + h.scheduled.push(t); + return t; + }, + cancel(handle) { + (handle as { cancelled: boolean }).cancelled = true; + }, + }; + return { h, doorbell: createDoorbell(deps) }; +} + +const MINT: RealtimeMint = { + token: "tok-1", + expiresIn: 900, + topics: { app: "bool:app_x:app", user: "bool:app_x:user:u1" }, +}; +const tick = () => new Promise((r) => setTimeout(r, 0)); + +describe("createDoorbell: the private path", () => { + test("mints, presents the wristband, joins app + user rooms privately", async () => { + const { h, doorbell } = makeHarness({ mints: [MINT] }); + doorbell.subscribe(() => {}); + await tick(); + expect(h.authed).toEqual(["tok-1"]); + expect(h.live().map((c) => [c.topic, c.priv])).toEqual([ + ["bool:app_x:app", true], + ["bool:app_x:user:u1", true], + ]); + }); + + test("anonymous mint (no user topic) joins the app room only", async () => { + const { h, doorbell } = makeHarness({ + mints: [{ ...MINT, topics: { app: "bool:app_x:app", user: null } }], + }); + doorbell.subscribe(() => {}); + await tick(); + expect(h.live().map((c) => c.topic)).toEqual(["bool:app_x:app"]); + }); + + test("payloads from either room reach every listener (one shared doorbell)", async () => { + const { h, doorbell } = makeHarness({ mints: [MINT] }); + const a: BoolChangePayload[] = []; + const b: BoolChangePayload[] = []; + doorbell.subscribe((p) => a.push(p)); + doorbell.subscribe((p) => b.push(p)); + await tick(); + expect(h.mintCalls).toBe(1); // second subscriber reuses the running doorbell + h.ding("bool:app_x:app", { table: "todos", op: "INSERT", id: "1", row: { id: "1" } }); + h.ding("bool:app_x:user:u1", { table: "notes", op: "UPDATE", id: "2" }); + expect(a).toHaveLength(2); + expect(b).toHaveLength(2); + expect(a[0]!.row).toEqual({ id: "1" }); + }); +}); + +describe("createDoorbell: refresh & revocation", () => { + test("re-mints at ~75% of the TTL and re-presents the new wristband", async () => { + const { h, doorbell } = makeHarness({ + mints: [MINT, { ...MINT, token: "tok-2" }], + }); + doorbell.subscribe(() => {}); + await tick(); + expect(h.scheduled[0]!.ms).toBe(900 * 1000 * 0.75); + await h.fire(); + expect(h.authed).toEqual(["tok-1", "tok-2"]); + // channels stay up — setAuth re-auths the connection, no rejoin + expect(h.live().length).toBe(2); + // and the next refresh is scheduled + expect(h.scheduled.length).toBe(2); + }); + + test("a refused re-mint (revoked access) silences the private rooms and falls back", async () => { + const { h, doorbell } = makeHarness({ mints: [MINT] }); // second mint → null + const got: BoolChangePayload[] = []; + doorbell.subscribe((p) => got.push(p)); + await tick(); + await h.fire(); + // private rooms left; only the public legacy channel remains + const live = h.live(); + expect(live.map((c) => [c.topic, c.priv])).toEqual([["bool:app_x", false]]); + // late payloads on the torn-down private room reach nobody + h.ding("bool:app_x:app", { table: "todos", op: "INSERT", id: "9" }); + expect(got).toHaveLength(0); + }); + + test("a nonsense TTL is clamped so refresh can't melt into a mint loop", async () => { + const { h, doorbell } = makeHarness({ mints: [{ ...MINT, expiresIn: 1 }] }); + doorbell.subscribe(() => {}); + await tick(); + expect(h.scheduled[0]!.ms).toBe(30_000); + }); +}); + +describe("createDoorbell: fallback & teardown", () => { + test("no wristband desk (mint null) → legacy public channel", async () => { + const { h, doorbell } = makeHarness({ mints: [null] }); + const got: BoolChangePayload[] = []; + doorbell.subscribe((p) => got.push(p)); + await tick(); + expect(h.authed).toHaveLength(0); + expect(h.live().map((c) => [c.topic, c.priv])).toEqual([["bool:app_x", false]]); + h.ding("bool:app_x", { table: "todos", op: "DELETE", id: "3" }); + expect(got).toHaveLength(1); + }); + + test("a refused private join degrades to the public ping (never a dead app)", async () => { + const { h, doorbell } = makeHarness({ mints: [MINT] }); + doorbell.subscribe(() => {}); + await tick(); + h.channels[0]!.status!("CHANNEL_ERROR"); + expect(h.live().map((c) => [c.topic, c.priv])).toEqual([["bool:app_x", false]]); + }); + + test("last unsubscribe tears everything down; a late mint resolution no-ops", async () => { + let release!: (m: RealtimeMint) => void; + const slowMint = new Promise((r) => (release = r)); + const h = { + channels: [] as FakeChannel[], + authed: [] as string[], + }; + const doorbell = createDoorbell({ + mint: () => slowMint as Promise, + setAuth: async (t) => void h.authed.push(t), + channel(topic, { private: priv }) { + const ch: FakeChannel = { topic, priv, joined: false, left: false, cb: null, status: null }; + h.channels.push(ch); + return { + onBroadcast() {}, + join() { + ch.joined = true; + }, + leave() { + ch.left = true; + }, + }; + }, + legacyTopic: "bool:app_x", + }); + const off = doorbell.subscribe(() => {}); + off(); // teardown while the mint is still in flight + release(MINT); + await tick(); + expect(h.authed).toHaveLength(0); // orphaned continuation did nothing + expect(h.channels).toHaveLength(0); + }); + + test("resubscribing after teardown starts a fresh doorbell", async () => { + const { h, doorbell } = makeHarness({ mints: [MINT, { ...MINT, token: "tok-2" }] }); + const off = doorbell.subscribe(() => {}); + await tick(); + off(); + expect(h.live()).toHaveLength(0); + doorbell.subscribe(() => {}); + await tick(); + expect(h.authed).toEqual(["tok-1", "tok-2"]); + expect(h.live()).toHaveLength(2); + }); +}); diff --git a/src/realtime.ts b/src/realtime.ts new file mode 100644 index 0000000..4a852ca --- /dev/null +++ b/src/realtime.ts @@ -0,0 +1,169 @@ +// The private doorbell client — the SDK half of Bool's "intercom + wristband" +// realtime design (platform: lib/bool-db.ts GATEWAY_GLOBAL_SETUP_SQL + the +// /_bool/v1/realtime/token plane). +// +// Live updates arrive as ROW-BEARING broadcasts on PRIVATE topics that only a +// gateway-minted token ("wristband") can join. This module owns the whole +// lifecycle so app code (and useEntity) just gets payloads: +// - mint the wristband from the gateway (which re-runs liveAccess/session on +// every mint — that's where access is actually decided), +// - setAuth + join the app room and, when signed in, the personal user room +// (the trigger picks disjoint audiences, so the two rooms never duplicate), +// - re-mint at ~75% of the TTL so the socket never hits token expiry; a +// failed re-mint (revoked access, gateway down) silences the private rooms +// rather than erroring the app, +// - fall back to the legacy PUBLIC row-data-free channel when the wristband +// desk is unavailable (older platform, 503) so live-ness degrades to +// ping+refetch instead of dying, +// - share ONE doorbell across every subscriber (each entities.
+// .subscribe used to open its own channel; now they ref-count this one). +// +// Everything effectful is injected (DoorbellDeps) so the machine is fully +// unit-testable without sockets — same dependency-injection discipline as the +// platform's plane handlers. + +import type { BoolChangePayload } from "./client.js"; + +/** What the wristband desk returns. Topic names are SERVER-authored so naming + * lives in exactly one repo. */ +export type RealtimeMint = { + token: string; + /** Seconds until the token expires. */ + expiresIn: number; + topics: { app: string; user: string | null }; +}; + +export type DoorbellChannel = { + /** Register the broadcast payload handler (must be called before join). */ + onBroadcast(cb: (payload: BoolChangePayload) => void): void; + /** Open the channel; `status` fires with supabase-style states. */ + join(status: (state: string) => void): void; + leave(): void; +}; + +export type DoorbellDeps = { + /** Mint one wristband; null when the desk is unavailable (non-2xx, network, + * or a pre-private-doorbell platform). */ + mint(): Promise; + /** Present the wristband to the realtime connection (supabase setAuth). */ + setAuth(token: string): Promise | void; + /** Create (not join) a channel on `topic`. */ + channel(topic: string, opts: { private: boolean }): DoorbellChannel; + /** The legacy public topic (`bool:`) for the fallback path. */ + legacyTopic: string; + /** Test seam; defaults to setTimeout/clearTimeout. */ + schedule?: (fn: () => void, ms: number) => unknown; + cancel?: (handle: unknown) => void; +}; + +// Re-mint at 75% of the TTL: early enough that a slow mint never races token +// expiry (which would close the channels underneath us), late enough that +// refresh traffic stays negligible. +const REFRESH_FRACTION = 0.75; +// A desk that answers but with a nonsense TTL shouldn't melt into a mint loop. +const MIN_REFRESH_MS = 30_000; + +export type Doorbell = { + /** Register a change listener. Starts the machinery on the first listener, + * tears it down after the last unsubscribes. Returns unsubscribe. */ + subscribe(listener: (payload: BoolChangePayload) => void): () => void; +}; + +export function createDoorbell(deps: DoorbellDeps): Doorbell { + const schedule = deps.schedule ?? ((fn, ms) => setTimeout(fn, ms)); + const cancel = deps.cancel ?? ((h) => clearTimeout(h as ReturnType)); + + const listeners = new Set<(p: BoolChangePayload) => void>(); + const fanout = (p: BoolChangePayload) => { + for (const l of listeners) l(p); + }; + + // One "generation" per start(); every async continuation checks it so a + // teardown (or a restart) orphans in-flight work instead of racing it. + let gen = 0; + let channels: DoorbellChannel[] = []; + let refreshHandle: unknown = null; + + function teardownChannels(): void { + for (const ch of channels) ch.leave(); + channels = []; + if (refreshHandle !== null) { + cancel(refreshHandle); + refreshHandle = null; + } + } + + function joinLegacy(myGen: number): void { + if (myGen !== gen) return; + const ch = deps.channel(deps.legacyTopic, { private: false }); + ch.onBroadcast(fanout); + ch.join(() => {}); + channels.push(ch); + } + + async function startPrivate(myGen: number): Promise { + const m = await deps.mint(); + if (myGen !== gen) return; + if (!m) return joinLegacy(myGen); + + await deps.setAuth(m.token); + if (myGen !== gen) return; + + // The app room, and the personal room when the mint carried a signed-in + // user. Audiences are disjoint by construction (the trigger rings an + // owned row's user room INSTEAD of the app room), so no dedupe is needed. + for (const topic of [m.topics.app, m.topics.user]) { + if (!topic) continue; + const ch = deps.channel(topic, { private: true }); + ch.onBroadcast(fanout); + ch.join((state) => { + // A refused/failed private join (policy missing on an older database, + // clock-skewed token) degrades to the public ping — worse latency, + // never a dead app. Guard on gen so late errors after teardown no-op. + if (state === "CHANNEL_ERROR" && myGen === gen) { + teardownChannels(); + joinLegacy(myGen); + } + }); + channels.push(ch); + } + + scheduleRefresh(myGen, m.expiresIn); + } + + function scheduleRefresh(myGen: number, expiresInSeconds: number): void { + const ms = Math.max(expiresInSeconds * 1000 * REFRESH_FRACTION, MIN_REFRESH_MS); + refreshHandle = schedule(async () => { + if (myGen !== gen) return; + const m = await deps.mint(); + if (myGen !== gen) return; + if (!m) { + // Wristband renewal refused: access was revoked or the gateway is + // unreachable. Go quiet on the private rooms (the whole point of the + // TTL) and keep the app alive on the public ping. + teardownChannels(); + return joinLegacy(myGen); + } + await deps.setAuth(m.token); // existing channels re-auth on the connection + if (myGen !== gen) return; + scheduleRefresh(myGen, m.expiresIn); + }, ms); + } + + return { + subscribe(listener) { + listeners.add(listener); + if (listeners.size === 1) { + gen++; + void startPrivate(gen); + } + return () => { + if (!listeners.delete(listener)) return; + if (listeners.size === 0) { + gen++; // orphan any in-flight start/refresh + teardownChannels(); + } + }; + }, + }; +}