diff --git a/package.json b/package.json index 6219369..aca6b98 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "bool-sdk", - "version": "0.3.3", - "description": "Client SDK for apps built on Bool \u2014 gateway data access, end-user auth, the AI battery, the React auth layer, and the local-dev CLI (link, types, deploy).", + "version": "0.4.0-next.7", + "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", "types": "./dist/index.d.ts", @@ -18,7 +18,7 @@ "default": "./dist/react.js" } }, - "_sideEffectsNote": "dist/react.js MUST stay listed below. It registers the live-query implementation into the React-free core at module scope, and apps load it via a bare `import \"bool-sdk/react\"` with no bindings used \u2014 so a blanket `\"sideEffects\": false` licenses bundlers to delete that import outright. Verified: with `false`, a real Vite production build drops it and every published app throws on first render (shipped as 0.3.0, fixed in 0.3.1).", + "_sideEffectsNote": "dist/react.js MUST stay listed below. It registers the live-query implementation into the React-free core at module scope, and apps load it via a bare `import \"bool-sdk/react\"` with no bindings used — so a blanket `\"sideEffects\": false` licenses bundlers to delete that import outright. Verified: with `false`, a real Vite production build drops it and every published app throws on first render (shipped as 0.3.0, fixed in 0.3.1).", "sideEffects": [ "./dist/react.js" ], diff --git a/src/client.ts b/src/client.ts index 96b3fe9..4366168 100644 --- a/src/client.ts +++ b/src/client.ts @@ -22,6 +22,14 @@ import { createClient, type SupabaseClient } from "@supabase/supabase-js"; import { createEntitiesModule, type EntitiesModule } from "./entities.js"; import { createDoorbell, type MintResult, type RealtimeMint } from "./realtime.js"; +import { + createBoolRoom, + createBoolRoomApi, + createRoomStore, + type BoolRoomApi, + type RoomStore, +} from "./room.js"; +import { createSession, type SessionChannel } from "./session.js"; /** Matches the server's append-only gateway path version. */ const GATEWAY_API = "v1"; @@ -197,6 +205,11 @@ export type BoolClient = { /** The AI battery: `ai.generate(prompt)` / `ai.generate({prompt, schema})` / * `ai.stream(prompt)`. Server-side AI with no API key in the bundle. */ ai: BoolAi; + /** The ephemeral lane: live cursors, presence, one-shot events — data that + * flies between the people currently in the app and is never stored. + * `bool.room.useOthers()` is the app-wide room; `bool.room("game:4")` is a + * scoped one with the identical surface. Durable data stays on `entities`. */ + room: BoolRoomApi; /** This app's private Postgres schema name. */ schema: string; /** Subscribe to the app's realtime "doorbell": fires whenever any row in the @@ -330,6 +343,13 @@ export function createBoolClient(config: BoolClientConfig): BoolClient { // Cast: Bun's `typeof fetch` demands a `preconnect` property that a plain // fetch-shaped function doesn't have; supabase-js only ever calls it. global: { fetch: proxyFetch as unknown as typeof fetch }, + // supabase-js defaults to 10 client events/sec, which a bool.room cursor + // stream (~40/s after the SDK's own throttle) would trip — the limiter + // silently drops the excess and cursors freeze. The room store is the only + // high-frequency sender and it throttles itself; this cap is the backstop. + // Declared per-connection send budget. 60Hz room state + events + margin; + // the server-side ceiling is the tenant config, not this. + realtime: { params: { eventsPerSecond: 500 } }, }); const authListeners = new Set(); @@ -666,20 +686,40 @@ export function createBoolClient(config: BoolClientConfig): BoolClient { } }; - const doorbell = createDoorbell({ - mint: mintRealtimeToken, + // ONE session for everything realtime — the doorbell and every room share + // its mint, its wristband refresh, and the underlying socket. Two machines + // used to run here with duplicate lifecycles; both of the July-2026 + // production fires had to be fixed twice because of it (see session.ts). + const session = createSession({ + mint: async () => { + const res = await mintRealtimeToken(); + if (!res.ok) return res; + return { + ok: true as const, + token: res.mint.token, + expiresIn: res.mint.expiresIn, + topics: { + app: res.mint.topics.app, + user: res.mint.topics.user ?? null, + room: res.mint.topics.room ?? null, + }, + }; + }, 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 } }); + }); + + const doorbell = createDoorbell({ + session, + makeChannel: (topic) => { + // The channel is created lazily-configured and joined by the session, + // so onBroadcast registration always precedes the subscribe. + const ch = db.channel(topic, { config: { private: true } }); return { onBroadcast(cb) { ch.on("broadcast", { event: "*" }, (msg) => - cb((msg as { payload?: BoolChangePayload }).payload ?? {}), + cb(msg as unknown as { event: string; payload: unknown }), ); }, join(status) { @@ -696,11 +736,63 @@ export function createBoolClient(config: BoolClientConfig): BoolClient { listener: (payload: BoolChangePayload) => void, ): (() => void) => doorbell.subscribe(listener); + const makeRoomChannel = (topic: string, key: string): SessionChannel => { + const ch = db.channel(topic, { + config: { + private: true, + presence: { key }, + // No server echo: the store already delivered the sender's copy + // locally and synchronously; a wire echo would double-fire. + broadcast: { self: false, ack: false }, + }, + }); + return { + track(state, done) { + // supabase-js RESOLVES with "ok" | "timed out" | "error" — it does not + // throw — so the result must be inspected or every failure is silent. + void ch.track(state).then( + (r) => done?.(typeof r === "string" ? r : "ok"), + () => done?.("error"), + ); + }, + onPresence(cb) { + ch.on("presence", { event: "sync" }, () => { + cb(ch.presenceState() as Record>>); + }); + }, + onBroadcast(cb) { + ch.on("broadcast", { event: "*" }, (msg) => + cb(msg as unknown as { event: string; payload: unknown }), + ); + }, + send(event, payload, done) { + void ch.send({ type: "broadcast", event, payload }).then( + (r) => done?.(typeof r === "string" ? r : "ok"), + () => done?.("error"), + ); + }, + join(status) { + ch.subscribe((state) => status(state)); + }, + leave() { + void db.removeChannel(ch); + }, + }; + }; + + // The default room + memoized scoped rooms, all channel groups on the one + // session: `bool.room.useOthers()` is the app-wide room, `bool.room("id")` + // is a scoped one with the identical surface. + const roomStoreFor = (roomId: string | null): RoomStore> => + createRoomStore({ session, roomId, makeChannel: makeRoomChannel }); + const room = createBoolRoomApi(createBoolRoom(roomStoreFor(null)), (id) => roomStoreFor(id)); + const client: BoolClient = { db, entities: createEntitiesModule(db, subscribeToChanges), auth, ai, + room, schema, subscribeToChanges, }; diff --git a/src/index.ts b/src/index.ts index bc06389..b698bc6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -37,3 +37,10 @@ export { type LiveQueryOptions, type LiveSnapshot, } from "./live.js"; +export { + colorForId, + type BoolRoom, + type RoomEvent, + type RoomPeer, + type RoomStatus, +} from "./room.js"; diff --git a/src/react.test.tsx b/src/react.test.tsx index 869bc2a..a2664b5 100644 --- a/src/react.test.tsx +++ b/src/react.test.tsx @@ -9,6 +9,7 @@ import { useEntity, } from "./react"; import { __registerEntityUseQuery } from "./entities"; +import { __registerRoomHooks } from "./room"; // 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 @@ -193,3 +194,51 @@ describe("bool.entities..useQuery", () => { expect(typeof Probe).toBe("function"); }); }); + +// bool.room hooks — same contract pins as useQuery: SSR-renderable initial +// state, and a clear thrown error when the React entry isn't loaded. +describe("bool.room hooks", () => { + test("useOthers/useStatus/useSetMe render on the server with honest initials", () => { + const client = createBoolClient(CONFIG); + function Probe() { + const others = client.room.useOthers<{ cursor: { x: number } }>(); + const setMe = client.room.useSetMe(); + const status = client.room.useStatus(); + return ( + + {others.length}:{status}:{typeof setMe === "function" ? "fn" : "bad"} + + ); + } + // SSR interleaves comment markers between expressions, so assert pieces. + const html = renderToString(); + expect(html).toContain("connecting"); + expect(html).toContain("fn"); + expect(html).not.toContain("bad"); + }); + + test("self is available without React and colors are deterministic", () => { + const client = createBoolClient(CONFIG); + expect(client.room.self.id.length).toBeGreaterThan(6); + expect(client.room.self.color).toMatch(/^hsl\(/); + }); + + test("without the React entry loaded, hooks throw instructions naming the fix", () => { + const prev = __registerRoomHooks(null); + try { + const client = createBoolClient(CONFIG); + expect(() => client.room.useOthers()).toThrow(/bool-sdk\/react/); + expect(() => client.room.useSetMe()).toThrow(/React entry/); + } finally { + __registerRoomHooks(prev); + } + }); + + test("hallucinated members are type errors, not runtime undefined", () => { + const client = createBoolClient(CONFIG); + // @ts-expect-error — no useMyPresence: BoolRoom has NO index signature, so + // Liveblocks-reflex members fail the build instead of failing users. + const bad = client.room.useMyPresence; + expect(bad).toBeUndefined(); + }); +}); diff --git a/src/react.tsx b/src/react.tsx index a5852c7..c90b832 100644 --- a/src/react.tsx +++ b/src/react.tsx @@ -30,6 +30,13 @@ import { type EntityHandler, type EntityQueryResult, } from "./entities.js"; +import { + __registerRoomHooks, + type RoomEvent, + type RoomPeer, + type RoomStatus, + type RoomStore, +} from "./room.js"; // Re-exported on purpose, and it is load-bearing. A React app creates its client // by importing createBoolClient FROM HERE: @@ -376,3 +383,83 @@ function useEntityHandler( // this import via src/lib/supabase.ts, so the hook Just Works everywhere the // data client does. __registerEntityUseQuery(useEntityHandler as Parameters[0]); + +// --------------------------------------------------------------------------- +// bool.room hooks. Same registration pattern as useQuery above: core is +// React-free, importing this entry arms the hooks. Every hook ACQUIRES the +// room machinery on mount and releases on unmount — the connection exists +// exactly while something on screen cares about it. + +type AnyRoomStore = RoomStore>; + +function useRoomAcquire(store: AnyRoomStore): void { + // StrictMode mounts, unmounts, and remounts: acquire/release must be exactly + // paired, and the store's ref-count (with its generation counter) makes the + // double-cycle safe. + useEffect(() => store.acquire(), [store]); +} + +function useRoomOthers(store: AnyRoomStore): ReadonlyArray>> { + useRoomAcquire(store); + return useSyncExternalStore( + (cb) => store.onOthers(cb), + () => store.getOthers(), + () => store.getOthers(), + ); +} + +function useRoomSetMe(store: AnyRoomStore): (patch: Record) => void { + useRoomAcquire(store); + // Track which keys THIS component wrote, and clear exactly those on unmount: + // a cursor must not outlive the screen that was publishing it, and two + // components publishing different keys must not clobber each other. + const keysRef = useRef>(new Set()); + useEffect(() => { + const keys = keysRef.current; + return () => { + if (keys.size > 0) store.clearMe([...keys]); + }; + }, [store]); + const [setter] = useState(() => (patch: Record) => { + for (const k of Object.keys(patch)) keysRef.current.add(k); + store.setMe(patch); + }); + return setter; +} + +function useRoomEventListener(store: AnyRoomStore, event: string, cb: (e: RoomEvent) => void): void { + useRoomAcquire(store); + // Latest-ref: the handler closes over fresh props/state on every render, + // while the subscription itself is stable — no stale closures, no resubscribe + // churn, nothing for StrictMode to double. + const cbRef = useRef(cb); + cbRef.current = cb; + const eventRef = useRef(event); + eventRef.current = event; + useEffect( + () => + store.onEvent((name, e) => { + if (name === eventRef.current) cbRef.current(e); + }), + [store], + ); +} + +function useRoomStatus(store: AnyRoomStore): RoomStatus { + useRoomAcquire(store); + return useSyncExternalStore( + (cb) => { + const off = store.onStatus(() => cb()); + return () => off(); + }, + () => store.status(), + () => store.status(), + ); +} + +__registerRoomHooks({ + useOthers: useRoomOthers, + useSetMe: useRoomSetMe, + useEventListener: useRoomEventListener, + useStatus: useRoomStatus, +}); diff --git a/src/realtime.test.ts b/src/realtime.test.ts index 79741b7..5a030b2 100644 --- a/src/realtime.test.ts +++ b/src/realtime.test.ts @@ -1,276 +1,158 @@ import { describe, expect, test } from "bun:test"; -import { - createDoorbell, - type DoorbellDeps, - type DoorbellStatus, - type MintResult, - type RealtimeMint, -} from "./realtime"; +import { createDoorbell, type DoorbellStatus } from "./realtime"; +import { createSession, type SessionMintResult } from "./session"; 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. +// The doorbell is a pure consumer: fan out payloads, pick topics, report +// status. Everything about the connection lifecycle (mint, refresh, backoff, +// wake, teardown, the sync-CLOSED scar) is the session's job and is tested +// once, in session.test.ts — not re-tested here. -type FakeChannel = { - topic: string; - priv: boolean; - joined: boolean; - left: boolean; - cb: ((p: BoolChangePayload) => void) | null; - status: ((s: string) => void) | null; -}; +const TOPICS = { app: "bool:app_x:app", user: "bool:app_x:user:u1", room: "bool:app_x:room" }; -function makeHarness(opts: { mints?: MintResult[] } = {}) { - const mints = [...(opts.mints ?? [])]; +function makeHarness(opts?: { mints?: SessionMintResult[] }) { + 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(); - }, + channels: [] as Array<{ + topic: string; + joinCb: ((s: string) => void) | null; + broadcastCb: ((msg: { event: string; payload: unknown }) => void) | null; + left: boolean; + }>, + statuses: [] as DoorbellStatus[], + live: () => h.channels.filter((c) => !c.left), ding(topic: string, payload: BoolChangePayload) { - for (const ch of h.channels) { - if (ch.topic === topic && ch.joined && !ch.left) ch.cb?.(payload); + for (const ch of h.live()) { + if (ch.topic === topic) ch.broadcastCb?.({ event: "change", payload }); } }, - live: () => h.channels.filter((c) => c.joined && !c.left), + joinAll: async () => { + for (let i = 0; i < 50 && h.live().some((c) => !c.joinCb); i++) await Promise.resolve(); + for (const c of h.live()) c.joinCb?.("SUBSCRIBED"); + await Promise.resolve(); + }, + tick: async () => { + for (let i = 0; i < 20; i++) await Promise.resolve(); + }, }; - const deps: DoorbellDeps = { - async mint(): Promise { + + const session = createSession({ + async mint() { h.mintCalls++; - return mints.length ? mints.shift()! : { ok: false, reason: "unauthorized" }; + return mints.length + ? mints.shift()! + : { ok: true, token: `tok-${h.mintCalls}`, expiresIn: 900, topics: TOPICS }; }, - 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); + setAuth: () => {}, + schedule: (fn, ms) => setTimeout(fn, Math.min(ms, 5)), + cancel: (t) => clearTimeout(t as ReturnType), + wake: () => () => {}, + }); + + const doorbell = createDoorbell({ + session, + makeChannel(topic) { + const rec = { + topic, + joinCb: null as ((s: string) => void) | null, + broadcastCb: null as ((msg: { event: string; payload: unknown }) => void) | null, + left: false, + }; + h.channels.push(rec); return { onBroadcast(cb) { - ch.cb = cb; + rec.broadcastCb = cb; }, - join(status) { - ch.joined = true; - ch.status = status; + join(cb) { + rec.joinCb = cb; }, leave() { - ch.left = true; + rec.left = true; }, }; }, - 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 RAW: RealtimeMint = { - token: "tok-1", - expiresIn: 900, - topics: { app: "bool:app_x:app", user: "bool:app_x:user:u1" }, -}; -const MINT: MintResult = { ok: true, mint: RAW }; -const ok = (m: Partial): MintResult => ({ ok: true, mint: { ...RAW, ...m } }); -const refused: MintResult = { ok: false, reason: "unauthorized" }; -const down: MintResult = { ok: false, reason: "unavailable" }; -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], - ]); + onStatus: (s) => h.statuses.push(s), }); - test("anonymous mint (no user topic) joins the app room only", async () => { - const { h, doorbell } = makeHarness({ - mints: [ok({ topics: { app: "bool:app_x:app", user: null } })], - }); - doorbell.subscribe(() => {}); - await tick(); - expect(h.live().map((c) => c.topic)).toEqual(["bool:app_x:app"]); - }); + return { h, doorbell, session }; +} - test("payloads from either room reach every listener (one shared doorbell)", async () => { - const { h, doorbell } = makeHarness({ mints: [MINT] }); +describe("doorbell: topics and fan-out", () => { + test("joins app + user rooms; payloads reach every listener via ONE doorbell", async () => { + const { h, doorbell } = makeHarness(); 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 + const offA = doorbell.subscribe((p) => a.push(p)); + const offB = doorbell.subscribe((p) => b.push(p)); + await h.tick(); + await h.joinAll(); + expect(h.mintCalls).toBe(1); // second subscriber reuses the running session + expect(h.live().map((c) => c.topic)).toEqual(["bool:app_x:app", "bool:app_x:user:u1"]); + 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" }); + offA(); + offB(); }); -}); -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, ok({ 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); + test("anonymous mint (no user topic) joins the app room only", async () => { + const { h, doorbell } = makeHarness({ + mints: [{ ok: true, token: "t", expiresIn: 900, topics: { ...TOPICS, user: null } }], + }); + const off = doorbell.subscribe(() => {}); + await h.tick(); + expect(h.live().map((c) => c.topic)).toEqual(["bool:app_x:app"]); + off(); }); - test("a refused re-mint (revoked access) drops the rooms and reports unauthorized", async () => { - // The TTL expiring IS the revocation mechanism: once the gateway stops - // issuing wristbands the socket must go quiet, not keep listening. - const { h, doorbell } = makeHarness({ mints: [MINT, refused] }); + test("the last unsubscribe stops delivery; late payloads reach nobody", async () => { + const { h, doorbell } = makeHarness(); const got: BoolChangePayload[] = []; - doorbell.subscribe((p) => got.push(p)); - await tick(); - await h.fire(); + const off = doorbell.subscribe((p) => got.push(p)); + await h.tick(); + await h.joinAll(); + off(); expect(h.live()).toHaveLength(0); - expect(doorbell.status()).toBe("unauthorized"); - // late payloads on a torn-down 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: [ok({ expiresIn: 1 })] }); - doorbell.subscribe(() => {}); - await tick(); - expect(h.scheduled[0]!.ms).toBe(30_000); - }); }); -// There is NO public channel to fall back to — every topic is wristband-gated. -// So a failure is reported and retried, never disguised as liveness. -describe("createDoorbell: failure is surfaced, not disguised", () => { +describe("doorbell: status is surfaced, never disguised", () => { test("a refused mint reports unauthorized and joins nothing", async () => { - const { h, doorbell } = makeHarness({ mints: [refused] }); - const got: BoolChangePayload[] = []; - doorbell.subscribe((p) => got.push(p)); - await tick(); + const { h, doorbell } = makeHarness({ + mints: [{ ok: false, reason: "unauthorized" }], + }); + const off = doorbell.subscribe(() => {}); + await h.tick(); expect(h.live()).toHaveLength(0); - expect(h.authed).toHaveLength(0); expect(doorbell.status()).toBe("unauthorized"); - expect(got).toHaveLength(0); + off(); }); test("an unreachable gateway reports unavailable — NOT unauthorized", async () => { - // A public app admits every visitor, so a dropped request must never be + // A public app admits every visitor; a dropped request must never be // presented as "you aren't allowed to watch this". - const { doorbell } = makeHarness({ mints: [down] }); - doorbell.subscribe(() => {}); - await tick(); - expect(doorbell.status()).toBe("unavailable"); - }); - - test("status transitions are reported to the client", async () => { - const seen: DoorbellStatus[] = []; - const { h, doorbell } = makeHarness({ mints: [MINT] }); - // onStatus isn't part of makeHarness; assert via the accessor instead. - expect(doorbell.status()).toBe("connecting"); - doorbell.subscribe(() => {}); - await tick(); - h.channels[0]!.status!("SUBSCRIBED"); - expect(doorbell.status()).toBe("live"); - void seen; - }); - - test("retries with capped backoff instead of hot-looping", async () => { - const { h, doorbell } = makeHarness({ mints: [refused, refused, refused, refused, refused] }); - doorbell.subscribe(() => {}); - await tick(); - const delays: number[] = [h.scheduled[0]!.ms]; - for (let i = 0; i < 3; i++) { - await h.fire(); - delays.push(h.scheduled.at(-1)!.ms); - } - expect(delays).toEqual([1_000, 5_000, 15_000, 60_000]); - // and it keeps trying — a granted-later viewer eventually goes live - expect(h.mintCalls).toBeGreaterThan(1); - }); - - test("a good join resets the backoff so a later blip starts quick again", async () => { - const { h, doorbell } = makeHarness({ mints: [refused, MINT] }); - doorbell.subscribe(() => {}); - await tick(); - expect(h.scheduled[0]!.ms).toBe(1_000); - await h.fire(); // second attempt mints fine - h.channels[0]!.status!("SUBSCRIBED"); - expect(doorbell.status()).toBe("live"); - // a refused join now → backoff restarts at the first step - h.channels[0]!.status!("CHANNEL_ERROR"); - expect(h.scheduled.at(-1)!.ms).toBe(1_000); - }); - - test("a refused private join tears down and retries (wristband/policy disagree)", async () => { - const { h, doorbell } = makeHarness({ mints: [MINT] }); - doorbell.subscribe(() => {}); - await tick(); - h.channels[0]!.status!("CHANNEL_ERROR"); - expect(h.live()).toHaveLength(0); - expect(doorbell.status()).toBe("unavailable"); - }); - - test("last unsubscribe tears everything down; a late mint resolution no-ops", async () => { - let release!: (m: MintResult) => void; - const slowMint = new Promise((r) => (release = r)); - const h = { channels: [] as FakeChannel[], authed: [] as string[] }; - const doorbell = createDoorbell({ - mint: () => slowMint, - 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; - }, - }; - }, + const { h, doorbell } = makeHarness({ + mints: [{ ok: false, reason: "unavailable" }], }); const off = doorbell.subscribe(() => {}); - off(); // teardown while the mint is still in flight - release(MINT); - await tick(); - expect(h.authed).toHaveLength(0); - expect(h.channels).toHaveLength(0); + await h.tick(); + expect(doorbell.status()).toBe("unavailable"); + off(); }); - test("resubscribing after teardown starts a fresh doorbell", async () => { - const { h, doorbell } = makeHarness({ mints: [MINT, ok({ token: "tok-2" })] }); + test("status transitions flow through to the client hook", async () => { + const { h, doorbell } = makeHarness(); const off = doorbell.subscribe(() => {}); - await tick(); + await h.tick(); + await h.joinAll(); + expect(doorbell.status()).toBe("live"); + expect(h.statuses).toContain("live"); 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 index 0bcf606..38a9fc3 100644 --- a/src/realtime.ts +++ b/src/realtime.ts @@ -1,219 +1,113 @@ -// The private doorbell client — the SDK half of Bool's realtime design -// (platform: lib/bool-db.ts GATEWAY_GLOBAL_SETUP_SQL + the -// /_bool/v1/realtime/token plane). +// The entity doorbell — change payloads for live entity views. // -// Live updates arrive as ROW-BEARING broadcasts on PRIVATE topics. Every topic -// is private: a socket holding no gateway-minted wristband hears nothing, and -// there is no public channel to fall back to. This module owns the 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 never duplicate), -// - re-mint at ~75% of the TTL so the socket never hits token expiry, -// - on failure, RETRY with capped backoff and report status — never pretend. -// A refused mint means the viewer isn't authorized (or the gateway is -// unreachable); either way the honest outcome is "not live", surfaced, with -// HTTP reads still working. Silently limping along on a degraded path is how -// a broken realtime layer goes unnoticed for a week. -// - share ONE doorbell across every subscriber (each entities.
-// .subscribe used to open its own channel; now they ref-count this one). +// A PURE CONSUMER of the shared RealtimeSession (session.ts): this module +// contains zero connection code. It declares which topics it wants (the app +// room, plus the personal user room when the wristband names a signed-in +// user), fans incoming change payloads out to every subscriber, and reports +// its delivery status. Mint, refresh, backoff, wake and teardown live in the +// session — written once, shared with bool.room. // -// Everything effectful is injected (DoorbellDeps) so the machine is fully -// unit-testable without sockets. +// There is NO public channel and no fallback: every topic is wristband-gated, +// so a failure is reported and retried, never disguised as liveness. The old +// public compat channel cost 2× writes forever, leaked row ids on an +// anon-joinable topic, and spawned three follow-on bugs to serve nobody. +// +// Trust note: payloads on these topics are merged into entity state as +// server-trusted data. That is safe ONLY because the server's send policy +// makes the doorbell topics client-unwritable (clients may publish solely to +// the room topic family). Never weaken that policy; never join these topics +// with anything but `private: true`. import type { BoolChangePayload } from "./client.js"; +import type { + ChannelGroup, + RealtimeSession, + RealtimeTopics, + SessionChannel, + SessionStatus, +} from "./session.js"; + +export type DoorbellStatus = SessionStatus; -/** What the wristband desk returns. Topic names are SERVER-authored so naming - * lives in exactly one place. */ +/** One wristband, as the gateway mints it (/_bool/v1/realtime/token). */ 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; + topics: RealtimeTopics; }; -/** Why the doorbell isn't delivering, when it isn't. - * - `connecting` — first mint/join in flight - * - `live` — joined; changes are arriving - * - `unauthorized` — the gateway refused a wristband (not authorized for this - * app, or end-user session expired). Retried, since access can be granted. - * - `unavailable` — gateway unreachable, realtime misconfigured, or the join - * was refused. Retried with backoff. */ -export type DoorbellStatus = "connecting" | "live" | "unauthorized" | "unavailable"; - -/** Outcome of one mint attempt. The two failure kinds are distinguished on - * purpose: "the gateway says no" and "I couldn't reach the gateway" mean very - * different things to a viewer (and to whoever is debugging), and a public app - * — where every visitor is normally admitted — must never be told it's - * unauthorized because of a dropped request. */ export type MintResult = | { ok: true; mint: RealtimeMint } | { ok: false; reason: "unauthorized" | "unavailable" }; export type DoorbellDeps = { - /** Mint one wristband. On failure report WHICH kind (see MintResult) — the - * doorbell retries either way and never silently gives up. */ - mint(): Promise; - /** Present the wristband to the realtime connection (supabase setAuth). */ - setAuth(token: string): Promise | void; - /** Create (not join) a private channel on `topic`. */ - channel(topic: string, opts: { private: boolean }): DoorbellChannel; + session: RealtimeSession; + /** Create (not join) a private channel on `topic` with broadcast handlers + * attachable. The session owns join/teardown. */ + makeChannel(topic: string): Pick; /** Called whenever delivery state changes, so a client can surface it. */ onStatus?: (status: DoorbellStatus) => void; - /** 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 with a nonsense TTL shouldn't melt into a mint loop. -const MIN_REFRESH_MS = 30_000; -// Retry backoff after a refused/failed mint or join: quick first, then capped so -// a long outage costs one request a minute rather than a hot loop. -const RETRY_MS = [1_000, 5_000, 15_000, 60_000]; - export type Doorbell = { - /** Register a change listener. Starts the machinery on the first listener, - * tears it down after the last unsubscribes. Returns unsubscribe. */ + /** Register a change listener. Starts delivery on the first listener, stops + * after the last unsubscribes. Returns unsubscribe. */ subscribe(listener: (payload: BoolChangePayload) => void): () => void; /** Current delivery state — for surfacing "not live" in a UI. */ status(): DoorbellStatus; }; -export function createDoorbell(deps: DoorbellDeps): Doorbell { - const schedule = deps.schedule ?? ((fn, ms) => setTimeout(fn, ms)); - const cancel = deps.cancel ?? ((h) => clearTimeout(h as ReturnType)); +/** Fill the SessionChannel members the doorbell never uses, so its adapter + * stays as small as what it actually does: receive broadcasts. */ +function asSessionChannel( + ch: Pick, +): SessionChannel { + return { + ...ch, + track: (_s, done) => done?.("ok"), + onPresence: () => {}, + send: (_e, _p, done) => done?.("ok"), + }; +} +export function createDoorbell(deps: DoorbellDeps): Doorbell { 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 timer: unknown = null; - let attempt = 0; let state: DoorbellStatus = "connecting"; - function setStatus(next: DoorbellStatus): void { - if (state === next) return; - state = next; - deps.onStatus?.(next); - } - - function clearTimer(): void { - if (timer !== null) { - cancel(timer); - timer = null; - } - } - - function teardownChannels(): void { - for (const ch of channels) ch.leave(); - channels = []; - clearTimer(); - } - - /** Schedule the next attempt after a failure, with capped backoff. */ - function retry(myGen: number, why: DoorbellStatus): void { - setStatus(why); - const ms = RETRY_MS[Math.min(attempt, RETRY_MS.length - 1)]!; - attempt++; - timer = schedule(() => { - if (myGen !== gen) return; - void start(myGen); - }, ms); - } - - async function start(myGen: number): Promise { - teardownChannels(); - if (myGen !== gen) return; - - const res = await deps.mint(); - if (myGen !== gen) return; - // No wristband: the app is NOT live. Report which kind of "no" it was and - // try again — never fake liveness. - if (!res.ok) return retry(myGen, res.reason); - const m = res.mint; - - await deps.setAuth(m.token); - if (myGen !== gen) return; - + const group: ChannelGroup = { // 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. - let joined = 0; - 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((s) => { - if (myGen !== gen) return; - if (s === "SUBSCRIBED") { - joined++; - attempt = 0; // a good join resets the backoff - setStatus("live"); - } else if (s === "CHANNEL_ERROR" || s === "TIMED_OUT") { - // A refused private join means the wristband and the policy disagree - // (misconfigured database, clock skew). Retry the whole handshake. - teardownChannels(); - retry(myGen, "unavailable"); - } + topics: (t) => [t.app, t.user].filter((x): x is string => Boolean(x)), + open(topic) { + const ch = deps.makeChannel(topic); + ch.onBroadcast((msg) => { + fanout(((msg as { payload?: BoolChangePayload }).payload ?? {}) as BoolChangePayload); }); - channels.push(ch); - } - if (joined === 0 && channels.length === 0) return retry(myGen, "unavailable"); - - scheduleRefresh(myGen, m.expiresIn); - } + return asSessionChannel(ch); + }, + onStatus(s) { + if (state === s) return; + state = s; + deps.onStatus?.(s); + }, + }; - function scheduleRefresh(myGen: number, expiresInSeconds: number): void { - const ms = Math.max(expiresInSeconds * 1000 * REFRESH_FRACTION, MIN_REFRESH_MS); - timer = schedule(async () => { - if (myGen !== gen) return; - const res = await deps.mint(); - if (myGen !== gen) return; - // Renewal refused — access was revoked, or the gateway is down. The TTL - // expiring is the whole revocation mechanism, so drop the rooms. - if (!res.ok) { - teardownChannels(); - return retry(myGen, res.reason); - } - await deps.setAuth(res.mint.token); // channels re-auth on the connection - if (myGen !== gen) return; - scheduleRefresh(myGen, res.mint.expiresIn); - }, ms); - } + let unregister: (() => void) | null = null; return { subscribe(listener) { listeners.add(listener); - if (listeners.size === 1) { - gen++; - attempt = 0; - setStatus("connecting"); - void start(gen); - } + if (listeners.size === 1) unregister = deps.session.register(group); return () => { if (!listeners.delete(listener)) return; if (listeners.size === 0) { - gen++; // orphan any in-flight start/refresh - teardownChannels(); - state = "connecting"; + unregister?.(); + unregister = null; } }; }, diff --git a/src/room.test.ts b/src/room.test.ts new file mode 100644 index 0000000..d3714fa --- /dev/null +++ b/src/room.test.ts @@ -0,0 +1,907 @@ +import { describe, expect, test } from "bun:test"; +import { + colorForId, + createRoomStore, + throttleForPeers, + validateRoomId, +} from "./room"; +import { createSession, type SessionChannel, type SessionMintResult } from "./session"; + +type RoomMintResult = + | { ok: true; token: string; expiresIn: number; topic: string | null } + | { ok: false; reason: "unauthorized" | "unavailable" }; +type RoomChannel = SessionChannel; + +// A fake transport: a "wire" shared by any number of stores, so multi-peer +// behavior (presence sync, broadcast fan-out, self-filtering) is exercised for +// real, without sockets. Mirrors the injection discipline of realtime.ts. +type Wire = { + presences: Map>; + presenceSubs: Set<(states: Record>>) => void>; + broadcastSubs: Set<{ selfKey: string; cb: (msg: { event: string; payload: unknown }) => void }>; + syncAll(): void; +}; + +function makeWire(): Wire { + const wire: Wire = { + presences: new Map(), + presenceSubs: new Set(), + broadcastSubs: new Set(), + syncAll() { + const states: Record>> = {}; + for (const [k, v] of wire.presences) states[k] = [{ presence_ref: "r", ...v }]; + for (const cb of wire.presenceSubs) cb(states); + }, + }; + return wire; +} + +function makeHarness(opts?: { + mint?: () => Promise; + wire?: Wire; + /** What the fake channel reports back from track() — supabase-js resolves + * with a STRING, and "timed out"/"error" must never be silent. */ + trackResult?: string; + roomId?: string | null; +}) { + const wire = opts?.wire ?? makeWire(); + const trackResult = opts?.trackResult ?? "ok"; + // Deterministic virtual clock so throttle behavior is testable exactly. + // Session and store share it — one clock drives the whole stack. + let clock = 0; + const timers: Array<{ at: number; fn: () => void; id: number }> = []; + let nextId = 1; + const advance = async (ms: number) => { + const target = clock + ms; + // run due timers in order, allowing them to schedule more + for (;;) { + timers.sort((a, b) => a.at - b.at); + const due = timers[0]; + if (!due || due.at > target) break; + clock = due.at; + timers.shift(); + due.fn(); + await flush(); + } + clock = target; + // Always flush, even when no timer was due: the handshake is a chain of + // awaits (mint race → setAuth → channel), so a caller that only advances + // the clock would otherwise observe a half-settled machine. + await flush(); + }; + const flush = async () => { + for (let i = 0; i < 20; i++) await Promise.resolve(); + }; + + let joinCb: ((state: string) => void) | null = null; + const sent: Array<{ event: string; payload: unknown }> = []; + let key = ""; + // Wake signals are injected the same way timers are, so "reconnects the + // instant the tab comes back" is testable without a real document. + const wakeCbs = new Set<() => void>(); + let mintCalls = 0; + + const baseMint = + opts?.mint ?? + (async (): Promise => ({ + ok: true, + token: "t", + expiresIn: 900, + topic: "bool:app_x:room", + })); + + const schedule = (fn: () => void, ms: number) => { + const id = nextId++; + timers.push({ at: clock + ms, fn, id }); + return id; + }; + const cancel = (h: unknown) => { + const i = timers.findIndex((t) => t.id === h); + if (i !== -1) timers.splice(i, 1); + }; + const now = () => clock; + + const session = createSession({ + mint: (): Promise => { + mintCalls++; + return baseMint().then((r) => + r.ok + ? { + ok: true as const, + token: r.token, + expiresIn: r.expiresIn, + topics: { app: "bool:app_x:app", user: null, room: r.topic }, + } + : r, + ); + }, + setAuth: () => {}, + schedule, + cancel, + now, + wake: (cb) => { + wakeCbs.add(cb); + return () => wakeCbs.delete(cb); + }, + }); + + let lastTopic = ""; + const makeChannel = (topic: string, k: string): RoomChannel => { + lastTopic = topic; + key = k; + const ch: RoomChannel = { + track(state, done) { + wire.presences.set(k, state); + wire.syncAll(); + done?.(trackResult); + }, + onPresence(cb) { + wire.presenceSubs.add(cb); + }, + onBroadcast(cb) { + wire.broadcastSubs.add({ selfKey: k, cb }); + }, + send(event, payload, done) { + sent.push({ event, payload }); + done?.("ok"); + // deliver to every OTHER subscriber on the wire (server self:false) + for (const sub of wire.broadcastSubs) { + if (sub.selfKey !== k) sub.cb({ event, payload }); + } + }, + join(status) { + joinCb = status; + }, + leave() { + wire.presences.delete(k); + wire.syncAll(); + }, + }; + return ch; + }; + + const store = createRoomStore({ + session, + roomId: opts?.roomId ?? null, + makeChannel, + schedule, + cancel, + now, + }); + return { + store, + wire, + sent, + advance, + selfKey: () => key, + lastTopic: () => lastTopic, + mintCalls: () => mintCalls, + /** Simulate the tab/network coming back. */ + wake: () => { + for (const cb of [...wakeCbs]) cb(); + }, + wakeSubs: () => wakeCbs.size, + // Wait for the store's real handshake (mint → setAuth → channel) instead of + // a fixed number of microtasks: the number changed when mint gained a + // timeout race, and a hardcoded tick count silently stopped joining at all. + join: async () => { + for (let i = 0; i < 50 && !joinCb; i++) await Promise.resolve(); + if (!joinCb) throw new Error("channel was never created — the store never reached join()"); + joinCb("SUBSCRIBED"); + await Promise.resolve(); + }, + fire: (state: string) => joinCb?.(state), + }; +} + +describe("colorForId", () => { + test("is deterministic and valid hsl", () => { + expect(colorForId("abc")).toBe(colorForId("abc")); + expect(colorForId("abc")).toMatch(/^hsl\(\d+ 85% 55%\)$/); + }); + + test("two clients compute the same color for the same peer (nothing transmitted)", () => { + // The property the design relies on — no color field on the wire. + const a = colorForId("peer-1"); + const b = colorForId("peer-1"); + expect(a).toBe(b); + }); +}); + +describe("room store: presence", () => { + test("others excludes self, includes peers, and strips presence_ref", async () => { + const wire = makeWire(); + const h = makeHarness({ wire }); + const release = h.store.acquire(); + await h.join(); + + // a peer lands on the wire + wire.presences.set("peer-1", { cursor: { x: 1, y: 2 } }); + wire.syncAll(); + + const others = h.store.getOthers(); + expect(others.length).toBe(1); + expect(others[0]!.id).toBe("peer-1"); + expect(others[0]!.presence).toEqual({ cursor: { x: 1, y: 2 } }); + expect("presence_ref" in others[0]!.presence).toBe(false); + expect(others.some((o) => o.id === h.store.self.id)).toBe(false); + release(); + }); + + test("a peer who joined but set nothing has EMPTY presence (the ghost-cursor case)", async () => { + const wire = makeWire(); + const h = makeHarness({ wire }); + const release = h.store.acquire(); + await h.join(); + + wire.presences.set("peer-1", {}); + wire.syncAll(); + const o = h.store.getOthers()[0]!; + // Partial

: reading o.presence.cursor must be survivable — this is the + // state real apps crash on if they don't guard. + expect(o.presence).toEqual({}); + release(); + }); + + // These four assert at the only altitude that survived the transport change: + // what a PEER on the same wire observes. (They used to peek at the wire's + // presence map — which went blank the day setMe state moved from presence, + // where the server throttles cursor-frequency writes to death, to broadcast.) + test("setMe merges patches and undefined deletes a key", async () => { + const wire = makeWire(); + const a = makeHarness({ wire }); + const b = makeHarness({ wire }); + const releaseA = a.store.acquire(); + await a.join(); + const releaseB = b.store.acquire(); + await b.join(); + + a.store.setMe({ cursor: { x: 1, y: 1 } }); + await a.advance(30); + a.store.setMe({ typing: true }); + await a.advance(30); + const seen = () => b.store.getOthers().find((o) => o.id === a.store.self.id)!.presence; + expect(seen()).toEqual({ cursor: { x: 1, y: 1 }, typing: true }); + + a.store.setMe({ typing: undefined }); + await a.advance(30); + expect(seen()).toEqual({ cursor: { x: 1, y: 1 } }); + releaseA(); + releaseB(); + }); + + test("setMe is trailing-edge throttled: a 60fps burst collapses but the FINAL position lands", async () => { + const wire = makeWire(); + const a = makeHarness({ wire }); + const b = makeHarness({ wire }); + const releaseA = a.store.acquire(); + await a.join(); + const releaseB = b.store.acquire(); + await b.join(); + const sendsBefore = a.sent.filter((s) => s.event === "~me").length; + + // 20 moves in ~80ms (way over the 25ms throttle) + for (let i = 1; i <= 20; i++) { + a.store.setMe({ cursor: { x: i, y: i } }); + await a.advance(4); + } + await a.advance(50); // let the trailing edge fire + const sends = a.sent.filter((s) => s.event === "~me").length - sendsBefore; + expect(sends).toBeLessThanOrEqual(6); // ~80ms / 25ms + trailing edge + expect(sends).toBeGreaterThan(0); + const seen = b.store.getOthers().find((o) => o.id === a.store.self.id)! + .presence as { cursor: { x: number } }; + expect(seen.cursor.x).toBe(20); // the last write always wins on the wire + releaseA(); + releaseB(); + }); + + test("state set before the join is replayed ON join (never invisible)", async () => { + const wire = makeWire(); + const b = makeHarness({ wire }); + const releaseB = b.store.acquire(); + await b.join(); + const a = makeHarness({ wire }); + const releaseA = a.store.acquire(); + a.store.setMe({ name: "jack" }); // before SUBSCRIBED + await a.join(); + await a.advance(5); + const seen = b.store.getOthers().find((o) => o.id === a.store.self.id)!.presence; + expect(seen).toEqual({ name: "jack" }); + releaseA(); + releaseB(); + }); + + test("a late joiner is hydrated by the existing peers' re-send (jittered)", async () => { + // The reverse of the test above: A already has state, THEN B joins. B has + // missed A's ~me broadcasts entirely; A must re-send when it sees a new + // member, or B renders A as an empty ghost until A's next move. + const wire = makeWire(); + const a = makeHarness({ wire }); + const releaseA = a.store.acquire(); + await a.join(); + a.store.setMe({ name: "jack" }); + await a.advance(30); + + const b = makeHarness({ wire }); + const releaseB = b.store.acquire(); + await b.join(); + // before A's hydration fires, B knows A only as a member + await a.advance(301); // past the max hydrate jitter + const seen = b.store.getOthers().find((o) => o.id === a.store.self.id)!.presence; + expect(seen).toEqual({ name: "jack" }); + releaseA(); + releaseB(); + }); + + test("clearMe removes exactly the named keys (hook-unmount semantics)", async () => { + const wire = makeWire(); + const a = makeHarness({ wire }); + const b = makeHarness({ wire }); + const releaseA = a.store.acquire(); + await a.join(); + const releaseB = b.store.acquire(); + await b.join(); + a.store.setMe({ cursor: { x: 1, y: 1 }, name: "jack" }); + await a.advance(30); + a.store.clearMe(["cursor"]); + await a.advance(30); + const seen = b.store.getOthers().find((o) => o.id === a.store.self.id)!.presence; + expect(seen).toEqual({ name: "jack" }); + releaseA(); + releaseB(); + }); + + test("app events can never squat the SDK's reserved wire namespace", async () => { + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + expect(() => h.store.broadcast("~me", { fake: true })).toThrow(/reserved for the SDK/); + expect(() => h.store.broadcast("~anything", 1)).toThrow(/reserved/); + release(); + }); +}); + +describe("room store: broadcast + echo", () => { + test("the sender's own listeners fire synchronously, before any network", async () => { + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + + const got: unknown[] = []; + h.store.onEvent((event, e) => got.push([event, e.from, e.data])); + h.store.broadcast("reaction", { emoji: "🎉" }); + // No advance(), no await: the echo is same-tick by contract. + expect(got).toEqual([["reaction", h.store.self.id, { emoji: "🎉" }]]); + release(); + }); + + test("a wire copy of my own event is NOT delivered twice", async () => { + const wire = makeWire(); + const h = makeHarness({ wire }); + const release = h.store.acquire(); + await h.join(); + + const got: unknown[] = []; + h.store.onEvent(() => got.push(1)); + h.store.broadcast("x", 1); + // simulate a server that echoes anyway (belt over the self:false config) + for (const sub of wire.broadcastSubs) sub.cb({ event: "x", payload: { f: h.store.self.id, d: 1 } }); + expect(got.length).toBe(1); + release(); + }); + + test("two stores on one wire: events cross with the sender id attached", async () => { + const wire = makeWire(); + const a = makeHarness({ wire }); + const b = makeHarness({ wire }); + const ra = a.store.acquire(); + const rb = b.store.acquire(); + await a.join(); + await b.join(); + + const got: Array<{ from: string; data: unknown }> = []; + b.store.onEvent((_ev, e) => got.push(e)); + a.store.broadcast("ping", { n: 1 }); + expect(got).toEqual([{ from: a.store.self.id, data: { n: 1 } }]); + ra(); + rb(); + }); + + test("an oversized payload throws with the fix, instead of a silent server drop", async () => { + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + expect(() => h.store.broadcast("img", "x".repeat(70_000))).toThrow(/60000-byte/); + release(); + }); +}); + +describe("room store: lifecycle + status", () => { + test("acquire starts, last release tears down and empties others", async () => { + const wire = makeWire(); + const h = makeHarness({ wire }); + const r1 = h.store.acquire(); + const r2 = h.store.acquire(); + await h.join(); + wire.presences.set("peer-1", { a: 1 }); + wire.syncAll(); + expect(h.store.getOthers().length).toBe(1); + + r1(); + expect(h.store.status()).toBe("live"); // still held + r2(); + expect(h.store.getOthers().length).toBe(0); + // releasing twice is a no-op (StrictMode calls cleanups it owns exactly + // once, but defensive release must not underflow the ref-count) + r2(); + }); + + test("an unauthorized mint reports honestly and retries — never fakes liveness", async () => { + let calls = 0; + const h = makeHarness({ + mint: async () => { + calls++; + return { ok: false, reason: "unauthorized" }; + }, + }); + const release = h.store.acquire(); + await h.advance(1); // settle the first mint + expect(h.store.status()).toBe("unauthorized"); + await h.advance(1_100); // first backoff step + expect(calls).toBeGreaterThanOrEqual(2); + release(); + }); + + test("a platform without the room topic is UNAVAILABLE, not silently dead", async () => { + const h = makeHarness({ + mint: async () => ({ ok: true, token: "t", expiresIn: 900, topic: null }), + }); + const release = h.store.acquire(); + await h.advance(1); + expect(h.store.status()).toBe("unavailable"); + release(); + }); + + test("status transitions notify subscribers", async () => { + const h = makeHarness(); + const seen: string[] = []; + h.store.onStatus((s) => seen.push(s)); + const release = h.store.acquire(); + await h.join(); + expect(seen).toContain("live"); + release(); + }); +}); + +// Regression suite for the FIRST real-world failure of bool.room, found on a +// preview app (2026-07-29). Symptom: the app rendered "ROOM IS LIVE" with 0 +// peers forever; an external observer confirmed the tab was absent from +// presence, and a 34-second WebSocket watch showed ZERO frames and ZERO +// sockets. Four independent defects conspired, each individually silent: +// +// 1. `CLOSED` was not a handled channel state, and supabase-js does NOT +// auto-rejoin — a dropped socket meant permanent death. (Verified against +// a real close: the status arrives with no follow-up SUBSCRIBED.) +// 2. teardown() assigned `state` directly instead of via setStatus, so every +// useSyncExternalStore subscriber kept a stale snapshot — which is why the +// badge said "live" while nothing was connected. +// 3. track()/send() failures were swallowed: supabase-js RESOLVES with the +// string "timed out" / "error" rather than throwing, so `void ch.track()` +// discarded every failure. +// 4. mint() had no timeout, so one hung fetch could wedge the machine with no +// channel, no status change and no retry. +describe("room store: recovery (the dead-but-'live' regression)", () => { + test("CLOSED recovers instead of dying silently", async () => { + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + expect(h.store.status()).toBe("live"); + + h.fire("CLOSED"); // socket dropped: blip, server close, token expiry + await h.advance(1); + // Honest status, and a retry scheduled — not a permanent "live" lie. + expect(h.store.status()).toBe("unavailable"); + await h.advance(1_100); + await h.join(); + expect(h.store.status()).toBe("live"); + release(); + }); + + test("every terminal channel state drives recovery", async () => { + for (const terminal of ["CHANNEL_ERROR", "TIMED_OUT", "CLOSED"]) { + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + h.fire(terminal); + await h.advance(1); + expect(h.store.status()).toBe("unavailable"); + release(); + } + }); + + test("status changes ALWAYS notify subscribers — a stale snapshot is the bug", async () => { + // useStatus() is a useSyncExternalStore over this; a direct `state = …` + // write leaves React rendering the old value forever. + const h = makeHarness(); + const seen: string[] = []; + h.store.onStatus((s) => seen.push(s)); + const release = h.store.acquire(); + await h.join(); + expect(seen).toContain("live"); + release(); // last release → teardown + expect(seen[seen.length - 1]).toBe("connecting"); + expect(h.store.status()).toBe("connecting"); + }); + + test("a failed join beacon warns but does NOT tear the room down", async () => { + // The rate-shaping lesson, pinned in BOTH directions. The old rule — + // "any failed track() forces a reconnect" — looked like healing and was: + // under load, per-move track timeouts became a reconnect storm, the UI + // flapping between "N people here" and "live view unavailable". Recovery + // that can be triggered BY load must be rate-shaped: the beacon is + // once-per-join, its failure only warns, and a genuinely dead channel + // reports its own terminal state — which the session heals (tested in + // session.test.ts and the sync-CLOSED suite below). + const h = makeHarness({ trackResult: "timed out" }); + const release = h.store.acquire(); + await h.join(); + expect(h.store.status()).toBe("live"); // NOT torn down by the beacon result + h.store.setMe({ cursor: { x: 1, y: 1 } }); + await h.advance(40); + expect(h.store.status()).toBe("live"); // setMe rides broadcast; no track at all + expect(h.mintCalls()).toBe(1); // and no reconnect storm + release(); + }); + + test("a hung mint cannot wedge the room forever", async () => { + const h = makeHarness({ mint: () => new Promise(() => {}) }); // never settles + const release = h.store.acquire(); + await h.advance(11_000); // past the mint timeout + expect(h.store.status()).toBe("unavailable"); + release(); + }); + + test("recovery is not attempted once nobody is watching", async () => { + // A released store must stay quiet; acquire() starts a fresh generation. + const h = makeHarness({ trackResult: "error" }); + const release = h.store.acquire(); + await h.join(); + release(); + const before = h.store.status(); + h.store.setMe({ cursor: { x: 2, y: 2 } }); + await h.advance(100); + expect(h.store.status()).toBe(before); + }); +}); + +describe("bool.room waking up", () => { + test("a wake reconnects immediately instead of waiting out the backoff", async () => { + // The failure this fixes, observed on a real deployed app: a backgrounded + // tab loses its socket, honestly reports OFFLINE, and then sits on the + // [1s, 5s, 15s, 60s] ladder. Coming back to a page that says "offline" for + // half a minute is indistinguishable from broken. + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + expect(h.store.status()).toBe("live"); + + // Burn through the early rungs so the next scheduled retry is far away. + for (let i = 0; i < 3; i++) { + h.fire("CLOSED"); + await h.advance(20_000); + } + h.fire("CLOSED"); + await h.advance(0); + expect(h.store.status()).toBe("unavailable"); + const beforeWake = h.mintCalls(); + + // No clock advance at all: the reconnect must not be waiting on a timer. + h.wake(); + await h.advance(0); + expect(h.mintCalls()).toBe(beforeWake + 1); + await h.join(); + expect(h.store.status()).toBe("live"); + release(); + }); + + test("a wake while live does nothing (no churn on every tab focus)", async () => { + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + const before = h.mintCalls(); + h.wake(); + await h.advance(0); + expect(h.mintCalls()).toBe(before); + expect(h.store.status()).toBe("live"); + release(); + }); + + test("simultaneous wake signals coalesce into one reconnect", async () => { + // focus + visibilitychange + online all fire on the same tab re-selection. + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + h.fire("CLOSED"); + await h.advance(0); + const before = h.mintCalls(); + h.wake(); + h.wake(); + h.wake(); + await h.advance(0); + expect(h.mintCalls()).toBe(before + 1); + release(); + }); + + test("a later wake still works once the coalescing window has passed", async () => { + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + h.fire("CLOSED"); + await h.advance(0); + h.wake(); + await h.advance(0); + const after = h.mintCalls(); + h.fire("CLOSED"); + await h.advance(600); // past WAKE_COALESCE_MS + const beforeSecond = h.mintCalls(); + h.wake(); + await h.advance(0); + expect(h.mintCalls()).toBe(beforeSecond + 1); + expect(h.mintCalls()).toBeGreaterThan(after); + release(); + }); + + test("wake listeners are released with the last hook (no leak, no zombie)", async () => { + const h = makeHarness(); + expect(h.wakeSubs()).toBe(0); + const a = h.store.acquire(); + const b = h.store.acquire(); + await h.join(); + expect(h.wakeSubs()).toBe(1); // ref-counted, one subscription total + a(); + expect(h.wakeSubs()).toBe(1); + b(); + expect(h.wakeSubs()).toBe(0); + + // And a wake after full release must not resurrect anything. + const before = h.mintCalls(); + h.wake(); + await h.advance(100); + expect(h.mintCalls()).toBe(before); + }); +}); + +describe("bool.room broadcast size limit", () => { + test("measures UTF-8 bytes, not UTF-16 code units", async () => { + // `JSON.stringify(x).length` counts code units, so a payload of multi-byte + // characters measured at a third of what it actually sends — waving through + // an over-limit message for the server to drop silently, which is the exact + // failure the guard exists to name. + const h = makeHarness(); + const release = h.store.acquire(); + await h.join(); + + // 25k emoji: 50k UTF-16 code units (under the 60k limit by the old check) + // but 100k UTF-8 bytes (over it). + const emoji = "🎉".repeat(25_000); + expect(JSON.stringify({ f: "x", d: emoji }).length).toBeLessThan(60_000); + expect(() => h.store.broadcast("party", emoji)).toThrow(/over the 60000-byte/); + + // Plain ASCII of the same code-unit length is genuinely under and allowed. + expect(() => h.store.broadcast("party", "a".repeat(50_000))).not.toThrow(); + release(); + }); +}); + +describe("bool.room surviving supabase's synchronous CLOSED", () => { + // supabase's removeChannel fires the channel's own status callback with + // CLOSED — synchronously, from inside leave(). The pre-session machines + // recursed teardown → leave → CLOSED → teardown to a stack overflow on every + // real drop of a healthy channel (seen in production as an endless stream of + // "Maximum call stack size exceeded"). The guard lives in session.ts and is + // unit-tested there; this exercises it through the full store stack. + function makeSyncCloseHarness() { + let joinCb: ((s: string) => void) | null = null; + let channelsMade = 0; + const timers: Array<{ at: number; fn: () => void; id: number }> = []; + let clock = 0; + let nextId = 1; + const schedule = (fn: () => void, ms: number) => { + const id = nextId++; + timers.push({ at: clock + ms, fn, id }); + return id; + }; + const cancel = (h: unknown) => { + const i = timers.findIndex((t) => t.id === h); + if (i !== -1) timers.splice(i, 1); + }; + const now = () => clock; + const session = createSession({ + mint: async () => ({ + ok: true, + token: "t", + expiresIn: 900, + topics: { app: "bool:x:app", user: null, room: "bool:x:room" }, + }), + setAuth: () => {}, + schedule, + cancel, + now, + wake: () => () => {}, + }); + const store = createRoomStore({ + session, + roomId: null, + makeChannel: (): RoomChannel => { + channelsMade++; + return { + track: (_s, done) => done?.("ok"), + onPresence: () => {}, + onBroadcast: () => {}, + send: () => {}, + join: (cb) => { + joinCb = cb; + }, + leave: () => { + joinCb?.("CLOSED"); // what removeChannel actually does + }, + }; + }, + schedule, + cancel, + now, + }); + const advance = async (ms: number) => { + const target = clock + ms; + for (;;) { + timers.sort((a, b) => a.at - b.at); + const due = timers[0]; + if (!due || due.at > target) break; + clock = due.at; + timers.shift(); + due.fn(); + for (let i = 0; i < 20; i++) await Promise.resolve(); + } + clock = target; + for (let i = 0; i < 20; i++) await Promise.resolve(); + }; + return { + store, + advance, + fire: (s: string) => joinCb?.(s), + channelsMade: () => channelsMade, + join: async () => { + for (let i = 0; i < 50 && !joinCb; i++) await Promise.resolve(); + joinCb!("SUBSCRIBED"); + await Promise.resolve(); + }, + }; + } + + test("a real drop does not recurse, and the room still recovers", async () => { + const h = makeSyncCloseHarness(); + const release = h.store.acquire(); + await h.advance(0); + await h.join(); + expect(h.store.status()).toBe("live"); + + h.fire("CLOSED"); // pre-fix: RangeError, maximum call stack exceeded + expect(h.store.status()).toBe("unavailable"); + + // and recovery is intact: the retry rung reconnects a fresh channel + const before = h.channelsMade(); + await h.advance(1_000); + expect(h.channelsMade()).toBe(before + 1); + await h.join(); + expect(h.store.status()).toBe("live"); + release(); + }); + + test("release() with a live channel does not recurse either", async () => { + const h = makeSyncCloseHarness(); + const release = h.store.acquire(); + await h.advance(0); + await h.join(); + release(); // teardown → leave → sync CLOSED, on the unmount path + expect(h.store.status()).toBe("connecting"); + // and nothing keeps retrying after teardown + const before = h.channelsMade(); + await h.advance(120_000); + expect(h.channelsMade()).toBe(before); + }); +}); + +describe("bool.room render economy", () => { + test("a flood of ~me messages coalesces to ~one emit per frame", async () => { + const wire = makeWire(); + const a = makeHarness({ wire }); + const b = makeHarness({ wire }); + const releaseA = a.store.acquire(); + await a.join(); + const releaseB = b.store.acquire(); + await b.join(); + + let emits = 0; + b.store.onOthers(() => emits++); + // 12 updates land within one 16ms frame (setMe throttle is bypassed by + // sending straight through the wire helper — we're testing B's inbound + // side, not A's outbound throttle) + for (let i = 1; i <= 12; i++) { + a.store.setMe({ cursor: { x: i, y: i } }); + await a.advance(30); // A sends each one (past its own throttle)... + await b.advance(1); // ...B absorbs them nearly back-to-back + } + await b.advance(20); // trailing edge + // 12 inbound messages, spread over vastly fewer emits than messages + expect(emits).toBeLessThan(12); + expect(emits).toBeGreaterThan(0); + const seen = b.store.getOthers().find((o) => o.id === a.store.self.id)! + .presence as { cursor: { x: number } }; + expect(seen.cursor.x).toBe(12); // the final state always lands + releaseA(); + releaseB(); + }); + + test("a peer that did not change keeps its object identity across rebuilds", async () => { + // The still person's must not re-render because someone else + // moved: React.memo relies on the peer object being the SAME reference. + const wire = makeWire(); + const still = makeHarness({ wire }); + const mover = makeHarness({ wire }); + const observer = makeHarness({ wire }); + const r1 = still.store.acquire(); + await still.join(); + const r2 = mover.store.acquire(); + await mover.join(); + const r3 = observer.store.acquire(); + await observer.join(); + + still.store.setMe({ name: "still" }); + await still.advance(30); + await observer.advance(20); + const before = observer.store.getOthers().find((o) => o.id === still.store.self.id); + + mover.store.setMe({ cursor: { x: 5, y: 5 } }); + await mover.advance(30); + await observer.advance(20); + const after = observer.store.getOthers().find((o) => o.id === still.store.self.id); + expect(after).toBe(before); // same reference, not merely equal + r1(); r2(); r3(); + }); + + test("throttleForPeers holds 60Hz through 10 people and degrades gently", () => { + expect(throttleForPeers(0)).toBe(16); // alone: 60Hz + expect(throttleForPeers(3)).toBe(16); // Jack + two colleagues: 60Hz + expect(throttleForPeers(9)).toBe(16); // 10 people: still 60Hz + expect(throttleForPeers(14)).toBe(27); // 15 people ≈ 37Hz + expect(throttleForPeers(19)).toBe(48); // 20 people ≈ 21Hz — smooth, not slideshow + }); +}); + +describe("named rooms", () => { + test("a scoped room joins the scoped topic; the default room joins the bare one", async () => { + const base = makeHarness(); + const rd = base.store.acquire(); + await base.join(); + expect(base.lastTopic()).toBe("bool:app_x:room"); + rd(); + + const scoped = makeHarness({ roomId: "game:4" }); + const rs = scoped.store.acquire(); + await scoped.join(); + expect(scoped.lastTopic()).toBe("bool:app_x:room:game:4"); + rs(); + }); + + test("room ids are validated loudly, not normalized silently", () => { + // Silent normalization would put two "different" ids in one room. + expect(validateRoomId("game:4")).toBe("game:4"); + expect(validateRoomId("board.2-a_b")).toBe("board.2-a_b"); + expect(() => validateRoomId("")).toThrow(/1–64 characters/); + expect(() => validateRoomId("has space")).toThrow(/bool\.room\("has space"\)/); + expect(() => validateRoomId("emoji🎉")).toThrow(); + expect(() => validateRoomId("-starts-wrong")).toThrow(); + expect(() => validateRoomId("x".repeat(65))).toThrow(); + }); +}); diff --git a/src/room.ts b/src/room.ts new file mode 100644 index 0000000..96e4317 --- /dev/null +++ b/src/room.ts @@ -0,0 +1,594 @@ +// bool.room — the ephemeral lane. Live cursors, mid-stroke points, typing +// indicators, one-off reactions: data that flies browser-to-browser over the +// realtime socket and is NEVER stored. The durable lane (bool.entities) is for +// everything that must survive a reload; this is for everything that only +// matters while people are here together. +// +// This module is a PURE CONSUMER: it contains zero connection code. The +// mint/refresh/backoff/wake/teardown machinery lives in session.ts (shared +// with the entity doorbell), and this store registers one ChannelGroup per +// room. Before the split, this file carried its own copy of that machine and +// the copies drifted — see session.ts's header for the receipts. +// +// Design decisions that survived a week of production fires (full record: +// the platform's docs/2026-07-ephemeral-broadcast.md §9–§10): +// - MEMBERSHIP rides Supabase presence (auto-clears on disconnect — no +// hand-rolled heartbeats, no ghost cursors), but STATE rides broadcast. +// Presence is a server-side CRDT built for low-frequency state and it +// froze at 3 movers × 40Hz (track acks stopped inside a second); the +// identical load over broadcast delivered 97–100% at ~10ms. So: one +// `track({})` per join for who's-here, and `setMe` state flows as +// throttled FULL-state broadcasts on the reserved `~me` event, +// fire-and-forget — drops are healed by the next send. +// - The throttle lives HERE, not in app code: a prompt rule asking the +// model to throttle is forgettable; a setter that throttles isn't. +// - Inbound coalesces to one emit per ~frame with STABLE peer identities — +// otherwise every useOthers subscriber re-renders per message, and a +// memoized cursor re-renders because someone else moved. +// - Broadcast echoes to the SENDER locally and synchronously (never a round +// trip): one code path for "everyone sees it, including me". +// - Peer colors derive from peer ids (same hash on every client), so every +// viewer agrees on everyone's color with nothing transmitted. +// - Named rooms scope the topic (`bool::room:`); the default +// room is the bare topic. Same wristband — the schema is the boundary. +// +// SECURITY: this module publishes ONLY to the dedicated room topic family +// (`bool::room[:]`), never to the doorbell topics. Doorbell +// messages are merged into entity state as trusted server data; the server's +// send policy is scoped so a client cannot publish there, and this module +// must never try. + +import type { RealtimeSession, SessionChannel, SessionStatus } from "./session.js"; + +export type RoomStatus = SessionStatus; + +/** One other person in the room. `presence` is Partial on purpose: someone who + * just joined has set nothing yet, so every field read must survive absence — + * and under `strict` TS, the compiler enforces exactly that. */ +export type RoomPeer

> = { + /** Stable per-tab id (not a user id — works in login-less apps). */ + id: string; + /** Deterministic per-id color, identical on every viewer's screen. */ + color: string; + presence: Partial

; +}; + +export type RoomEvent = { + /** Sender's per-tab id; `bool.room.self.id` for your own echoes. */ + from: string; + data: T; +}; + +// State sends coalesce on a trailing edge so the FINAL position always lands. +// 16ms ≈ 60/s — one send per frame, the ceiling a 60fps screen can show. +const TRACK_THROTTLE_MS = 16; +// The reserved broadcast event carrying a peer's full `setMe` state. App code +// cannot use "~"-prefixed event names (broadcast() rejects them), so this can +// never collide with a real event. +const ME_EVENT = "~me"; +// When someone new joins, every peer re-sends its state once so the newcomer +// hydrates immediately instead of waiting for everyone's next move. Jittered +// so N peers don't stampede the channel in the same tick. +const HYDRATE_JITTER_MS = 300; +// Inbound coalescing: a room of movers can deliver hundreds of ~me messages a +// second; one emit per ~frame is all a screen can show anyway. +const EMIT_COALESCE_MS = 16; +// Realtime rejects large messages; failing loudly here names the actual +// problem instead of a silent server-side drop. +const MAX_PAYLOAD_BYTES = 60_000; + +/** Per-message fan-out is O(peers), so the whole room's wire cost grows with + * peers². Scale the send interval so a big room degrades to slower cursors + * instead of a saturated channel. Budgeted against the raised tenant ceiling + * on Bool's user-apps projects (10k events/s): full 60Hz through 10 people, + * ~37Hz at 15, ~21Hz at 20. */ +export function throttleForPeers(othersCount: number): number { + const n = othersCount + 1; + const budgetMs = Math.ceil((n * (n - 1)) / 8); + return Math.max(TRACK_THROTTLE_MS, budgetMs); +} + +/** Room ids become topic suffixes, so the charset is the policy's charset. + * Loud rejection beats silent normalization: a normalized id would put two + * "different" ids in one room. */ +export const ROOM_ID_RE = /^[A-Za-z0-9][A-Za-z0-9:_.-]{0,63}$/; +export function validateRoomId(id: string): string { + if (!ROOM_ID_RE.test(id)) { + throw new Error( + `bool.room("${id}"): a room id is 1–64 characters of letters, digits, ` + + `":", "_", ".", or "-" (starting with a letter or digit). ` + + `Derive it from your own data — bool.room(\`game:\${gameId}\`).`, + ); + } + return id; +} + +// 12 distinguishable hues; index by a stable hash of the peer id so every +// client computes the same color for the same peer. +const HUES = [4, 32, 56, 96, 152, 176, 200, 224, 256, 284, 312, 340]; +function hashOf(id: string): number { + let h = 0; + for (let i = 0; i < id.length; i++) h = (h * 31 + id.charCodeAt(i)) | 0; + return h; +} +export function colorForId(id: string): string { + return `hsl(${HUES[Math.abs(hashOf(id)) % HUES.length]} 85% 55%)`; +} + +// The wire limit is BYTES, and `String.length` counts UTF-16 code units — so a +// string of emoji or CJK measures at roughly half to a third of what it +// actually sends. Measuring the encoded form is the only correct check. +function byteLength(s: string): number { + if (typeof TextEncoder !== "undefined") return new TextEncoder().encode(s).length; + let bytes = 0; + for (const ch of s) { + const cp = ch.codePointAt(0)!; + bytes += cp < 0x80 ? 1 : cp < 0x800 ? 2 : cp < 0x10000 ? 3 : 4; + } + return bytes; +} + +function newTabId(): string { + try { + return crypto.randomUUID(); + } catch { + return `tab-${Math.random().toString(36).slice(2, 12)}`; + } +} + +export type RoomStore

> = { + self: { id: string; color: string }; + /** Ref-count the machinery: first acquire registers with the session, last + * release unregisters. Every React hook acquires on mount. */ + acquire(): () => void; + status(): RoomStatus; + onStatus(cb: (s: RoomStatus) => void): () => void; + /** Everyone else, newest snapshot. Stable identity when nothing changed. */ + getOthers(): ReadonlyArray>; + onOthers(cb: () => void): () => void; + /** Shallow-merge into my presence; throttled; `undefined` deletes a key. */ + setMe(patch: Record): void; + /** Remove specific keys (a hook clears its own keys on unmount). */ + clearMe(keys: string[]): void; + broadcast(event: string, data: unknown): void; + onEvent(cb: (event: string, e: RoomEvent) => void): () => void; +}; + +export type RoomStoreDeps = { + session: RealtimeSession; + /** null = the app-wide default room; a string scopes the topic. */ + roomId: string | null; + /** Create (not join) the private room channel; `key` is this tab's presence + * key. The session joins it with its own guarded status callback. */ + makeChannel(topic: string, key: string): SessionChannel; + /** Test seams for the throttle/coalesce timers; default to timers. */ + schedule?: (fn: () => void, ms: number) => unknown; + cancel?: (handle: unknown) => void; + now?: () => number; +}; + +export function createRoomStore

>(deps: RoomStoreDeps): RoomStore

{ + const schedule = deps.schedule ?? ((fn, ms) => setTimeout(fn, ms)); + const cancel = deps.cancel ?? ((h) => clearTimeout(h as ReturnType)); + const now = deps.now ?? (() => Date.now()); + + const self = { id: newTabId(), color: "" }; + self.color = colorForId(self.id); + + // ---- reactive bits -------------------------------------------------------- + let others: ReadonlyArray> = []; + const othersListeners = new Set<() => void>(); + const emitOthers = () => { + for (const l of othersListeners) l(); + }; + + let state: RoomStatus = "connecting"; + const statusListeners = new Set<(s: RoomStatus) => void>(); + function setStatus(next: RoomStatus): void { + if (state === next) return; + state = next; + for (const l of statusListeners) l(next); + } + + const eventListeners = new Set<(event: string, e: RoomEvent) => void>(); + const dispatch = (event: string, e: RoomEvent) => { + for (const l of eventListeners) l(event, e); + }; + + // ---- peers: membership (presence) ∪ state (~me broadcasts) --------------- + // Membership is authoritative — state whose owner left is dropped, which is + // what auto-clears a closed tab's cursor. Old-SDK peers still put state in + // their presence meta, so meta is overlaid under ~me state for mixed rooms. + let memberMeta = new Map>(); + const stateById = new Map>(); + + // Stable peer identities: a peer whose inputs didn't change keeps the SAME + // object between rebuilds, so a memoized for a still + // person doesn't re-render because someone ELSE moved. + const peerCache = new Map< + string, + { meta: Record; state: Record | undefined; peer: RoomPeer

} + >(); + + function rebuildOthers(): void { + const next: Array> = []; + for (const [id, meta] of memberMeta) { + if (id === self.id) continue; // others NEVER includes you + const peerState = stateById.get(id); + const cached = peerCache.get(id); + if (cached && cached.meta === meta && cached.state === peerState) { + next.push(cached.peer); + continue; + } + const presence = { ...meta, ...peerState }; + const peer: RoomPeer

= { id, color: colorForId(id), presence: presence as Partial

}; + peerCache.set(id, { meta, state: peerState, peer }); + next.push(peer); + } + for (const id of peerCache.keys()) { + if (!memberMeta.has(id)) peerCache.delete(id); + } + next.sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)); + others = next; + scheduleEmit(); + } + + // One emit per ~frame no matter how many messages landed inside it. The + // FIRST change in a quiet period emits immediately (a reaction should not + // wait 16ms); the flood behind it coalesces onto the trailing edge. + let emitTimer: unknown = null; + let lastEmitAt = -Infinity; + function scheduleEmit(): void { + const elapsed = now() - lastEmitAt; + if (elapsed >= EMIT_COALESCE_MS) { + lastEmitAt = now(); + emitOthers(); + return; + } + if (emitTimer !== null) return; + emitTimer = schedule(() => { + emitTimer = null; + lastEmitAt = now(); + emitOthers(); + }, EMIT_COALESCE_MS - elapsed); + } + + // ---- my state: merge + trailing-edge throttle, published over broadcast -- + const myState: Record = {}; + let channel: SessionChannel | null = null; + let joined = false; + let trackTimer: unknown = null; + let lastTrackAt = 0; + + /** Publish my full state now, fire-and-forget. Full state, not a delta: + * drops are healed by the next send, and a late joiner needs one message, + * not a replay. No ack — at cursor frequency, waiting on per-message acks + * is what melted the presence transport. */ + function pushMe(): void { + if (!channel || !joined) return; + lastTrackAt = now(); + channel.send(ME_EVENT, { f: self.id, s: { ...myState } }); + } + + function scheduleMe(): void { + if (!channel || !joined) return; // replayed on join instead + const throttle = throttleForPeers(others.length); + const elapsed = now() - lastTrackAt; + if (elapsed >= throttle) return pushMe(); + if (trackTimer !== null) return; // trailing edge already armed + trackTimer = schedule(() => { + trackTimer = null; + pushMe(); + }, throttle - elapsed); + } + + // One pending hydration re-send, jittered off the peer-id hash so N peers + // answering the same join don't stampede the channel in one tick. + let hydrateTimer: unknown = null; + function scheduleHydrate(): void { + if (hydrateTimer !== null) return; + if (Object.keys(myState).length === 0) return; // nothing to tell them + const jitter = 1 + (Math.abs(hashOf(self.id)) % HYDRATE_JITTER_MS); + hydrateTimer = schedule(() => { + hydrateTimer = null; + pushMe(); + }, jitter); + } + + function clearLocalTimers(): void { + for (const t of [trackTimer, hydrateTimer, emitTimer]) if (t !== null) cancel(t); + trackTimer = hydrateTimer = emitTimer = null; + } + + // ---- the channel group (all connection logic lives in the session) ------- + const group = { + topics(t: { room: string | null }): string[] { + // A platform that predates the room lane has no room topic; returning [] + // makes the session report this group "unavailable" (retried on every + // reconnect, so a platform deploy turns it on without a reload). + if (!t.room) return []; + return [deps.roomId === null ? t.room : `${t.room}:${deps.roomId}`]; + }, + open(topic: string): SessionChannel { + const ch = deps.makeChannel(topic, self.id); + ch.onPresence((states) => { + const nextMeta = new Map>(); + let sawNewMember = false; + for (const [key, metas] of Object.entries(states)) { + // Supabase keeps one meta per connection under the key; last wins. + const meta = metas[metas.length - 1] ?? {}; + const { presence_ref: _ref, ...rest } = meta as Record; + nextMeta.set(key, rest); + if (key !== self.id && !memberMeta.has(key)) sawNewMember = true; + } + for (const id of stateById.keys()) { + if (!nextMeta.has(id)) stateById.delete(id); + } + memberMeta = nextMeta; + rebuildOthers(); + // Someone new arrived: re-send my state once (jittered) so they see + // me now instead of on my next move. + if (sawNewMember) scheduleHydrate(); + }); + ch.onBroadcast((msg) => { + if (msg.event === ME_EVENT) { + const env = (msg.payload ?? {}) as { f?: string; s?: Record }; + if (!env.f || env.f === self.id) return; + // State can outrun membership by a beat (broadcast delivers before + // the presence diff); hold it either way — rebuildOthers keys off + // membership, so it shows the moment the member appears. + stateById.set(env.f, env.s ?? {}); + if (memberMeta.has(env.f)) rebuildOthers(); + return; + } + const env = (msg.payload ?? {}) as { f?: string; d?: unknown }; + if (env.f === self.id) return; // already delivered locally, synchronously + dispatch(msg.event, { from: env.f ?? "", data: env.d }); + }); + channel = ch; + return ch; + }, + onJoin(): void { + joined = true; + // ONE membership beacon per join — presence carries who's-here, never + // state. At one-per-join a failure can't melt into the reconnect storm + // the per-move version caused; a genuinely dead channel reports its own + // terminal state, which the session heals. + channel?.track({}, (result) => { + if (result !== "ok" && typeof console !== "undefined") { + console.warn("bool.room: presence join beacon failed; peers may not see you until the next reconnect."); + } + }); + // Join/rejoin replays my current state so a reconnect (or a late first + // join) never leaves me invisible. + pushMe(); + }, + onDown(): void { + channel = null; + joined = false; + clearLocalTimers(); + memberMeta = new Map(); + stateById.clear(); + peerCache.clear(); + if (others.length > 0) { + others = []; + // Direct, not coalesced: "the room emptied" must never be absorbed + // into a cancelled trailing edge. + emitOthers(); + } + }, + onStatus(s: SessionStatus): void { + setStatus(s); + }, + }; + + // ---- public surface -------------------------------------------------------- + let holds = 0; + let unregister: (() => void) | null = null; + + return { + self, + acquire() { + holds++; + if (holds === 1) unregister = deps.session.register(group); + let released = false; + return () => { + if (released) return; + released = true; + holds--; + if (holds === 0) { + unregister?.(); + unregister = null; + setStatus("connecting"); + } + }; + }, + status: () => state, + onStatus(cb) { + statusListeners.add(cb); + return () => statusListeners.delete(cb); + }, + getOthers: () => others, + onOthers(cb) { + othersListeners.add(cb); + return () => othersListeners.delete(cb); + }, + setMe(patch) { + for (const [k, v] of Object.entries(patch)) { + if (v === undefined) delete myState[k]; + else myState[k] = v; + } + scheduleMe(); + }, + clearMe(keys) { + let changed = false; + for (const k of keys) { + if (k in myState) { + delete myState[k]; + changed = true; + } + } + if (changed) scheduleMe(); + }, + broadcast(event, data) { + if (event.startsWith("~")) { + // "~" names the SDK's own wire events (~me carries setMe state). A + // user event on that namespace would be read back as peer state. + throw new Error( + `bool.room.broadcast("${event}"): event names starting with "~" are reserved for the SDK. Pick another name.`, + ); + } + const envelope = { f: self.id, d: data }; + const size = byteLength(JSON.stringify(envelope) ?? ""); + if (size > MAX_PAYLOAD_BYTES) { + throw new Error( + `bool.room.broadcast("${event}") payload is ${size} bytes — over the ${MAX_PAYLOAD_BYTES}-byte realtime limit. ` + + `Send a reference (an id) instead of the thing itself, or save the thing via bool.entities.`, + ); + } + // Local echo FIRST, synchronously: your own reaction never waits for the + // network, and a one-tab app behaves identically to a ten-tab one. + dispatch(event, { from: self.id, data }); + channel?.send(event, envelope); + }, + onEvent(cb) { + eventListeners.add(cb); + return () => eventListeners.delete(cb); + }, + }; +} + +// --------------------------------------------------------------------------- +// Public surface + React registration. +// +// Core stays React-free (this module never imports React). The hooks below +// delegate to an implementation that importing "bool-sdk/react" registers — +// the same pattern, and the same hard-won lesson, as entities' useQuery: the +// react entry is loaded via a USED BINDING in the seeded client file, because +// a bare side-effect import binds no names and bundlers delete it from +// production builds (that shipped once, as 0.3.0, and broke every published +// app while previews looked fine). +// +// Deliberately NO string index signature on BoolRoom: a hallucinated member +// (`useStorage`, `useMyPresence`, `RoomProvider`…) must be a build error, not +// an `undefined is not a function` in front of a user. + +export type BoolRoom = { + /** This tab's identity: a per-tab id (not a user id — login-less apps have + * cursors too) and the color every OTHER viewer will show for you. */ + self: { id: string; color: string }; + /** React hook: everyone else in the room, live. Hydrated on join, + * auto-cleared when someone disconnects, `[]` until connected. Peers only — + * never you. */ + useOthers

>(): ReadonlyArray>; + /** React hook returning the presence setter. Shallow-merges, throttled + * inside the SDK (never throttle it yourself), and the keys this component + * set are cleared when it unmounts — so a cursor never outlives its screen. + * Call it straight from the raw event handler; putting pointer positions in + * React state re-renders per mousemove. */ + useSetMe

>(): (patch: Partial

) => void; + /** One-shot to everyone in the room — including this tab, synchronously, so + * your own reaction never waits for the network. Fire-and-forget: returns + * void, delivery is best-effort, nothing is stored. */ + broadcast(event: string, data: unknown): void; + /** React hook: receive one named event. Subscription lifecycle, StrictMode + * safety and latest-callback semantics are handled inside. */ + useEventListener(event: string, cb: (e: RoomEvent) => void): void; + /** React hook: honest delivery state. `[]` others + `"unavailable"` means + * "can't know who's here", not "alone" — say so instead of guessing. */ + useStatus(): RoomStatus; +}; + +/** The callable default room: `bool.room.useOthers()` is the app-wide room; + * `bool.room("game:4")` is a scoped one with the identical surface. Same id → + * same room instance (and one shared socket underneath them all). */ +export type BoolRoomApi = BoolRoom & { + (id: string): BoolRoom; +}; + +type RoomHooksImpl = { + useOthers(store: RoomStore>): ReadonlyArray>>; + useSetMe(store: RoomStore>): (patch: Record) => void; + useEventListener( + store: RoomStore>, + event: string, + cb: (e: RoomEvent) => void, + ): void; + useStatus(store: RoomStore>): RoomStatus; +}; + +let roomHooksImpl: RoomHooksImpl | null = null; +/** Registered by "bool-sdk/react" at import time. Returns the previous impl so + * tests can restore it (module state is process-global under bun test). */ +export function __registerRoomHooks(impl: RoomHooksImpl | null): RoomHooksImpl | null { + const prev = roomHooksImpl; + roomHooksImpl = impl; + return prev; +} + +function requireHooks(member: string): RoomHooksImpl { + if (!roomHooksImpl) { + throw new Error( + `bool.room.${member}() needs the React entry loaded once per app — ` + + `add \`import { createBoolClient } from "bool-sdk/react";\` (Bool apps already create ` + + `their client from that entry in src/lib/supabase.ts). Outside React, bool.room has no surface yet.`, + ); + } + return roomHooksImpl; +} + +let warnedUnmountedBroadcast = false; + +/** Bind one store to the public `bool.room` shape. */ +export function createBoolRoom(store: RoomStore>): BoolRoom { + return { + self: store.self, + useOthers

>() { + return requireHooks("useOthers").useOthers(store) as ReadonlyArray>; + }, + useSetMe

>() { + return requireHooks("useSetMe").useSetMe(store) as (patch: Partial

) => void; + }, + broadcast(event, data) { + if (store.status() !== "live" && !warnedUnmountedBroadcast) { + warnedUnmountedBroadcast = true; + console.warn( + `bool.room.broadcast("${event}"): the room isn't connected yet, so only this tab saw it. ` + + `The room connects while a bool.room hook is mounted (useOthers / useEventListener / useStatus).`, + ); + } + store.broadcast(event, data); + }, + useEventListener(event: string, cb: (e: RoomEvent) => void) { + requireHooks("useEventListener").useEventListener(store, event, cb as (e: RoomEvent) => void); + }, + useStatus() { + return requireHooks("useStatus").useStatus(store); + }, + }; +} + +/** Build the callable `bool.room`: the default room's members on the function + * itself, scoped rooms (memoized by id) on invocation. */ +export function createBoolRoomApi( + defaultRoom: BoolRoom, + storeFor: (id: string) => RoomStore>, +): BoolRoomApi { + const scoped = new Map(); + const fn = ((id: string): BoolRoom => { + validateRoomId(id); + let room = scoped.get(id); + if (!room) { + room = createBoolRoom(storeFor(id)); + scoped.set(id, room); + } + return room; + }) as BoolRoomApi; + fn.self = defaultRoom.self; + fn.useOthers = defaultRoom.useOthers; + fn.useSetMe = defaultRoom.useSetMe; + fn.broadcast = defaultRoom.broadcast; + fn.useEventListener = defaultRoom.useEventListener; + fn.useStatus = defaultRoom.useStatus; + return fn; +} diff --git a/src/session.test.ts b/src/session.test.ts new file mode 100644 index 0000000..e8f60ed --- /dev/null +++ b/src/session.test.ts @@ -0,0 +1,368 @@ +import { describe, expect, test } from "bun:test"; +import { + createSession, + type ChannelGroup, + type SessionChannel, + type SessionMintResult, + type SessionStatus, +} from "./session"; + +// One test per scar. Each of these failures happened in production during the +// week the two pre-session machines existed; the session exists so each guard +// is written (and tested) exactly once. + +const TOPICS = { app: "bool:app_x:app", user: "bool:app_x:user:u1", room: "bool:app_x:room" }; + +function makeHarness(opts?: { + mints?: SessionMintResult[]; + /** leave() fires the join callback with CLOSED synchronously — what + * supabase's removeChannel actually does (verified against the real client). */ + syncClosedOnLeave?: boolean; +}) { + const mints = [...(opts?.mints ?? [])]; + let clock = 0; + const timers: Array<{ at: number; fn: () => void; id: number }> = []; + let nextId = 1; + const wakeCbs = new Set<() => void>(); + const h = { + mintCalls: 0, + authed: [] as string[], + channels: [] as Array<{ + topic: string; + joinCb: ((s: string) => void) | null; + left: boolean; + sent: Array<{ event: string; payload: unknown }>; + }>, + statuses: [] as SessionStatus[], + clockNow: () => clock, + wake: () => { + for (const cb of [...wakeCbs]) cb(); + }, + wakeSubs: () => wakeCbs.size, + /** Run everything due within the next `ms` virtual milliseconds. */ + advance: async (ms: number) => { + const target = clock + ms; + for (;;) { + timers.sort((a, b) => a.at - b.at); + const due = timers[0]; + if (!due || due.at > target) break; + clock = due.at; + timers.shift(); + due.fn(); + for (let i = 0; i < 20; i++) await Promise.resolve(); + } + clock = target; + for (let i = 0; i < 20; i++) await Promise.resolve(); + }, + live: () => h.channels.filter((c) => !c.left), + }; + + const session = createSession({ + async mint() { + h.mintCalls++; + return mints.length + ? mints.shift()! + : { ok: true, token: `tok-${h.mintCalls}`, expiresIn: 900, topics: TOPICS }; + }, + setAuth(token) { + h.authed.push(token); + }, + onStatus: (s) => h.statuses.push(s), + schedule: (fn, ms) => { + const id = nextId++; + timers.push({ at: clock + ms, fn, id }); + return id; + }, + cancel: (handle) => { + const i = timers.findIndex((t) => t.id === handle); + if (i !== -1) timers.splice(i, 1); + }, + now: () => clock, + wake: (cb) => { + wakeCbs.add(cb); + return () => wakeCbs.delete(cb); + }, + }); + + function makeGroup( + topicsFor: (t: typeof TOPICS) => string[], + hooks?: Partial>, + ): ChannelGroup { + return { + topics: topicsFor as ChannelGroup["topics"], + open(topic): SessionChannel { + const rec = { + topic, + joinCb: null as ((s: string) => void) | null, + left: false, + sent: [] as Array<{ event: string; payload: unknown }>, + }; + h.channels.push(rec); + return { + track: (_s, done) => done?.("ok"), + onPresence: () => {}, + onBroadcast: () => {}, + send: (event, payload, done) => { + rec.sent.push({ event, payload }); + done?.("ok"); + }, + join: (cb) => { + rec.joinCb = cb; + }, + leave: () => { + rec.left = true; + if (opts?.syncClosedOnLeave) rec.joinCb?.("CLOSED"); + }, + }; + }, + ...hooks, + }; + } + + /** Drive every un-joined channel to SUBSCRIBED. */ + const joinAll = async () => { + for (let i = 0; i < 50 && h.live().some((c) => !c.joinCb); i++) await Promise.resolve(); + for (const c of h.live()) c.joinCb?.("SUBSCRIBED"); + await Promise.resolve(); + }; + + return { session, h, makeGroup, joinAll }; +} + +describe("session lifecycle", () => { + test("mints once, auths, opens every group's channels, joins them", async () => { + const { session, h, makeGroup, joinAll } = makeHarness(); + const doorbell = makeGroup((t) => [t.app, t.user!]); + const room = makeGroup((t) => [t.room!]); + const off1 = session.register(doorbell); + const off2 = session.register(room); + await h.advance(0); + await joinAll(); + expect(h.mintCalls).toBe(1); // ONE mint serves every consumer + expect(h.authed).toEqual(["tok-1"]); + expect(h.live().map((c) => c.topic)).toEqual([ + "bool:app_x:app", + "bool:app_x:user:u1", + "bool:app_x:room", + ]); + expect(session.status()).toBe("live"); + off1(); + off2(); + }); + + test("a group registered mid-flight rides the current mint (no second handshake)", async () => { + const { session, h, makeGroup, joinAll } = makeHarness(); + const off1 = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + await joinAll(); + const before = h.mintCalls; + const off2 = session.register(makeGroup((t) => [`${t.room!}:board-2`])); + await h.advance(0); + expect(h.mintCalls).toBe(before); // opened a second named room for free + expect(h.live().map((c) => c.topic)).toContain("bool:app_x:room:board-2"); + off1(); + off2(); + }); + + test("a group whose topic the mint lacks reports unavailable, not a limp", async () => { + const { session, h, makeGroup, joinAll } = makeHarness({ + mints: [ + { ok: true, token: "t", expiresIn: 900, topics: { ...TOPICS, room: null } }, + ], + }); + const seen: SessionStatus[] = []; + const room = makeGroup((t) => (t.room ? [t.room] : []), { onStatus: (s) => seen.push(s) }); + const off = session.register(room); + await h.advance(0); + await joinAll(); + expect(seen.at(-1)).toBe("unavailable"); + off(); + }); + + test("unregistering one group closes only its channels", async () => { + const { session, h, makeGroup, joinAll } = makeHarness(); + const off1 = session.register(makeGroup((t) => [t.app])); + const off2 = session.register(makeGroup((t) => [t.room!])); + await h.advance(0); + await joinAll(); + off2(); + expect(h.live().map((c) => c.topic)).toEqual(["bool:app_x:app"]); + expect(session.status()).toBe("live"); // the other consumer is untouched + off1(); + }); +}); + +describe("scar: synchronous CLOSED from inside leave()", () => { + // supabase's removeChannel fires the channel's status callback with CLOSED + // synchronously. The pre-session machines recursed teardown → leave → CLOSED + // → teardown to a stack overflow on every real drop of a healthy channel. + test("a real drop does not recurse, and the session still recovers", async () => { + const { session, h, makeGroup, joinAll } = makeHarness({ syncClosedOnLeave: true }); + const off = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + await joinAll(); + expect(session.status()).toBe("live"); + + h.live()[0]!.joinCb!("CLOSED"); // pre-fix: RangeError, max call stack + expect(session.status()).toBe("unavailable"); + + const channelsBefore = h.channels.length; + await h.advance(1_000); // first retry rung + expect(h.channels.length).toBe(channelsBefore + 1); // fresh channel opened + await joinAll(); + expect(session.status()).toBe("live"); + off(); + }); + + test("unregister with a live channel does not recurse either, and stays quiet", async () => { + const { session, h, makeGroup, joinAll } = makeHarness({ syncClosedOnLeave: true }); + const off = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + await joinAll(); + off(); // teardown → leave → sync CLOSED, on the unmount path + const before = h.channels.length; + await h.advance(120_000); + expect(h.channels.length).toBe(before); // nothing keeps retrying + }); +}); + +describe("scar: waking up", () => { + test("a wake reconnects immediately instead of waiting out the backoff", async () => { + const { session, h, makeGroup, joinAll } = makeHarness(); + const off = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + await joinAll(); + // drop, then keep dropping every fresh channel so the ladder climbs to + // the 60s rung + h.live()[0]!.joinCb!("CLOSED"); + for (let i = 0; i < 3; i++) { + await h.advance(60_000); // the pending rung fires, a fresh channel opens + const fresh = h.live()[0]; + fresh?.joinCb?.("CLOSED"); + await h.advance(0); + } + expect(session.status()).toBe("unavailable"); + const before = h.mintCalls; + h.wake(); // no clock advance: must not be waiting on any timer + await h.advance(0); + expect(h.mintCalls).toBe(before + 1); + await joinAll(); + expect(session.status()).toBe("live"); + off(); + }); + + test("a wake while live does nothing, and simultaneous wakes coalesce", async () => { + const { session, h, makeGroup, joinAll } = makeHarness(); + const off = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + await joinAll(); + const atLive = h.mintCalls; + h.wake(); + await h.advance(0); + expect(h.mintCalls).toBe(atLive); // no churn on tab focus + + h.live()[0]!.joinCb!("CLOSED"); + await h.advance(0); + const beforeWakes = h.mintCalls; + h.wake(); + h.wake(); + h.wake(); // focus + visibilitychange + online, same instant + await h.advance(0); + expect(h.mintCalls).toBe(beforeWakes + 1); + off(); + }); + + test("wake listeners are released with the last group (no zombie reconnects)", async () => { + const { session, h, makeGroup } = makeHarness(); + expect(h.wakeSubs()).toBe(0); + const off1 = session.register(makeGroup((t) => [t.app])); + const off2 = session.register(makeGroup((t) => [t.room!])); + expect(h.wakeSubs()).toBe(1); // one subscription for the whole session + off1(); + expect(h.wakeSubs()).toBe(1); + off2(); + expect(h.wakeSubs()).toBe(0); + const before = h.mintCalls; + h.wake(); + await h.advance(100); + expect(h.mintCalls).toBe(before); + }); +}); + +describe("scar: the wristband desk", () => { + test("a hung mint cannot wedge the session forever", async () => { + const { session, h, makeGroup } = makeHarness({ + mints: [new Promise(() => {}) as never], // never settles + }); + // a mint that never resolves isn't representable via the mints queue; + // inject directly instead: + const s2 = createSession({ + mint: () => new Promise(() => {}), + setAuth: () => {}, + schedule: (fn, ms) => setTimeout(fn, ms / 1000), // compress time 1000× + cancel: (h2) => clearTimeout(h2 as ReturnType), + wake: () => () => {}, + }); + const off = s2.register(makeGroup((t) => [t.app])); + await new Promise((r) => setTimeout(r, 25)); // > compressed MINT_TIMEOUT + expect(s2.status()).toBe("unavailable"); + off(); + void session; + void h; + }); + + test("refresh re-auths in place — channels do NOT rejoin", async () => { + const { session, h, makeGroup, joinAll } = makeHarness(); + const off = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + await joinAll(); + const channelsBefore = h.channels.length; + await h.advance(900 * 1000 * 0.75 + 1); // the refresh rung + expect(h.authed).toEqual(["tok-1", "tok-2"]); + expect(h.channels.length).toBe(channelsBefore); // same channels, new token + expect(session.status()).toBe("live"); + off(); + }); + + test("a refused re-mint drops every channel and reports unauthorized", async () => { + // The TTL expiring IS the revocation mechanism: once the desk stops + // issuing wristbands the socket must go quiet, not keep listening. + const { session, h, makeGroup, joinAll } = makeHarness({ + mints: [ + { ok: true, token: "tok-1", expiresIn: 900, topics: TOPICS }, + { ok: false, reason: "unauthorized" }, + ], + }); + const off = session.register(makeGroup((t) => [t.app, t.user!])); + await h.advance(0); + await joinAll(); + await h.advance(900 * 1000 * 0.75 + 1); + expect(h.live()).toHaveLength(0); + expect(session.status()).toBe("unauthorized"); + off(); + }); + + test("an unreachable desk reports unavailable — NEVER unauthorized", async () => { + // A public app admits every visitor; a dropped request must not be + // presented as "you aren't allowed to watch this". + const { session, h, makeGroup } = makeHarness({ + mints: [{ ok: false, reason: "unavailable" }], + }); + const off = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + expect(session.status()).toBe("unavailable"); + off(); + }); + + test("the last unregister NOTIFIES the status reset (no stale live snapshot)", async () => { + const { session, h, makeGroup, joinAll } = makeHarness(); + const off = session.register(makeGroup((t) => [t.app])); + await h.advance(0); + await joinAll(); + expect(session.status()).toBe("live"); + h.statuses.length = 0; + off(); + expect(session.status()).toBe("connecting"); + expect(h.statuses).toEqual(["connecting"]); // the notification, not just the field + }); +}); diff --git a/src/session.ts b/src/session.ts new file mode 100644 index 0000000..b0432f2 --- /dev/null +++ b/src/session.ts @@ -0,0 +1,328 @@ +// RealtimeSession — the ONE connection machine for everything realtime. +// +// Before this file existed, the entity doorbell (realtime.ts) and the room +// (room.ts) each carried their own copy of the connection lifecycle: mint a +// wristband, present it, open channels, refresh before the TTL, back off on +// failure, reconnect on wake, tear down without racing yourself. The copies +// drifted, and in one 24-hour stretch (2026-07-29) the two worst production +// bugs — the synchronous-CLOSED teardown recursion and the missing +// wake-reconnect — each had to be diagnosed and fixed TWICE, once per copy. +// This file is those fixes written once. The doorbell and every room are now +// pure consumers that register a ChannelGroup and contain zero connection code. +// +// Every guard below is a scar from a specific production failure. Deleting one +// re-earns it the hard way; the comment on each names the failure it prevents. + +import { onWake } from "./wake.js"; + +export type SessionStatus = "connecting" | "live" | "unauthorized" | "unavailable"; + +/** The topics one wristband admits. `user` is null for anonymous visitors; + * `room` is null on a platform that predates the room lane. */ +export type RealtimeTopics = { + app: string; + user: string | null; + room: string | null; +}; + +export type SessionMintResult = + | { ok: true; token: string; expiresIn: number; topics: RealtimeTopics } + | { ok: false; reason: "unauthorized" | "unavailable" }; + +/** The transport seam for one channel — everything effectful is injected so + * the machine is unit-testable without sockets. Consumers create the channel + * (they know their own presence/broadcast config); the session owns join, + * teardown, and every status transition. */ +export type SessionChannel = { + track(state: Record, done?: (result: string) => void): void; + onPresence(cb: (states: Record>>) => void): void; + onBroadcast(cb: (msg: { event: string; payload: unknown }) => void): void; + send(event: string, payload: unknown, done?: (result: string) => void): void; + join(status: (state: string) => void): void; + leave(): void; +}; + +/** One consumer's channels. The doorbell is one group (app + user topics); + * every room — default or named — is its own group. Groups share the session's + * single mint/auth/socket lifecycle. */ +export type ChannelGroup = { + /** Which topics this group wants under the current mint. Return [] when the + * mint can't serve this group (e.g. no room topic on an old platform) — the + * group is reported "unavailable" and re-asked on every reconnect/refresh, + * so a platform deploy can turn it on without a reload. */ + topics(t: RealtimeTopics): string[]; + /** Create ONE channel for `topic` and attach message handlers. Do NOT join — + * the session joins with its own guarded status callback. */ + open(topic: string): SessionChannel; + /** One of this group's channels reached SUBSCRIBED (initial join or rejoin + * after a drop). Re-publish anything a fresh channel must carry (the room's + * membership beacon + current state live here). */ + onJoin?(ch: SessionChannel, topic: string): void; + /** This group's channels are gone (teardown, restart, or unregister). */ + onDown?(): void; + /** This GROUP's delivery status: the session's status, degraded to + * "connecting" until every channel this group asked for has joined, and to + * "unavailable" when the mint has no topic for it. */ + onStatus?(s: SessionStatus): void; +}; + +export type SessionDeps = { + mint(): Promise; + setAuth(token: string): Promise | void; + onStatus?: (s: SessionStatus) => void; + /** Test seams; default to timers/Date/DOM wake listeners. */ + schedule?: (fn: () => void, ms: number) => unknown; + cancel?: (handle: unknown) => void; + now?: () => number; + wake?: (cb: () => void) => () => void; +}; + +export type RealtimeSession = { + /** Add a group. Ref-counted: the first registration starts the machine, the + * last unregister stops it. Registering mid-flight opens the group's + * channels under the current mint immediately. Returns unregister. */ + register(group: ChannelGroup): () => void; + status(): SessionStatus; + onStatus(cb: (s: SessionStatus) => void): () => void; +}; + +// Backoff after a refused/failed mint or a dropped channel: quick first retry, +// capped so a long outage costs one request a minute rather than a hot loop. +const RETRY_MS = [1_000, 5_000, 15_000, 60_000]; +// A mint that never settles would wedge the machine — no channel, no status +// change, no retry — indistinguishable from "live" to anything watching. Cap +// it and treat a stall as an ordinary "unavailable", which retries. +const MINT_TIMEOUT_MS = 10_000; +// Re-mint at 75% of the wristband TTL: early enough that a slow mint never +// races token expiry (which closes channels underneath us), late enough that +// refresh traffic stays negligible. +const REFRESH_FRACTION = 0.75; +// A desk that answers with a nonsense TTL must not melt into a mint loop. +const MIN_REFRESH_MS = 30_000; +// `focus` + `visibilitychange` fire together on tab re-selection, and `online` +// can pile on; one wake must be one reconnect, not three. +const WAKE_COALESCE_MS = 500; + +type OpenChannel = { ch: SessionChannel; topic: string; joined: boolean }; + +export function createSession(deps: SessionDeps): RealtimeSession { + const schedule = deps.schedule ?? ((fn, ms) => setTimeout(fn, ms)); + const cancel = deps.cancel ?? ((h) => clearTimeout(h as ReturnType)); + const now = deps.now ?? (() => Date.now()); + const subscribeWake = deps.wake ?? onWake; + + // One "generation" per start(); every async continuation and every channel + // status callback checks it, so a teardown or restart orphans in-flight work + // instead of racing it. + let gen = 0; + let attempt = 0; + let timer: unknown = null; + let state: SessionStatus = "connecting"; + let topics: RealtimeTopics | null = null; + + const groups = new Set(); + const open = new Map(); + + const statusListeners = new Set<(s: SessionStatus) => void>(); + function setStatus(next: SessionStatus): void { + if (state === next) return; + // setStatus, never a bare assignment: a direct write leaves every + // subscriber holding the stale value — the "torn-down but still says + // live" bug, found independently in BOTH pre-session machines. + state = next; + deps.onStatus?.(next); + for (const l of statusListeners) l(next); + for (const g of groups) emitGroupStatus(g); + } + + /** A group's status is the session's, degraded by its own facts: no topic in + * the mint → unavailable; channels not all joined yet → connecting. */ + function groupStatus(g: ChannelGroup): SessionStatus { + if (state !== "live") return state; + const chans = open.get(g); + if (!chans) return "connecting"; + if (chans.length === 0) return "unavailable"; // mint has no topic for it + return chans.every((c) => c.joined) ? "live" : "connecting"; + } + function emitGroupStatus(g: ChannelGroup): void { + g.onStatus?.(groupStatus(g)); + } + + function clearTimer(): void { + if (timer !== null) { + cancel(timer); + timer = null; + } + } + + /** Close one group's channels. Detach BEFORE leaving: supabase's + * removeChannel fires the channel's own status callback with CLOSED — + * synchronously, from inside leave() — and the terminal-state handler tears + * down in response. Leave-then-detach recursed teardown → leave → CLOSED → + * teardown to a stack overflow on every real network drop (observed in + * production as an endless "Maximum call stack size exceeded"). The + * membership check in the join callback is the other half of this fix. */ + function closeGroup(g: ChannelGroup): void { + const chans = open.get(g); + if (!chans) return; + open.delete(g); + for (const c of chans) c.ch.leave(); + g.onDown?.(); + } + + function teardownAll(): void { + for (const g of [...open.keys()]) closeGroup(g); + clearTimer(); + } + + function retry(myGen: number, why: SessionStatus): void { + setStatus(why); + const ms = RETRY_MS[Math.min(attempt, RETRY_MS.length - 1)]!; + attempt++; + timer = schedule(() => { + if (myGen !== gen) return; + void start(myGen); + }, ms); + } + + /** Open one group's channels under the current mint and join them. Safe to + * call for a group registered mid-flight. */ + function openGroup(g: ChannelGroup, myGen: number): void { + if (!topics || open.has(g)) return; + const wanted = g.topics(topics); + const chans: OpenChannel[] = []; + open.set(g, chans); + for (const topic of wanted) { + const ch = g.open(topic); + const entry: OpenChannel = { ch, topic, joined: false }; + chans.push(entry); + ch.join((s) => { + // Membership, not just generation: closeGroup() empties the group's + // entry and then leave() makes THIS callback fire CLOSED in the same + // tick with the same gen. Without the check, that re-entered the + // terminal branch and recursed to a stack overflow. + if (myGen !== gen || open.get(g) !== chans) return; + if (s === "SUBSCRIBED") { + entry.joined = true; + attempt = 0; // a good join resets the backoff ladder + setStatus("live"); + emitGroupStatus(g); + g.onJoin?.(ch, topic); + } else if (s === "CHANNEL_ERROR" || s === "TIMED_OUT" || s === "CLOSED") { + // Every terminal state must drive recovery: supabase-js does NOT + // auto-rejoin, and leaving one unhandled (CLOSED, originally) meant + // a dropped socket left delivery dead forever while the last + // reported status stayed "live". One channel down restarts the + // whole session: everything shares one socket and one token, so + // partial restarts buy complexity, not resilience. + teardownAll(); + retry(myGen, "unavailable"); + } + }); + } + emitGroupStatus(g); + } + + async function start(myGen: number): Promise { + teardownAll(); + if (myGen !== gen) return; + + const res = await Promise.race([ + deps.mint(), + new Promise((resolve) => + schedule(() => resolve({ ok: false, reason: "unavailable" }), MINT_TIMEOUT_MS), + ), + ]); + if (myGen !== gen) return; + // No wristband: nothing is live. Report WHICH kind of "no" — a public app + // where every visitor is normally admitted must never be told it's + // unauthorized because of a dropped request — and retry. Never fake it. + if (!res.ok) return retry(myGen, res.reason); + + await deps.setAuth(res.token); + if (myGen !== gen) return; + + topics = res.topics; + setStatus("live"); // mint-level; groups stay "connecting" until joined + for (const g of groups) openGroup(g, myGen); + scheduleRefresh(myGen, res.expiresIn); + } + + function scheduleRefresh(myGen: number, expiresInSeconds: number): void { + const ms = Math.max(expiresInSeconds * 1000 * REFRESH_FRACTION, MIN_REFRESH_MS); + timer = schedule(async () => { + if (myGen !== gen) return; + const res = await deps.mint(); + if (myGen !== gen) return; + // Renewal refused — access revoked, or the gateway is down. The TTL + // expiring IS the revocation mechanism, so drop the channels. + if (!res.ok) { + teardownAll(); + return retry(myGen, res.reason); + } + await deps.setAuth(res.token); // channels re-auth in place, no rejoin + if (myGen !== gen) return; + topics = res.topics; + scheduleRefresh(myGen, res.expiresIn); + }, ms); + } + + // ---- wake: come back the instant the tab/network does --------------------- + // Browsers throttle background tabs and drop idle sockets; the drop is + // reported honestly, but waiting out a [1s,5s,15s,60s] ladder after the + // ordinary switch-tabs-and-return means up to a minute of "offline" on a + // page the user is actively looking at — indistinguishable from broken. + let stopWake: (() => void) | null = null; + let lastWakeAt = -Infinity; + function onWakeSignal(): void { + // Already live: the socket is fine — reconnecting on every tab focus would + // churn. A socket that is secretly dead reports its own terminal state, + // which lands in retry(), which the next wake (or the 1s rung) picks up. + if (groups.size === 0 || state === "live") return; + const t = now(); + if (t - lastWakeAt < WAKE_COALESCE_MS) return; + lastWakeAt = t; + gen++; // orphan the pending retry and any in-flight start + attempt = 0; + setStatus("connecting"); + void start(gen); + } + + return { + register(group) { + groups.add(group); + if (groups.size === 1) { + gen++; + attempt = 0; + setStatus("connecting"); + stopWake = subscribeWake(onWakeSignal); + void start(gen); + } else if (state === "live") { + // Joined mid-flight (a second room opened): ride the current mint. + openGroup(group, gen); + } else { + emitGroupStatus(group); + } + let released = false; + return () => { + if (released) return; + released = true; + if (!groups.delete(group)) return; + closeGroup(group); + if (groups.size === 0) { + gen++; // orphan any in-flight start/refresh + stopWake?.(); + stopWake = null; + teardownAll(); + topics = null; + setStatus("connecting"); + } + }; + }, + status: () => state, + onStatus(cb) { + statusListeners.add(cb); + return () => statusListeners.delete(cb); + }, + }; +} diff --git a/src/wake.ts b/src/wake.ts new file mode 100644 index 0000000..e9d6440 --- /dev/null +++ b/src/wake.ts @@ -0,0 +1,53 @@ +/** + * "The environment just came back" — the signal that a dropped realtime + * connection should retry NOW instead of waiting out its backoff. + * + * Why this exists: the single most common realtime failure is also the most + * mundane one. A backgrounded tab gets its timers throttled and its idle socket + * dropped, so switching away and coming back finds the connection dead. The + * drop itself IS reported (the channel reports a terminal state on return, which + * lands in retry()), so recovery does happen — but on a [1s, 5s, 15s, 60s] + * ladder it can take a full minute, and a minute of "offline" on a page the user + * is actively looking at is indistinguishable from broken. + * + * Observed on a real deployed app: a tab left in the background reported OFFLINE + * honestly and then took ~30s to come back. Every signal here resets the ladder + * and reconnects immediately, which turns that into an unnoticeable blip. + * + * Both realtime machines (the entity doorbell and the room) subscribe to this; + * neither had any wake handling before, and the gap was identical in both. + */ +export type WakeUnsubscribe = () => void; + +/** Call `cb` whenever the tab/network wakes up. Returns unsubscribe. + * A no-op (and safe) outside a DOM — server rendering, Node, tests. */ +export function onWake(cb: () => void): WakeUnsubscribe { + const offs: WakeUnsubscribe[] = []; + + const listen = (target: EventTarget | undefined, type: string, handler: () => void): void => { + if (!target || typeof target.addEventListener !== "function") return; + target.addEventListener(type, handler); + offs.push(() => target.removeEventListener(type, handler)); + }; + + const doc = typeof document === "undefined" ? undefined : document; + const win = typeof window === "undefined" ? undefined : window; + + // Becoming visible is a wake; going hidden is not. Firing on hide would + // reconnect a socket nobody is watching, only for it to be dropped again. + listen(doc, "visibilitychange", () => { + if (!doc || doc.visibilityState === "visible") cb(); + }); + // The network came back — the other way a connection dies without the page + // ever being hidden. + listen(win, "online", cb); + // A desktop window that was never `hidden` (just covered by another app) can + // still have had its socket time out. `focus` catches that case; it fires + // alongside `visibilitychange`, which is why callers coalesce. + listen(win, "focus", cb); + + return () => { + for (const off of offs) off(); + offs.length = 0; + }; +}