diff --git a/web/pnpm-lock.yaml b/web/pnpm-lock.yaml index 1f1d842..e52ac0d 100644 --- a/web/pnpm-lock.yaml +++ b/web/pnpm-lock.yaml @@ -119,6 +119,9 @@ importers: emoji-mart: specifier: ^5.6.0 version: 5.6.0 + lib0: + specifier: ^0.2.99 + version: 0.2.117 lucide-react: specifier: ^0.469.0 version: 0.469.0(react@18.3.1) diff --git a/web/standalone/package.json b/web/standalone/package.json index 813a762..aa8720d 100644 --- a/web/standalone/package.json +++ b/web/standalone/package.json @@ -29,6 +29,7 @@ "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "emoji-mart": "^5.6.0", + "lib0": "^0.2.99", "lucide-react": "^0.469.0", "react": "^18.3.1", "react-dom": "^18.3.1", diff --git a/web/standalone/src/wasm/collab/gateway.test.ts b/web/standalone/src/wasm/collab/gateway.test.ts new file mode 100644 index 0000000..60e50e5 --- /dev/null +++ b/web/standalone/src/wasm/collab/gateway.test.ts @@ -0,0 +1,337 @@ +import * as decoding from "lib0/decoding"; +import * as encoding from "lib0/encoding"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import * as syncProtocol from "y-protocols/sync"; +import * as Y from "yjs"; +import { + parseGatewayClientMsg, + tagGatewayFrame, + untagGatewayFrame, + type GatewayClientMsg, +} from "@pcbjam/shared"; +import { CollabSubRejectedError, GatewayDocFacade } from "./gateway"; + +/** + * The gateway facade is a mini y-websocket client over one shared multiplexed + * socket (load-path-rework 0003 §6). These tests script the socket directly: + * what a passive vs active subscription puts on the wire is the laziness + * contract — a passive warm-pool sheet must never emit a doc frame. + */ + +// --- scripted WebSocket ------------------------------------------------------ + +class FakeWebSocket { + static instances: FakeWebSocket[] = []; + static OPEN = 1; + static CLOSED = 3; + readyState = 0; + binaryType = "blob"; + sent: Array = []; + onopen: (() => void) | null = null; + onmessage: ((e: { data: unknown }) => void) | null = null; + onclose: (() => void) | null = null; + onerror: (() => void) | null = null; + constructor(public url: string) { + FakeWebSocket.instances.push(this); + } + send(data: string | ArrayBuffer | Uint8Array): void { + this.sent.push(data); + } + close(): void { + this.readyState = FakeWebSocket.CLOSED; + this.onclose?.(); + } + // test drivers + open(): void { + this.readyState = FakeWebSocket.OPEN; + this.onopen?.(); + } + receiveText(text: string): void { + this.onmessage?.({ data: text }); + } + receiveFrame(ch: number, frame: Uint8Array): void { + const tagged = tagGatewayFrame(ch, frame); + const buf = new ArrayBuffer(tagged.length); + new Uint8Array(buf).set(tagged); + this.onmessage?.({ data: buf }); + } + drop(): void { + this.readyState = FakeWebSocket.CLOSED; + this.onclose?.(); + } + controls(): GatewayClientMsg[] { + return this.sent + .filter((d): d is string => typeof d === "string") + .map((d) => parseGatewayClientMsg(d)!) + .filter(Boolean); + } + frames(): Array<{ ch: number; type: number; frame: Uint8Array }> { + return this.sent + .filter((d): d is ArrayBuffer | Uint8Array => typeof d !== "string") + .map((d) => { + const bytes = d instanceof ArrayBuffer ? new Uint8Array(d) : d; + const tagged = untagGatewayFrame(bytes)!; + const dec = decoding.createDecoder(tagged.frame.slice()); + return { + ch: tagged.ch, + type: decoding.readVarUint(dec), + frame: tagged.frame.slice(), + }; + }); + } +} + +vi.stubGlobal("WebSocket", FakeWebSocket); + +let projSeq = 0; +function facadeOpts(docPath: string, passive = false) { + return { + endpoint: "http://localhost:3055", + scopeId: "scope-x", + // Fresh project per call site keeps the module-global connection registry + // from leaking one test's socket into the next. + projectId: `proj-${projSeq}`, + docPath, + passive, + }; +} + +function newProject(): void { + projSeq++; +} + +const step2From = (serverDoc: Y.Doc, clientStep1?: Uint8Array): Uint8Array => { + // Answer a client Step1 the way a BoardRoom would. + let sv: Uint8Array | undefined; + if (clientStep1) { + const dec = decoding.createDecoder(clientStep1.slice()); + decoding.readVarUint(dec); // MESSAGE_SYNC + decoding.readVarUint(dec); // messageYjsSyncStep1 + sv = decoding.readVarUint8Array(dec); + } + const enc = encoding.createEncoder(); + encoding.writeVarUint(enc, 0); + syncProtocol.writeSyncStep2(enc, serverDoc, sv); + return encoding.toUint8Array(enc); +}; + +const cleanups: Array<() => void> = []; +afterEach(() => { + for (const c of cleanups.splice(0)) c(); + FakeWebSocket.instances = []; +}); + +function track(facade: GatewayDocFacade, ...docs: Y.Doc[]): void { + cleanups.push(() => { + try { + facade.destroy(); + } catch { + /* already destroyed */ + } + for (const d of docs) d.destroy(); + }); +} + +describe("gateway facade — active documents", () => { + it("subscribes, syncs via Step1/Step2, and pushes local updates", async () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("a.kicad_sch")); + track(facade, doc); + const ws = FakeWebSocket.instances.at(-1)!; + expect(ws.url).toContain("/parties/project-room/project%3Ascope-x%3A"); + + ws.open(); + const [sub] = ws.controls(); + expect(sub).toEqual({ + t: "sub", + ch: expect.any(Number) as number, + doc: "a.kicad_sch", + mode: "active", + }); + const step1 = ws.frames().find((f) => f.type === 0); + expect(step1).toBeTruthy(); + + // Server state lands via Step2 → whenSynced resolves, doc holds it. + const serverDoc = new Y.Doc(); + serverDoc.getMap("m").set("k", "v"); + cleanups.push(() => serverDoc.destroy()); + ws.receiveFrame(sub!.ch, step2From(serverDoc, step1!.frame)); + await facade.whenSynced(); + expect(doc.getMap("m").get("k")).toBe("v"); + + // A local edit goes out as a sync update frame. + ws.sent = []; + doc.getMap("m").set("mine", 1); + expect(ws.frames().some((f) => f.type === 0)).toBe(true); + }); + + it("re-sends Step1 on a resync prompt", () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("a.kicad_sch")); + track(facade, doc); + const ws = FakeWebSocket.instances.at(-1)!; + ws.open(); + const ch = ws.controls()[0]!.ch; + ws.sent = []; + ws.receiveText(JSON.stringify({ t: "resync", ch })); + expect(ws.frames().filter((f) => f.type === 0).length).toBe(1); + }); +}); + +describe("gateway facade — passive warm pool (the laziness contract)", () => { + it("puts NO doc frame on the wire, and whenSynced resolves on subscribe", async () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("b.kicad_sch", true)); + track(facade, doc); + const ws = FakeWebSocket.instances.at(-1)!; + ws.open(); + expect(ws.controls()[0]).toMatchObject({ t: "sub", mode: "passive" }); + // Awareness/query frames are allowed (they're not demand); sync is not. + expect(ws.frames().filter((f) => f.type === 0)).toEqual([]); + await facade.whenSynced(); // resolves without any doc state + expect(doc.share.size).toBe(0); + }); + + it("touched marks dirty via callback; activate() upgrades and truly syncs", async () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("b.kicad_sch", true)); + track(facade, doc); + const ws = FakeWebSocket.instances.at(-1)!; + ws.open(); + const ch = ws.controls()[0]!.ch; + + let touched = 0; + facade.onTouched(() => touched++); + ws.receiveText(JSON.stringify({ t: "touched", ch })); + expect(touched).toBe(1); + + ws.sent = []; + const syncing = facade.activate(); + expect(ws.controls()).toEqual([{ t: "act", ch }]); + const step1 = ws.frames().find((f) => f.type === 0); + expect(step1).toBeTruthy(); + const serverDoc = new Y.Doc(); + serverDoc.getMap("m").set("from", "server"); + cleanups.push(() => serverDoc.destroy()); + ws.receiveFrame(ch, step2From(serverDoc, step1!.frame)); + await syncing; + expect(doc.getMap("m").get("from")).toBe("server"); + }); + + it("a local write into a passive doc auto-activates instead of silently desyncing", () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("b.kicad_sch", true)); + track(facade, doc); + const ws = FakeWebSocket.instances.at(-1)!; + ws.open(); + ws.sent = []; + doc.getMap("m").set("stray", true); + expect(ws.controls().some((m) => m.t === "act")).toBe(true); + expect(ws.frames().some((f) => f.type === 0)).toBe(true); + }); +}); + +describe("gateway facade — suberr is terminal", () => { + it("rejects pending and future syncs with CollabSubRejectedError", async () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("bad.kicad_sch")); + track(facade, doc); + const ws = FakeWebSocket.instances.at(-1)!; + ws.open(); + const ch = ws.controls()[0]!.ch; + const pending = facade.whenSynced(); + ws.receiveText( + JSON.stringify({ t: "suberr", ch, status: 409, message: "flagged" }), + ); + await expect(pending).rejects.toBeInstanceOf(CollabSubRejectedError); + await expect(facade.activate()).rejects.toMatchObject({ status: 409 }); + }); +}); + +describe("gateway facade — presence channel", () => { + it("never speaks sync; publishes and receives awareness", async () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("~presence")); + track(facade, doc); + const ws = FakeWebSocket.instances.at(-1)!; + ws.open(); + await facade.whenSynced(); + expect(ws.frames().filter((f) => f.type === 0)).toEqual([]); + // Query for peers went out on open. + expect(ws.frames().some((f) => f.type === 3)).toBe(true); + + ws.sent = []; + facade.awareness.setLocalState({ user: { id: "me" } }); + expect(ws.frames().some((f) => f.type === 1)).toBe(true); + + // An inbound query is answered with our own state. + const ch = 0; + ws.sent = []; + ws.receiveFrame(ch, new Uint8Array([3])); + expect(ws.frames().some((f) => f.type === 1)).toBe(true); + }); +}); + +describe("gateway connection — mux + reconnect", () => { + it("facades of one project share one socket with distinct channels", () => { + newProject(); + const docA = new Y.Doc(); + const docB = new Y.Doc(); + const a = new GatewayDocFacade(docA, facadeOpts("a.kicad_sch")); + const b = new GatewayDocFacade(docB, facadeOpts("b.kicad_sch", true)); + track(a, docA); + track(b, docB); + expect(FakeWebSocket.instances.length).toBe(1); + const ws = FakeWebSocket.instances[0]!; + ws.open(); + const subs = ws.controls().filter((m) => m.t === "sub"); + expect(subs.length).toBe(2); + expect(new Set(subs.map((m) => m.ch)).size).toBe(2); + }); + + it("reconnect re-subscribes with the CURRENT mode and re-syncs actives", async () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("a.kicad_sch", true)); + track(facade, doc); + const ws1 = FakeWebSocket.instances.at(-1)!; + ws1.open(); + const ch = ws1.controls()[0]!.ch; + // Activate (and settle the sync) on socket 1. + const syncing = facade.activate(); + const step1 = ws1.frames().find((f) => f.type === 0)!; + const serverDoc = new Y.Doc(); + cleanups.push(() => serverDoc.destroy()); + ws1.receiveFrame(ch, step2From(serverDoc, step1.frame)); + await syncing; + + vi.useFakeTimers(); + cleanups.push(() => vi.useRealTimers()); + ws1.drop(); + vi.advanceTimersByTime(1_000); // past the reconnect backoff + const ws2 = FakeWebSocket.instances.at(-1)!; + expect(ws2).not.toBe(ws1); + ws2.open(); + // The once-passive sub comes back as ACTIVE, followed by a fresh Step1. + expect(ws2.controls()[0]).toMatchObject({ t: "sub", mode: "active" }); + expect(ws2.frames().some((f) => f.type === 0)).toBe(true); + }); + + it("destroying the last facade closes the shared socket", () => { + newProject(); + const doc = new Y.Doc(); + const facade = new GatewayDocFacade(doc, facadeOpts("a.kicad_sch")); + const ws = FakeWebSocket.instances.at(-1)!; + ws.open(); + facade.destroy(); + doc.destroy(); + expect(ws.readyState).toBe(FakeWebSocket.CLOSED); + }); +}); diff --git a/web/standalone/src/wasm/collab/gateway.ts b/web/standalone/src/wasm/collab/gateway.ts new file mode 100644 index 0000000..2278ccf --- /dev/null +++ b/web/standalone/src/wasm/collab/gateway.ts @@ -0,0 +1,543 @@ +/** + * Client side of the ProjectRoom gateway (load-path-rework 0003): ONE + * websocket per (endpoint, project) carrying every per-doc collab room plus + * the `~presence` channel, replacing one socket per room. Facades implement + * the existing {@link YjsProvider} seam, so sheet-manager / sibling-restage / + * cross-app keep working unchanged; the sheet manager's warm pool additionally + * uses `passive` subscriptions (register interest, no doc sync) + `activate()` + * so parked sheets never wake their BoardRoom (the 0001 §5 lazy amendment). + * + * The per-facade y-protocol client is the ~150 lines y-partyserver's provider + * used to do for us (Step1/Step2/Update via y-protocols/sync, awareness via + * y-protocols/awareness), reading/writing frames tagged with the facade's + * channel id (@pcbjam/shared gateway-wire). + */ + +import * as decoding from "lib0/decoding"; +import * as encoding from "lib0/encoding"; +import { + applyAwarenessUpdate, + Awareness, + encodeAwarenessUpdate, + removeAwarenessStates, +} from "y-protocols/awareness"; +import * as syncProtocol from "y-protocols/sync"; +import type * as Y from "yjs"; +import { + type GatewayClientMsg, + type GatewayServerMsg, + type GatewaySubMode, + parseGatewayServerMsg, + PRESENCE_DOC_PATH, + projectRoomName, + tagGatewayFrame, + untagGatewayFrame, +} from "@pcbjam/shared"; +import { cwarn } from "./debug"; +import type { YjsProvider } from "./provider"; + +const MESSAGE_SYNC = 0; +const MESSAGE_AWARENESS = 1; +const MESSAGE_QUERY_AWARENESS = 3; + +const MAX_BACKOFF_MS = 30_000; +const ACTIVATE_TIMEOUT_MS = 30_000; + +/** + * The gateway refused a subscription (`suberr`): 409 invalid-file, 403 + * presence-as-readonly… Terminal for the channel — retrying cannot help until + * the underlying condition changes, so callers must NOT enter a retry ladder + * (sheet-manager treats it like SexprVersionError). + */ +export class CollabSubRejectedError extends Error { + constructor( + public readonly docPath: string, + public readonly status: number, + message: string, + ) { + super(`collab subscription to ${docPath} rejected (${status}): ${message}`); + this.name = "CollabSubRejectedError"; + } +} + +function gatewayWsUrl(endpoint: string, room: string, token?: string): string { + let base = endpoint.replace(/\/$/, ""); + if (!/^[a-z]+:\/\//i.test(base)) { + const local = /^(localhost|127\.0\.0\.1|\[::1\])(:|$)/.test(base); + base = `${local ? "ws" : "wss"}://${base}`; + } + base = base.replace(/^http/, "ws"); + const url = new URL( + `${base}/parties/project-room/${encodeURIComponent(room)}`, + ); + url.searchParams.set("_pk", Math.random().toString(36).slice(2, 12)); + if (token) url.searchParams.set("token", token); + return url.toString(); +} + +// --- the shared per-project socket ------------------------------------------ + +class GatewayConnection { + private ws: WebSocket | null = null; + private open = false; + private closed = false; + private attempts = 0; + private reconnectTimer: ReturnType | undefined; + private chSeq = 0; + private readonly facades = new Map(); + refs = 0; + + constructor( + private readonly endpoint: string, + private readonly room: string, + private readonly token: string | undefined, + private readonly onGone: () => void, + ) { + this.dial(); + } + + private dial(): void { + if (this.closed) return; + let ws: WebSocket; + try { + ws = new WebSocket(gatewayWsUrl(this.endpoint, this.room, this.token)); + } catch (err) { + cwarn(`[gateway] dial failed for ${this.room}`, err); + this.scheduleReconnect(); + return; + } + ws.binaryType = "arraybuffer"; + this.ws = ws; + ws.onopen = () => { + if (this.closed || this.ws !== ws) return; + this.open = true; + this.attempts = 0; + // Re-establish every live subscription, then let each facade run its + // (re)open protocol — sub before Step1, ordered on one socket. + for (const facade of this.facades.values()) { + this.sendControl(facade.subMsg()); + facade.handleSocketOpen(); + } + }; + ws.onmessage = (e) => { + if (this.ws !== ws) return; + const data: unknown = e.data; + if (typeof data === "string") { + const msg = parseGatewayServerMsg(data); + if (msg) this.facades.get(msg.ch)?.handleControl(msg); + return; + } + if (data instanceof ArrayBuffer) { + const tagged = untagGatewayFrame(new Uint8Array(data)); + if (tagged) this.facades.get(tagged.ch)?.handleFrame(tagged.frame); + } + }; + const down = () => { + if (this.ws !== ws) return; + const wasOpen = this.open; + this.open = false; + this.ws = null; + if (wasOpen) { + for (const facade of this.facades.values()) facade.handleSocketDown(); + } + this.scheduleReconnect(); + }; + ws.onclose = down; + ws.onerror = down; + } + + private scheduleReconnect(): void { + if (this.closed || this.reconnectTimer !== undefined) return; + // Same ladder the per-room provider used (100ms · 2^n, capped) — but ONE + // ladder for the whole project instead of one per room. + const delay = Math.min(100 * 2 ** this.attempts, MAX_BACKOFF_MS); + this.attempts++; + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = undefined; + this.dial(); + }, delay); + } + + isOpen(): boolean { + return this.open; + } + + register(facade: GatewayDocFacade): number { + const ch = this.chSeq++; + this.facades.set(ch, facade); + return ch; + } + + /** Called by a registered facade once it knows its ch (post-construction). */ + announce(facade: GatewayDocFacade): void { + if (this.open) { + this.sendControl(facade.subMsg()); + facade.handleSocketOpen(); + } + } + + unregister(ch: number): void { + if (!this.facades.delete(ch)) return; + this.sendControl({ t: "unsub", ch }); + this.refs--; + if (this.refs <= 0) this.shutdown(); + } + + sendControl(msg: GatewayClientMsg): void { + if (this.open && this.ws?.readyState === WebSocket.OPEN) { + this.ws.send(JSON.stringify(msg)); + } + } + + sendFrame(ch: number, frame: Uint8Array): void { + if (this.open && this.ws?.readyState === WebSocket.OPEN) { + this.ws.send(tagGatewayFrame(ch, frame)); + } + } + + private shutdown(): void { + this.closed = true; + if (this.reconnectTimer !== undefined) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = undefined; + } + try { + this.ws?.close(); + } catch { + /* already dead */ + } + this.ws = null; + this.open = false; + this.onGone(); + } +} + +const connections = new Map(); + +function acquireConnection( + endpoint: string, + scopeId: string, + projectId: string, + token?: string, +): GatewayConnection { + const room = projectRoomName(scopeId, projectId); + const key = `${endpoint}|${token ?? ""}|${room}`; + let conn = connections.get(key); + if (!conn) { + const created: GatewayConnection = new GatewayConnection( + endpoint, + room, + token, + () => { + if (connections.get(key) === created) connections.delete(key); + }, + ); + conn = created; + connections.set(key, conn); + } + conn.refs++; + return conn; +} + +// --- per-doc facade ---------------------------------------------------------- + +export interface GatewayFacadeOpts { + endpoint: string; + scopeId: string; + projectId: string; + docPath: string; + token?: string; + /** Passive = register interest only (parked warm-pool sheet): no SyncStep1, + * no BoardRoom wake; `touched` hints + awareness still flow. */ + passive?: boolean; +} + +/** + * A YjsProvider backed by one gateway sub-channel. Owns its own Awareness + * (skeleton presence publishes a different state per parked room — 0003 §6). + */ +export class GatewayDocFacade implements YjsProvider { + readonly awareness: Awareness; + private readonly conn: GatewayConnection; + private readonly ch: number; + private readonly isPresence: boolean; + private mode: GatewaySubMode; + private dead: CollabSubRejectedError | null = null; + private destroyed = false; + private synced = false; + private subEverSent = false; + private readonly touchedCbs: Array<() => void> = []; + private readonly syncWaiters: Array<{ + resolve: () => void; + reject: (e: unknown) => void; + }> = []; + private readonly subWaiters: Array<{ + resolve: () => void; + reject: (e: unknown) => void; + }> = []; + + constructor( + private readonly doc: Y.Doc, + opts: GatewayFacadeOpts, + ) { + this.isPresence = opts.docPath === PRESENCE_DOC_PATH; + this.docPath = opts.docPath; + this.mode = opts.passive && !this.isPresence ? "passive" : "active"; + this.awareness = new Awareness(doc); + this.conn = acquireConnection( + opts.endpoint, + opts.scopeId, + opts.projectId, + opts.token, + ); + this.ch = this.conn.register(this); + + this.awareness.on("update", this.onAwarenessUpdate); + if (!this.isPresence) this.doc.on("update", this.onDocUpdate); + + this.conn.announce(this); + } + + readonly docPath: string; + + // --- YjsProvider ---------------------------------------------------------- + + whenSynced(): Promise { + if (this.dead) return Promise.reject(this.dead); + if (this.mode === "active" && !this.isPresence) return this.activate(); + // Passive/presence: "synced" = the subscription reached an open socket. + if (this.subEverSent) return Promise.resolve(); + return new Promise((resolve, reject) => { + this.subWaiters.push({ resolve, reject }); + }); + } + + /** + * Upgrade to a real y-protocol participant and resolve once the doc holds + * the server state (first SyncStep2). Idempotent; rejects with + * {@link CollabSubRejectedError} if the gateway killed the channel. + */ + activate(): Promise { + if (this.dead) return Promise.reject(this.dead); + if (this.isPresence) return this.whenSynced(); + if (this.synced) return Promise.resolve(); + const wasPassive = this.mode === "passive"; + this.mode = "active"; + if (this.conn.isOpen()) { + if (wasPassive) this.conn.sendControl({ t: "act", ch: this.ch }); + this.beginSync(); + } + return new Promise((resolve, reject) => { + const waiter = { + resolve: () => { + clearTimeout(timer); + resolve(); + }, + reject: (e: unknown) => { + clearTimeout(timer); + reject(e); + }, + }; + // Bounded: doSwitch serializes on this — a dead socket must surface as + // a retryable failure, not a hung queue. + const timer = setTimeout(() => { + const i = this.syncWaiters.indexOf(waiter); + if (i >= 0) this.syncWaiters.splice(i, 1); + reject( + new Error( + `gateway activate for ${this.docPath} exceeded ${ACTIVATE_TIMEOUT_MS}ms`, + ), + ); + }, ACTIVATE_TIMEOUT_MS); + this.syncWaiters.push(waiter); + }); + } + + /** `touched` hints (doc changed while passive) — sheet dirty flag. */ + onTouched(cb: () => void): void { + this.touchedCbs.push(cb); + } + + destroy(): void { + if (this.destroyed) return; + this.destroyed = true; + // Clean departure: the removal frame reaches peers while the socket is + // still up (the gateway's synthesized tombstone is only the crash path). + try { + removeAwarenessStates(this.awareness, [this.doc.clientID], "destroy"); + } catch { + /* awareness already torn down */ + } + this.awareness.off("update", this.onAwarenessUpdate); + if (!this.isPresence) this.doc.off("update", this.onDocUpdate); + this.awareness.destroy(); + this.conn.unregister(this.ch); + const err = this.dead ?? new Error("gateway facade destroyed"); + for (const w of this.syncWaiters.splice(0)) w.reject(err); + for (const w of this.subWaiters.splice(0)) w.reject(err); + } + + // --- wire (called by GatewayConnection) ----------------------------------- + + subMsg(): GatewayClientMsg { + return { t: "sub", ch: this.ch, doc: this.docPath, mode: this.mode }; + } + + handleSocketOpen(): void { + if (this.destroyed || this.dead) return; + this.subEverSent = true; + for (const w of this.subWaiters.splice(0)) w.resolve(); + // (Re)announce presence: a query for peers' states + our own, if any. + this.sendQueryAwareness(); + if (this.awareness.getLocalState() !== null) this.publishLocalAwareness(); + if (this.mode === "active" && !this.isPresence) this.beginSync(); + } + + handleSocketDown(): void { + // Same as y-websocket: peers' awareness is stale the moment the socket + // drops; local state survives and re-publishes on reconnect. + const remote = [...this.awareness.getStates().keys()].filter( + (id) => id !== this.doc.clientID, + ); + if (remote.length > 0) { + removeAwarenessStates(this.awareness, remote, "connection closed"); + } + this.synced = false; + } + + handleControl(msg: GatewayServerMsg): void { + if (this.destroyed) return; + if (msg.t === "suberr") { + this.dead = new CollabSubRejectedError( + this.docPath, + msg.status, + msg.message, + ); + cwarn(`[gateway] ${this.dead.message}`); + for (const w of this.syncWaiters.splice(0)) w.reject(this.dead); + for (const w of this.subWaiters.splice(0)) w.reject(this.dead); + return; + } + if (msg.t === "resync") { + // The doc's relay (re)connected server-side — our Step1 pulls the news. + if (this.mode === "active" && !this.isPresence) this.sendSyncStep1(); + return; + } + // touched + for (const cb of this.touchedCbs) cb(); + } + + handleFrame(frame: Uint8Array): void { + if (this.destroyed || frame.length === 0) return; + try { + this.handleFrameInner(frame); + } catch (err) { + // A malformed frame must never take down the shared socket's handler. + cwarn(`[gateway] dropped undecodable frame for ${this.docPath}`, err); + } + } + + private handleFrameInner(frame: Uint8Array): void { + const decoder = decoding.createDecoder(frame.slice()); + const type = decoding.readVarUint(decoder); + if (type === MESSAGE_SYNC) { + if (this.isPresence) return; + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_SYNC); + const messageType = syncProtocol.readSyncMessage( + decoder, + encoder, + this.doc, + this, + ); + if (encoding.length(encoder) > 1) { + this.send(encoding.toUint8Array(encoder)); + } + if (messageType === syncProtocol.messageYjsSyncStep2 && !this.synced) { + this.synced = true; + for (const w of this.syncWaiters.splice(0)) w.resolve(); + } + return; + } + if (type === MESSAGE_AWARENESS) { + applyAwarenessUpdate( + this.awareness, + decoding.readVarUint8Array(decoder), + "gateway-remote", + ); + return; + } + if (type === MESSAGE_QUERY_AWARENESS) { + // A peer asks who is here — answer with our own state only. + if (this.awareness.getLocalState() !== null) this.publishLocalAwareness(); + } + } + + // --- internals ------------------------------------------------------------ + + private send(frame: Uint8Array): void { + this.conn.sendFrame(this.ch, frame); + } + + private beginSync(): void { + this.sendSyncStep1(); + } + + private sendSyncStep1(): void { + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_SYNC); + syncProtocol.writeSyncStep1(encoder, this.doc); + this.send(encoding.toUint8Array(encoder)); + } + + private sendQueryAwareness(): void { + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_QUERY_AWARENESS); + this.send(encoding.toUint8Array(encoder)); + } + + private publishLocalAwareness(): void { + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_AWARENESS); + encoding.writeVarUint8Array( + encoder, + encodeAwarenessUpdate(this.awareness, [this.doc.clientID]), + ); + this.send(encoding.toUint8Array(encoder)); + } + + private readonly onDocUpdate = (update: Uint8Array, origin: unknown): void => { + if (origin === this) return; // our own readSyncMessage apply + if (this.mode === "passive") { + // A local write into a passive doc (activate-before-write slipped) — + // auto-activate so the edit is neither dropped nor pushed unsynced. + void this.activate().catch(() => { + /* surfaced via the activate caller/suberr path */ + }); + } + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_SYNC); + syncProtocol.writeUpdate(encoder, update); + this.send(encoding.toUint8Array(encoder)); + }; + + private readonly onAwarenessUpdate = ( + { + added, + updated, + removed, + }: { added: number[]; updated: number[]; removed: number[] }, + origin: unknown, + ): void => { + if (origin === "gateway-remote") return; // no echo loops + const changed = added.concat(updated).concat(removed); + if (changed.length === 0) return; + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_AWARENESS); + encoding.writeVarUint8Array( + encoder, + encodeAwarenessUpdate(this.awareness, changed), + ); + this.send(encoding.toUint8Array(encoder)); + }; +} diff --git a/web/standalone/src/wasm/collab/index.ts b/web/standalone/src/wasm/collab/index.ts index 815e534..847aba6 100644 --- a/web/standalone/src/wasm/collab/index.ts +++ b/web/standalone/src/wasm/collab/index.ts @@ -155,6 +155,11 @@ export async function connectKicadDoc(opts: { signal?: AbortSignal; /** Whole-path deadline; defaults to 30s. */ timeoutMs?: number; + /** Gateway transport (load-path-rework 0003): register interest without + * demanding doc state — the warm pool's parked sheets. `whenSynced` then + * resolves on subscription, and `provider.activate?.()` is the real sync + * barrier before any bind/read/write of the doc. */ + passive?: boolean; }): Promise { const timeoutMs = opts.timeoutMs ?? CONNECT_TIMEOUT_MS; // An already-aborted owner never gets a session — even one that could @@ -186,6 +191,7 @@ export async function connectKicadDoc(opts: { const providerPromise = connectProvider(doc, opts.provider, { room: opts.room, + passive: opts.passive, }); providerPromise.catch(() => {}); // may lose the race and reject later diff --git a/web/standalone/src/wasm/collab/provider.ts b/web/standalone/src/wasm/collab/provider.ts index 364b4f3..9b05a89 100644 --- a/web/standalone/src/wasm/collab/provider.ts +++ b/web/standalone/src/wasm/collab/provider.ts @@ -1,5 +1,6 @@ import type * as Y from "yjs"; import { Awareness } from "y-protocols/awareness"; +import { parseCollabRoomId } from "@pcbjam/shared"; import { connectBroadcastChannel } from "./broadcast-transport"; import { connectAwarenessBroadcast } from "./awareness-bc"; @@ -27,6 +28,18 @@ export interface YjsProvider { * sibling channel. Absent for `none` — presence UI then renders nothing. */ awareness?: Awareness; + /** + * Gateway transport only (load-path-rework 0003): upgrade a `passive` + * subscription to a real y-protocol participant and resolve once the doc + * holds the server state. Providers without it are always fully synced + * after `whenSynced` — callers treat absence as an already-synced no-op. + */ + activate?(): Promise; + /** + * Gateway transport only: the doc changed server-side while this + * subscription was passive (the sheet manager's parked-dirty flag). + */ + onTouched?(cb: () => void): void; } export type ProviderKind = @@ -154,17 +167,38 @@ async function hocuspocusProvider( /** * Connect a Y.Doc to the env-selected provider. Network libraries are imported * lazily so a deployment only ships the one it uses. + * + * `passive` (gateway transport only): register interest without demanding doc + * state — the sheet manager's warm pool. Other providers ignore it and stay + * fully synced, a harmless superset. */ export async function connectProvider( doc: Y.Doc, config: ProviderConfig, - opts: { room: string }, + opts: { room: string; passive?: boolean }, ): Promise { switch (config.kind) { case "broadcastchannel": return broadcastChannelProvider(doc, opts.room, config.settleMs ?? 300); - case "partykit": + case "partykit": { + // Load-path-rework 0003: board/presence rooms ride the per-project + // gateway socket (one ws per project instead of one per room). The + // legacy per-room dial stays only for room ids that don't parse — an + // operator/debug shape the gateway can't address. + const parsed = parseCollabRoomId(opts.room); + if (parsed) { + const { GatewayDocFacade } = await import("./gateway"); + return new GatewayDocFacade(doc, { + endpoint: requireEndpoint(config), + scopeId: parsed.scopeId, + projectId: parsed.projectId, + docPath: parsed.docPath, + token: config.params?.token, + passive: opts.passive, + }); + } return partyKitProvider(doc, requireEndpoint(config), opts.room, config.params); + } case "hocuspocus": return hocuspocusProvider(doc, requireEndpoint(config), opts.room, config.params); case "none": diff --git a/web/standalone/src/wasm/collab/sheet-manager.ts b/web/standalone/src/wasm/collab/sheet-manager.ts index 8991c31..27d6581 100644 --- a/web/standalone/src/wasm/collab/sheet-manager.ts +++ b/web/standalone/src/wasm/collab/sheet-manager.ts @@ -175,7 +175,10 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab } } - async function ensureRoom(sheetPath: string): Promise { + async function ensureRoom( + sheetPath: string, + connectOpts?: { passive?: boolean }, + ): Promise { const existing = rooms.get(sheetPath); if (existing) return existing; const inflight = connecting.get(sheetPath); @@ -185,6 +188,7 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab const session = await connectKicadDoc({ provider, room: collabRoomId(scopeId, projectId, sheetPath), + passive: connectOpts?.passive, }); if (destroyed) { // The manager died while this connect was in flight (unmount during @@ -203,6 +207,12 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab editorMatchesDoc: false, dirty: false, }; + // Gateway transport: a `touched` hint means the doc changed server-side + // while this subscription was passive — same catch-up contract as the + // parked-update watch. + session.provider.onTouched?.(() => { + room.dirty = true; + }); rooms.set(sheetPath, room); log(`[sheet] warm room connected: ${sheetPath}`); // A room warmed after the first bind starts parked — give it a skeleton @@ -261,6 +271,12 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab const room = await ensureRoom(sheetPath); if (destroyed) return; + // Gateway transport: a passively-warmed sheet holds no doc state yet — + // activate() is the real sync barrier (no-op for other providers and on + // revisits). Runs BEFORE the watch detaches so catch-up updates still + // mark `dirty` for the adopt decision below. + await room.session.provider.activate?.(); + if (destroyed) return; // Activating: stop tracking parked updates and bind the (warm) doc to the editor. room.detachWatch?.(); @@ -316,8 +332,11 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab // Terminal version skew (findings C-5): retrying cannot make an // incompatible doc encoding bindable. No retry timer — rethrow so the // caller surfaces it (boot converts it into the user-visible version - // error instead of reporting "connected" with no binding). - if ((err as { name?: string } | null)?.name === "SexprVersionError") { + // error instead of reporting "connected" with no binding). A gateway + // subscription rejection (invalid-file 409 — load-path-rework 0003) + // is terminal the same way: the flag only clears on a new upload. + const name = (err as { name?: string } | null)?.name; + if (name === "SexprVersionError" || name === "CollabSubRejectedError") { throw err; } // Still the sheet the editor shows and not yet bound → retry with backoff, @@ -352,21 +371,36 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab function syncLayoutFromSave(sheetPath: string, fileText: string): void { const room = rooms.get(sheetPath); if (!room) return; // not a collab sheet (or still onboarding) — nothing to sync - try { - // Writing to a PARKED room's doc marks it dirty via startWatch — fine: - // the diff-on-rebind adopt makes the catch-up cost the real delta only. - if (syncLayoutToY(fileToDoc(fileText), room.doc, "layout-save")) { - clog(`[sheet] layout save-sync: ${sheetPath} updated`); + const write = (): void => { + try { + // Writing to a PARKED room's doc marks it dirty via startWatch — fine: + // the diff-on-rebind adopt makes the catch-up cost the real delta only. + if (syncLayoutToY(fileToDoc(fileText), room.doc, "layout-save")) { + clog(`[sheet] layout save-sync: ${sheetPath} updated`); + } + } catch (err) { + cwarn(`[sheet] layout save-sync failed for ${sheetPath}`, err); } - } catch (err) { - cwarn(`[sheet] layout save-sync failed for ${sheetPath}`, err); + }; + // Gateway transport: a passively-warmed doc is empty — writing into it + // would push partial state. A save IS write demand, so sync first. + const activate = room.session.provider.activate?.(); + if (activate) { + void activate.then(write).catch((err) => { + cwarn(`[sheet] layout save-sync activate failed for ${sheetPath}`, err); + }); + } else { + write(); } } async function connectAll(sheetPaths: string[]): Promise { await Promise.all( sheetPaths.map((p) => - ensureRoom(p).catch((err) => { + // Passive on the gateway (load-path-rework 0003): warm-all becomes + // pure registration — awareness + touched hints flow, but no + // BoardRoom wakes for a sheet nobody opens or edits. + ensureRoom(p, { passive: true }).catch((err) => { cwarn(`[sheet] failed to warm ${p}`, err); return null; }),