From feb40f2091c28afb0097366d06498a36c6d36a43 Mon Sep 17 00:00:00 2001 From: kelmith Date: Tue, 14 Jul 2026 12:29:05 +0530 Subject: [PATCH 1/3] feat: websocket migration of canvas --- package-lock.json | 24 ++ package.json | 1 + scripts/seed-local-swarm.ts | 112 +++++++ ...rgCanvasBackground.multiSelectKey.test.tsx | 6 +- .../[githubLogin]/canvas/relay-token/route.ts | 95 ++++++ .../connections/OrgCanvasBackground.tsx | 4 +- src/hooks/useCanvasCollaboration.ts | 2 +- src/hooks/useCanvasCollaborationViaRelay.ts | 285 ++++++++++++++++++ 8 files changed, 522 insertions(+), 7 deletions(-) create mode 100644 scripts/seed-local-swarm.ts create mode 100644 src/app/api/orgs/[githubLogin]/canvas/relay-token/route.ts create mode 100644 src/hooks/useCanvasCollaborationViaRelay.ts diff --git a/package-lock.json b/package-lock.json index b60834ada4..37b09f3c1f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -198,6 +198,30 @@ "vite": "^7.3.1" } }, + "../system-canvas/packages/collab": { + "name": "system-canvas-collab", + "version": "0.2.57", + "extraneous": true, + "license": "MIT", + "devDependencies": { + "@types/react": "^18.0.0 || ^19.0.0", + "react": "^19.0.0", + "system-canvas": "^0.2.57", + "typescript": "^5.4.0", + "vitest": "^4.1.7", + "yjs": "^13.6.0" + }, + "peerDependencies": { + "react": "^18.0.0 || ^19.0.0", + "system-canvas": "^0.2.57", + "yjs": "^13.6.0" + }, + "peerDependenciesMeta": { + "react": { + "optional": true + } + } + }, "node_modules/@adobe/css-tools": { "version": "4.4.4", "resolved": "https://registry.npmjs.org/@adobe/css-tools/-/css-tools-4.4.4.tgz", diff --git a/package.json b/package.json index a3ec51d02b..b163abc8c5 100644 --- a/package.json +++ b/package.json @@ -30,6 +30,7 @@ "validate:e2e-migration": "tsx scripts/validate-e2e-migration.ts", "rotate-keys": "tsx scripts/rotate-encryption-key.ts", "seed:db": "tsx scripts/helpers/seed-database.ts", + "seed:local-swarm": "dotenv -e .env.local -- tsx scripts/seed-local-swarm.ts", "seed:auto-seed": "tsx scripts/seed-from-github-account.ts", "seed:prompts": "tsx scripts/seed-stakwork-prompts.ts", "seed:action-nodes": "tsx scripts/helpers/seed-graph-action-nodes.ts", diff --git a/scripts/seed-local-swarm.ts b/scripts/seed-local-swarm.ts new file mode 100644 index 0000000000..72763fa947 --- /dev/null +++ b/scripts/seed-local-swarm.ts @@ -0,0 +1,112 @@ +/** + * Seed a local mock user + org + workspace + swarm wired for hive-relay. + * + * Mock login already creates the whole chain (via ensureMockWorkspaceForUser) + * but leaves the swarm WITHOUT a `swarmApiKey` and the org WITHOUT a + * `defaultWorkspaceId`. This script fills both so the relay-token flow works: + * - sets a known `swarmApiKey` (encrypted) + a localhost `swarmUrl`, and + * - makes the workspace the org's default (the Org Canvas resolves its + * swarm via org -> defaultWorkspace -> swarm). + * + * Run: + * npx dotenv -e .env.local -- npx tsx scripts/seed-local-swarm.ts + * Optional overrides: + * SEED_USERNAME=dev SEED_SWARM_API_KEY= + * + * Then start hive-relay with the printed SWARM_API_KEY. Idempotent. + */ +// Relative imports (matching the other scripts) so the editor resolves them; +// the `@/` aliases inside these modules still resolve at runtime under tsx. +import { WorkspaceRole } from "@prisma/client"; +import { db } from "../src/lib/db"; +import { ensureMockWorkspaceForUser } from "../src/utils/mockSetup"; +import { saveOrUpdateSwarm } from "../src/services/swarm/db"; + +const USERNAME = process.env.SEED_USERNAME ?? "dev"; +// A second member of the SAME org, so you can log in as two different users +// (in two browsers) and see each other's cursors on /org/. +const SECOND_USERNAME = process.env.SEED_SECOND_USERNAME ?? "dev2"; +// Stable default so re-running the seed doesn't change the key (you set the +// relay's SWARM_API_KEY once). Override with SEED_SWARM_API_KEY for a custom one. +const SWARM_API_KEY = process.env.SEED_SWARM_API_KEY ?? "local-dev-swarm-key"; + +async function main() { + const userId = `mock-${USERNAME}`; + + // 1. The mock user (mock login creates this on sign-in; we do it up-front + // so the seed is runnable before any login). + await db.user.upsert({ + where: { id: userId }, + update: {}, + create: { id: userId, name: USERNAME, email: `${USERNAME}@mock.dev` }, + }); + + // 2. Org + workspace + repository + swarm (the real mock setup). Idempotent: + // returns early if the user already has a workspace. + const slug = await ensureMockWorkspaceForUser(userId); + const workspace = await db.workspace.findUnique({ + where: { slug }, + select: { id: true, sourceControlOrgId: true }, + }); + if (!workspace) throw new Error(`workspace not found for slug ${slug}`); + + // 3. Set a known swarmApiKey (encrypted by saveOrUpdateSwarm) + a localhost + // swarmUrl so getRelayUrl() -> http://localhost:3333. + await saveOrUpdateSwarm({ + workspaceId: workspace.id, + swarmApiKey: SWARM_API_KEY, + swarmUrl: "http://localhost/api", + }); + + // 4. Make this workspace the org's default so the Org Canvas can resolve the + // swarm (org -> defaultWorkspace -> swarm). + let orgLogin = "(no org linked)"; + if (workspace.sourceControlOrgId) { + const org = await db.sourceControlOrg.update({ + where: { id: workspace.sourceControlOrgId }, + data: { defaultWorkspaceId: workspace.id }, + select: { githubLogin: true }, + }); + orgLogin = org.githubLogin; + } + + // 5. A second user, added as a member of the same workspace → both belong to + // the org, so 2-user presence works on /org/. + const secondId = `mock-${SECOND_USERNAME}`; + await db.user.upsert({ + where: { id: secondId }, + update: {}, + create: { + id: secondId, + name: SECOND_USERNAME, + email: `${SECOND_USERNAME}@mock.dev`, + }, + }); + await db.workspaceMember.upsert({ + where: { workspaceId_userId: { workspaceId: workspace.id, userId: secondId } }, + update: { role: WorkspaceRole.DEVELOPER, leftAt: null }, + create: { + workspaceId: workspace.id, + userId: secondId, + role: WorkspaceRole.DEVELOPER, + }, + }); + + console.log("\n✅ Local swarm ready for hive-relay\n"); + console.log(" mock logins :", USERNAME, "(owner) &", SECOND_USERNAME, "(member)"); + console.log(" workspace slug :", slug); + console.log(" org (githubLogin) :", orgLogin, " -> /org/" + orgLogin); + console.log(" SWARM_API_KEY :", SWARM_API_KEY); + console.log("\n Start hive-relay with:"); + console.log( + ` SWARM_API_KEY=${SWARM_API_KEY} CORS_ORIGIN=http://localhost:3000 PORT=3333 npm run dev`, + ); + console.log(""); +} + +main() + .then(() => process.exit(0)) + .catch((e) => { + console.error(e); + process.exit(1); + }); diff --git a/src/__tests__/unit/canvas/OrgCanvasBackground.multiSelectKey.test.tsx b/src/__tests__/unit/canvas/OrgCanvasBackground.multiSelectKey.test.tsx index 8c877263dd..479aacf238 100644 --- a/src/__tests__/unit/canvas/OrgCanvasBackground.multiSelectKey.test.tsx +++ b/src/__tests__/unit/canvas/OrgCanvasBackground.multiSelectKey.test.tsx @@ -33,11 +33,9 @@ vi.mock("@/hooks/useWorkspace", () => ({ }), })); -vi.mock("@/hooks/useCanvasCollaboration", () => ({ - useCanvasCollaboration: () => ({ +vi.mock("@/hooks/useCanvasCollaborationViaRelay", () => ({ + useCanvasCollaborationViaRelay: () => ({ collaborators: [], - broadcastCursor: vi.fn(), - broadcastSelection: vi.fn(), }), })); diff --git a/src/app/api/orgs/[githubLogin]/canvas/relay-token/route.ts b/src/app/api/orgs/[githubLogin]/canvas/relay-token/route.ts new file mode 100644 index 0000000000..641890324a --- /dev/null +++ b/src/app/api/orgs/[githubLogin]/canvas/relay-token/route.ts @@ -0,0 +1,95 @@ +import { NextRequest, NextResponse } from "next/server"; +import { db } from "@/lib/db"; +import { validateUserBelongsToOrg } from "@/services/workspace"; +import { getSwarmAccessByWorkspaceId } from "@/lib/helpers/swarm-access"; +import { getMiddlewareContext, requireAuth } from "@/lib/middleware/utils"; +import { signRelayToken } from "@/lib/relay-token"; +import { getRelayUrl } from "@/lib/utils/swarm"; + +const TOKEN_TTL_SECONDS = 300; + +/** + * GET /api/orgs/[githubLogin]/canvas/relay-token + * + * Issues a short-lived JWT the browser passes in the socket.io handshake + * to the per-swarm relay. Capability-scoped to exactly this org's canvas + * via the `resource` claim (`canvas:`), so the relay only + * lets the socket join that one room. Signed HS256 with the org's home + * swarm's `swarmApiKey` (org -> defaultWorkspace -> swarm). + * + * Returns { token, url, expiresInSeconds }. + */ +export async function GET( + request: NextRequest, + { params }: { params: Promise<{ githubLogin: string }> }, +) { + try { + const context = getMiddlewareContext(request); + const userOrResponse = requireAuth(context); + if (userOrResponse instanceof NextResponse) return userOrResponse; + + const { githubLogin } = await params; + + const isMember = await validateUserBelongsToOrg(githubLogin, userOrResponse.id); + if (!isMember) { + return NextResponse.json( + { error: "Organization not found" }, + { status: 404 }, + ); + } + + const org = await db.sourceControlOrg.findUnique({ + where: { githubLogin }, + select: { defaultWorkspaceId: true }, + }); + if (!org?.defaultWorkspaceId) { + return NextResponse.json( + { error: "Relay unavailable", reason: "NO_DEFAULT_WORKSPACE" }, + { status: 503 }, + ); + } + + const swarmAccess = await getSwarmAccessByWorkspaceId(org.defaultWorkspaceId); + if (!swarmAccess.success) { + return NextResponse.json( + { error: "Relay unavailable", reason: swarmAccess.error.type }, + { status: 503 }, + ); + } + if (!swarmAccess.data.swarmApiKey) { + return NextResponse.json( + { error: "Relay unavailable", reason: "SWARM_API_KEY_MISSING" }, + { status: 503 }, + ); + } + if (!swarmAccess.data.swarmName) { + return NextResponse.json( + { error: "Relay unavailable", reason: "SWARM_NAME_MISSING" }, + { status: 503 }, + ); + } + + const token = signRelayToken( + { + userId: userOrResponse.id, + name: userOrResponse.name, + image: null, + resource: `canvas:${githubLogin}`, + }, + swarmAccess.data.swarmApiKey, + TOKEN_TTL_SECONDS, + ); + + return NextResponse.json({ + token, + url: getRelayUrl(swarmAccess.data.swarmName, swarmAccess.data.swarmUrl), + expiresInSeconds: TOKEN_TTL_SECONDS, + }); + } catch (error) { + console.error("Error issuing canvas relay token:", error); + return NextResponse.json( + { error: "Failed to issue relay token" }, + { status: 500 }, + ); + } +} diff --git a/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx b/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx index 00211e7f99..b3f77cb7de 100644 --- a/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx +++ b/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx @@ -35,7 +35,7 @@ import { CreateServiceCanvasDialog } from "../_components/CreateServiceCanvasDia import { categoryAllowedOnScope, CATEGORY_REGISTRY } from "./canvas-categories"; import { useCanvasChatStore } from "../_state/canvasChatStore"; import { useSession } from "next-auth/react"; -import { useCanvasCollaboration } from "@/hooks/useCanvasCollaboration"; +import { useCanvasCollaborationViaRelay } from "@/hooks/useCanvasCollaborationViaRelay"; import { toast } from "sonner"; import { Spinner } from "@/components/ui/spinner"; import { Button } from "@/components/ui/button"; @@ -737,7 +737,7 @@ export function OrgCanvasBackground({ }); // Real-time canvas presence — cursors, selection halos, conflict flash - const { collaborators } = useCanvasCollaboration({ + const { collaborators } = useCanvasCollaborationViaRelay({ githubLogin, canvasRef: currentRef, userId: session?.user?.id ?? "", diff --git a/src/hooks/useCanvasCollaboration.ts b/src/hooks/useCanvasCollaboration.ts index 327ca4876a..76b12b5bc5 100644 --- a/src/hooks/useCanvasCollaboration.ts +++ b/src/hooks/useCanvasCollaboration.ts @@ -19,7 +19,7 @@ interface CollaboratorState extends Omit`); each cursor carries its + * `canvasRef` so peers only render cursors/selection for the sub-canvas + * they're currently viewing. + */ + +interface UseCanvasCollaborationResult { + collaborators: CanvasCollaboratorInfo[]; +} + +const CLIENT_TTL_MS = 60_000; +const PRUNE_INTERVAL_MS = 15_000; +const CURSOR_THROTTLE_MS = 50; + +function teamAvatarColor(userId: string): string { + const palette = [ + "#5EEAD4", "#38BDF8", "#A78BFA", "#FB923C", "#34D399", "#F472B6", + "#FACC15", "#60A5FA", "#F87171", "#C084FC", "#2DD4BF", "#4ADE80", + ]; + let hash = 0; + for (let i = 0; i < userId.length; i++) { + hash = (hash * 31 + userId.charCodeAt(i)) | 0; + } + return palette[Math.abs(hash) % palette.length]; +} + +/** One connected peer socket. Keyed by senderId (socket.id). */ +interface PeerState { + userId: string; + name: string; + color: string; + image: string | null; + cursor: { x: number; y: number } | null; + selectedNodeId: string | null; + /** Sub-canvas the cursor/selection is on; only rendered when it matches ours. */ + cursorRef: string; + lastSeenAt: number; +} + +interface RosterEntry { + odinguserId: string; + name: string; + image: string | null; + color: string; + joinedAt: number; + senderId: string; +} + +export function useCanvasCollaborationViaRelay({ + githubLogin, + canvasRef, + userId, + userName, + userImage, + getViewport, + getSvgElement, + containerRef, + selectedNodeId, + enabled = true, +}: UseCanvasCollaborationOptions): UseCanvasCollaborationResult { + const [peers, setPeers] = useState>(new Map()); + const socketRef = useRef(null); + + // Stable refs for use inside socket/pointer callbacks. + const userIdRef = useRef(userId); + userIdRef.current = userId; + const canvasRefRef = useRef(canvasRef); + canvasRefRef.current = canvasRef; + const selectedNodeIdRef = useRef(selectedNodeId); + selectedNodeIdRef.current = selectedNodeId; + const lastCursorRef = useRef<{ x: number; y: number } | null>(null); + + const upsertPeer = useCallback( + (senderId: string, patch: Partial) => { + setPeers((prev) => { + const next = new Map(prev); + const existing: PeerState = next.get(senderId) ?? { + userId: "", + name: "", + color: "#888", + image: null, + cursor: null, + selectedNodeId: null, + cursorRef: "", + lastSeenAt: Date.now(), + }; + next.set(senderId, { ...existing, ...patch, lastSeenAt: Date.now() }); + return next; + }); + }, + [], + ); + + const removePeer = useCallback((senderId: string) => { + setPeers((prev) => { + if (!prev.has(senderId)) return prev; + const next = new Map(prev); + next.delete(senderId); + return next; + }); + }, []); + + // Connect + handle presence events. + useEffect(() => { + if (!enabled || !userId) return; + let cancelled = false; + let socket: Socket | null = null; + + (async () => { + let token: string; + let url: string; + try { + const res = await fetch(`/api/orgs/${githubLogin}/canvas/relay-token`); + if (!res.ok) { + console.warn("[canvas-relay] token fetch failed:", res.status); + return; + } + ({ token, url } = await res.json()); + } catch (err) { + console.error("[canvas-relay] token fetch error:", err); + return; + } + if (cancelled) return; + + socket = io(url, { auth: { token }, transports: ["websocket"] }); + socketRef.current = socket; + + socket.on("room:roster", (data: { collaborators: RosterEntry[] }) => { + for (const c of data.collaborators) { + upsertPeer(c.senderId, { + userId: c.odinguserId, + name: c.name, + color: c.color, + image: c.image, + }); + } + }); + + socket.on("user:join", (data: { user: RosterEntry }) => { + upsertPeer(data.user.senderId, { + userId: data.user.odinguserId, + name: data.user.name, + color: data.user.color, + image: data.user.image, + }); + }); + + socket.on("user:leave", (data: { senderId: string }) => { + removePeer(data.senderId); + }); + + socket.on( + "cursor:update", + (data: { + senderId: string; + userId?: string; + cursor: { x: number; y: number } | null; + color: string; + selectedNodeId: string | null; + canvasRef: string; + username?: string; + }) => { + upsertPeer(data.senderId, { + cursor: data.cursor, + color: data.color, + selectedNodeId: data.selectedNodeId, + cursorRef: data.canvasRef ?? "", + // Carry identity so the self-filter works even if we never saw a + // roster/join for this socket (e.g. a reconnect ghost or another + // tab of the same user). + ...(data.userId ? { userId: data.userId } : {}), + ...(data.username ? { name: data.username } : {}), + }); + }, + ); + })(); + + return () => { + cancelled = true; + socket?.disconnect(); + socketRef.current = null; + setPeers(new Map()); + }; + }, [githubLogin, userId, enabled, upsertPeer, removePeer]); + + // Broadcast cursor via pointermove (throttled), in canvas space. + useEffect(() => { + const el = containerRef.current; + if (!el || !enabled || !userId) return; + let lastFired = 0; + + const onMove = (e: PointerEvent) => { + const now = Date.now(); + if (now - lastFired < CURSOR_THROTTLE_MS) return; + lastFired = now; + + const vp = getViewport(); + const svgEl = getSvgElement(); + const rect = (svgEl ?? el).getBoundingClientRect(); + const canvasX = (e.clientX - rect.left - vp.x) / vp.zoom; + const canvasY = (e.clientY - rect.top - vp.y) / vp.zoom; + lastCursorRef.current = { x: canvasX, y: canvasY }; + + socketRef.current?.emit("cursor:update", { + cursor: { x: canvasX, y: canvasY }, + color: teamAvatarColor(userId), + selectedNodeId: selectedNodeIdRef.current, + canvasRef: canvasRefRef.current, + }); + }; + + el.addEventListener("pointermove", onMove); + return () => el.removeEventListener("pointermove", onMove); + }, [containerRef, getViewport, getSvgElement, userId, enabled]); + + // Broadcast selection changes (piggybacks on cursor:update with last cursor). + const prevSelected = useRef(undefined); + useEffect(() => { + if (!enabled || !userId) return; + if ( + prevSelected.current === undefined && + (selectedNodeId === null || selectedNodeId === undefined) + ) { + prevSelected.current = selectedNodeId ?? null; + return; + } + if (prevSelected.current === (selectedNodeId ?? null)) return; + prevSelected.current = selectedNodeId ?? null; + + socketRef.current?.emit("cursor:update", { + cursor: lastCursorRef.current, + color: teamAvatarColor(userId), + selectedNodeId: selectedNodeId ?? null, + canvasRef: canvasRefRef.current, + }); + }, [selectedNodeId, userId, enabled]); + + // Client-side TTL prune. + useEffect(() => { + if (!enabled) return; + const interval = setInterval(() => { + const cutoff = Date.now() - CLIENT_TTL_MS; + setPeers((prev) => { + let changed = false; + const next = new Map(prev); + for (const [id, entry] of prev) { + if (entry.lastSeenAt < cutoff) { + next.delete(id); + changed = true; + } + } + return changed ? next : prev; + }); + }, PRUNE_INTERVAL_MS); + return () => clearInterval(interval); + }, [enabled]); + + // Derive collaborators (self-filtered; cursor/selection only for THIS ref). + const collaborators: CanvasCollaboratorInfo[] = Array.from(peers.values()) + .filter((p) => p.userId !== userId) + .map((p) => ({ + id: p.userId || "unknown", + name: p.name, + color: p.color, + image: p.image, + cursor: p.cursorRef === canvasRef ? p.cursor : null, + selectedNodeId: + p.cursorRef === canvasRef ? p.selectedNodeId ?? undefined : undefined, + })); + + return { collaborators }; +} From b2d298bf7a7ee443ea30bbfca3579df1cb2210ec Mon Sep 17 00:00:00 2001 From: kelmith Date: Tue, 14 Jul 2026 16:02:57 +0530 Subject: [PATCH 2/3] feat: yjs position sync for canvas --- package-lock.json | 70 ++++++++++++ package.json | 2 + .../connections/OrgCanvasBackground.tsx | 71 ++++++++++-- src/hooks/useCanvasCollaborationViaRelay.ts | 102 +++++++++++++++++- 4 files changed, 235 insertions(+), 10 deletions(-) diff --git a/package-lock.json b/package-lock.json index 37b09f3c1f..c3e2f95c19 100644 --- a/package-lock.json +++ b/package-lock.json @@ -134,6 +134,8 @@ "task-master-ai": "^0.19.0", "three": "^0.180.0", "tiny-secp256k1": "^2.2.4", + "y-protocols": "^1.0.7", + "yjs": "^13.6.31", "zod": "^4.1.12", "zustand": "^5.0.6" }, @@ -23156,6 +23158,16 @@ "integrity": "sha512-4hLB8Py4zZce5s4yd9XzopqwVv/yGNhV1Bl8NTmCq1763HeK2+EwVTv+leGeL13Dnh2wfbqowVPXCIO0z4taYw==", "license": "(MIT AND Zlib)" }, + "node_modules/isomorphic.js": { + "version": "0.2.5", + "resolved": "https://registry.npmjs.org/isomorphic.js/-/isomorphic.js-0.2.5.tgz", + "integrity": "sha512-PIeMbHqMt4DnUP3MA/Flc0HElYjMXArsw1qwJZcm9sqR8mq3l8NYizFMty0pWwE/tzIGH3EKK5+jes5mAr85yw==", + "license": "MIT", + "funding": { + "type": "GitHub Sponsors ❤", + "url": "https://github.com/sponsors/dmonad" + } + }, "node_modules/iterator.prototype": { "version": "1.1.5", "resolved": "https://registry.npmjs.org/iterator.prototype/-/iterator.prototype-1.1.5.tgz", @@ -23680,6 +23692,27 @@ "node": ">= 0.8.0" } }, + "node_modules/lib0": { + "version": "0.2.117", + "resolved": "https://registry.npmjs.org/lib0/-/lib0-0.2.117.tgz", + "integrity": "sha512-DeXj9X5xDCjgKLU/7RR+/HQEVzuuEUiwldwOGsHK/sfAfELGWEyTcf0x+uOvCvK3O2zPmZePXWL85vtia6GyZw==", + "license": "MIT", + "dependencies": { + "isomorphic.js": "^0.2.4" + }, + "bin": { + "0ecdsa-generate-keypair": "bin/0ecdsa-generate-keypair.js", + "0gentesthtml": "bin/gentesthtml.js", + "0serve": "bin/0serve.js" + }, + "engines": { + "node": ">=16" + }, + "funding": { + "type": "GitHub Sponsors ❤", + "url": "https://github.com/sponsors/dmonad" + } + }, "node_modules/lie": { "version": "3.3.0", "resolved": "https://registry.npmjs.org/lie/-/lie-3.3.0.tgz", @@ -32569,6 +32602,26 @@ "node": ">=0.4" } }, + "node_modules/y-protocols": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/y-protocols/-/y-protocols-1.0.7.tgz", + "integrity": "sha512-YSVsLoXxO67J6eE/nV4AtFtT3QEotZf5sK5BHxFBXso7VDUT3Tx07IfA6hsu5Q5OmBdMkQVmFZ9QOA7fikWvnw==", + "license": "MIT", + "dependencies": { + "lib0": "^0.2.85" + }, + "engines": { + "node": ">=16.0.0", + "npm": ">=8.0.0" + }, + "funding": { + "type": "GitHub Sponsors ❤", + "url": "https://github.com/sponsors/dmonad" + }, + "peerDependencies": { + "yjs": "^13.0.0" + } + }, "node_modules/y18n": { "version": "5.0.8", "resolved": "https://registry.npmjs.org/y18n/-/y18n-5.0.8.tgz", @@ -32643,6 +32696,23 @@ "node": ">=8" } }, + "node_modules/yjs": { + "version": "13.6.31", + "resolved": "https://registry.npmjs.org/yjs/-/yjs-13.6.31.tgz", + "integrity": "sha512-Eq+5BRfbeGyqGVrTJL3bEcr8gKkxPuyuoHmAwpk52fDb8kOVMrfVSTRPd6yiGgX5Fskb96qCRjzjbRjrL4YEnw==", + "license": "MIT", + "dependencies": { + "lib0": "^0.2.99" + }, + "engines": { + "node": ">=16.0.0", + "npm": ">=8.0.0" + }, + "funding": { + "type": "GitHub Sponsors ❤", + "url": "https://github.com/sponsors/dmonad" + } + }, "node_modules/yocto-queue": { "version": "0.1.0", "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-0.1.0.tgz", diff --git a/package.json b/package.json index b163abc8c5..3f1ddd2d61 100644 --- a/package.json +++ b/package.json @@ -167,6 +167,8 @@ "task-master-ai": "^0.19.0", "three": "^0.180.0", "tiny-secp256k1": "^2.2.4", + "y-protocols": "^1.0.7", + "yjs": "^13.6.31", "zod": "^4.1.12", "zustand": "^5.0.6" }, diff --git a/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx b/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx index b3f77cb7de..1bf3851e7e 100644 --- a/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx +++ b/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx @@ -14,6 +14,7 @@ import { type EdgeUpdate, type NodeContextMenuConfig, type NodeMenuOption, + type NodeUpdate, type SystemCanvasHandle, } from "system-canvas-react"; import type { @@ -736,8 +737,10 @@ export function OrgCanvasBackground({ refreshRootHiddenLive, }); - // Real-time canvas presence — cursors, selection halos, conflict flash - const { collaborators } = useCanvasCollaborationViaRelay({ + // Real-time canvas presence — cursors, selection halos, conflict flash — + // plus live node-position sync (moves broadcast over the same socket). + const { collaborators, positionOverrides, publishNodePositions } = + useCanvasCollaborationViaRelay({ githubLogin, canvasRef: currentRef, userId: session?.user?.id ?? "", @@ -750,6 +753,54 @@ export function OrgCanvasBackground({ enabled: !!(session?.user?.id), }); + // Broadcast node moves to collaborators (positions overlay), then run the + // normal handler (local state + Postgres autosave). `canvasRef` undefined + // means the root canvas ("" key). + const handleNodeUpdateSync = useCallback( + (id: string, patch: NodeUpdate, canvasRef: string | undefined) => { + if (typeof patch.x === "number" || typeof patch.y === "number") { + publishNodePositions([ + { ref: canvasRef ?? "", id, x: patch.x, y: patch.y }, + ]); + } + handleNodeUpdate(id, patch, canvasRef); + }, + [handleNodeUpdate, publishNodePositions], + ); + + const handleNodesUpdateSync = useCallback( + ( + updates: { id: string; patch: NodeUpdate }[], + canvasRef: string | undefined, + ) => { + const moves = updates + .filter( + (u) => typeof u.patch.x === "number" || typeof u.patch.y === "number", + ) + .map((u) => ({ ref: canvasRef ?? "", id: u.id, x: u.patch.x, y: u.patch.y })); + if (moves.length > 0) publishNodePositions(moves); + handleNodesUpdate(updates, canvasRef); + }, + [handleNodesUpdate, publishNodePositions], + ); + + // Merge collaborators' live positions onto a canvas at render. Unmoved + // nodes keep their projection position; overrides win per-coordinate. + const applyPositionOverrides = useCallback( + (data: CanvasData, ref: string): CanvasData => { + if (positionOverrides.size === 0) return data; + let changed = false; + const nodes = (data.nodes ?? []).map((n) => { + const o = positionOverrides.get(`${ref}:${n.id}`); + if (!o) return n; + changed = true; + return { ...n, x: o.x ?? n.x, y: o.y ?? n.y }; + }); + return changed ? { ...data, nodes } : data; + }, + [positionOverrides], + ); + const { handleEdgeAdd, @@ -803,8 +854,12 @@ export function OrgCanvasBackground({ ); const canvasForRender = useMemo( - () => decorateEdgesWithLinkVisual(root ?? { nodes: [], edges: [] }), - [root, decorateEdgesWithLinkVisual], + () => + applyPositionOverrides( + decorateEdgesWithLinkVisual(root ?? { nodes: [], edges: [] }), + "", + ), + [root, decorateEdgesWithLinkVisual, applyPositionOverrides], ); // Sub-canvases also need decoration so links are highlighted on @@ -813,10 +868,10 @@ export function OrgCanvasBackground({ const subCanvasesForRender = useMemo>(() => { const out: Record = {}; for (const [ref, data] of Object.entries(subCanvases)) { - out[ref] = decorateEdgesWithLinkVisual(data); + out[ref] = applyPositionOverrides(decorateEdgesWithLinkVisual(data), ref); } return out; - }, [subCanvases, decorateEdgesWithLinkVisual]); + }, [subCanvases, decorateEdgesWithLinkVisual, applyPositionOverrides]); // Set of connection ids referenced by at least one edge across all // canvases we've loaded this session. Walked off the same `root` + @@ -1191,8 +1246,8 @@ export function OrgCanvasBackground({ onBreadcrumbsChange={handleBreadcrumbsChange} onSelectionChange={handleSelectionChange} onNodeAdd={handleNodeAdd} - onNodeUpdate={handleNodeUpdate} - onNodesUpdate={handleNodesUpdate} + onNodeUpdate={handleNodeUpdateSync} + onNodesUpdate={handleNodesUpdateSync} onNodeDelete={handleNodeDelete} onNodesDelete={handleNodesDelete} onEdgeAdd={handleEdgeAdd} diff --git a/src/hooks/useCanvasCollaborationViaRelay.ts b/src/hooks/useCanvasCollaborationViaRelay.ts index 24f3a9c182..df29f0d441 100644 --- a/src/hooks/useCanvasCollaborationViaRelay.ts +++ b/src/hooks/useCanvasCollaborationViaRelay.ts @@ -1,12 +1,30 @@ "use client"; -import { useCallback, useEffect, useRef, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { io, type Socket } from "socket.io-client"; +import * as Y from "yjs"; import type { CanvasCollaboratorInfo, UseCanvasCollaborationOptions, } from "./useCanvasCollaboration"; +/** A live position override for a node, keyed `${canvasRef}:${nodeId}`. */ +export interface NodePositionOverride { + x?: number; + y?: number; +} + +/** A node move to publish: `ref` is the sub-canvas ("" = root). */ +export interface NodePositionUpdate { + ref: string; + id: string; + x?: number; + y?: number; +} + +const LOCAL_ORIGIN = "local"; +const REMOTE_ORIGIN = "remote"; + /** * Relay (Socket.IO) drop-in replacement for `useCanvasCollaboration`. * @@ -19,6 +37,14 @@ import type { interface UseCanvasCollaborationResult { collaborators: CanvasCollaboratorInfo[]; + /** + * Live node-position overrides from collaborators, keyed + * `${canvasRef}:${nodeId}` ("" ref = root). Merge these onto the + * projection at render so remote moves show without a refetch. + */ + positionOverrides: Map; + /** Broadcast local node moves to collaborators (conflict-free, LWW per node). */ + publishNodePositions: (updates: NodePositionUpdate[]) => void; } const CLIENT_TTL_MS = 60_000; @@ -83,6 +109,22 @@ export function useCanvasCollaborationViaRelay({ selectedNodeIdRef.current = selectedNodeId; const lastCursorRef = useRef<{ x: number; y: number } | null>(null); + // ---- Document sync (node positions) over the SAME socket ------------- + // A single Y.Doc per org canvas holds a positions map keyed by + // `${canvasRef}:${nodeId}`. Positions are LWW per node, so peers that + // seed independently from the same Postgres projection converge. The + // map starts empty — unmoved nodes fall back to the projection. + const docRef = useRef(null); + if (!docRef.current) docRef.current = new Y.Doc(); + const doc = docRef.current; + const positions = useMemo( + () => doc.getMap("positions"), + [doc], + ); + const [positionOverrides, setPositionOverrides] = useState< + Map + >(new Map()); + const upsertPeer = useCallback( (senderId: string, patch: Partial) => { setPeers((prev) => { @@ -186,6 +228,19 @@ export function useCanvasCollaborationViaRelay({ }); }, ); + + // Binary Yjs position deltas from collaborators. + socket.on("yupdate", (data: { update: ArrayBuffer | Uint8Array }) => { + try { + const u = + data.update instanceof Uint8Array + ? data.update + : new Uint8Array(data.update); + Y.applyUpdate(doc, u, REMOTE_ORIGIN); + } catch (err) { + console.warn("[canvas-relay] bad yupdate:", err); + } + }); })(); return () => { @@ -268,6 +323,49 @@ export function useCanvasCollaborationViaRelay({ return () => clearInterval(interval); }, [enabled]); + // Broadcast local position deltas + mirror the Y.Map into React state. + // Yjs `update` fires once per transaction with that transaction's binary + // delta — exactly what we send. Local deltas go out; remote ones don't + // echo (guarded by origin). + useEffect(() => { + if (!enabled) return; + const onUpdate = (update: Uint8Array, origin: unknown) => { + if (origin === LOCAL_ORIGIN) { + socketRef.current?.emit("yupdate", { update }); + } + const next = new Map(); + positions.forEach((v, k) => next.set(k, v)); + setPositionOverrides(next); + }; + doc.on("update", onUpdate); + return () => { + doc.off("update", onUpdate); + }; + }, [doc, positions, enabled]); + + // Destroy the doc on unmount (frees observers + memory). + useEffect(() => { + const d = docRef.current; + return () => { + d?.destroy(); + docRef.current = null; + }; + }, []); + + const publishNodePositions = useCallback( + (updates: NodePositionUpdate[]) => { + if (updates.length === 0) return; + doc.transact(() => { + for (const u of updates) { + const key = `${u.ref}:${u.id}`; + const cur = positions.get(key) ?? {}; + positions.set(key, { x: u.x ?? cur.x, y: u.y ?? cur.y }); + } + }, LOCAL_ORIGIN); + }, + [doc, positions], + ); + // Derive collaborators (self-filtered; cursor/selection only for THIS ref). const collaborators: CanvasCollaboratorInfo[] = Array.from(peers.values()) .filter((p) => p.userId !== userId) @@ -281,5 +379,5 @@ export function useCanvasCollaborationViaRelay({ p.cursorRef === canvasRef ? p.selectedNodeId ?? undefined : undefined, })); - return { collaborators }; + return { collaborators, positionOverrides, publishNodePositions }; } From 7bbc2c750b65355a8f09e6b80195cdee90e83ebb Mon Sep 17 00:00:00 2001 From: kelmith Date: Tue, 14 Jul 2026 20:12:19 +0530 Subject: [PATCH 3/3] feat: full crdt document sync for canvas --- package-lock.json | 17 ++ package.json | 1 + .../connections/OrgCanvasBackground.tsx | 165 ++++++---- src/hooks/useCanvasCollaborationViaRelay.ts | 288 ++++++++++++------ 4 files changed, 314 insertions(+), 157 deletions(-) diff --git a/package-lock.json b/package-lock.json index c3e2f95c19..25f1acad9d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -129,6 +129,7 @@ "sonner": "^2.0.7", "stripe": "^21.0.1", "system-canvas": "^0.2.57", + "system-canvas-collab": "file:../system-canvas/packages/collab/system-canvas-collab-0.2.57.tgz", "system-canvas-react": "^0.2.57", "tailwind-merge": "^3.3.1", "task-master-ai": "^0.19.0", @@ -30088,6 +30089,22 @@ "integrity": "sha512-0gpb9ePVupbjMrYfyYZfnvcalYUvfktC9VX7SIO2LPeNsCmA5KtSxjRlRfVWJa8aECtFRjORhM+5vbKaaq9w3w==", "license": "MIT" }, + "node_modules/system-canvas-collab": { + "version": "0.2.57", + "resolved": "file:../system-canvas/packages/collab/system-canvas-collab-0.2.57.tgz", + "integrity": "sha512-ts0nwNpVVJApJ+1RPSsJT7/qNORtW4K/1NCt+LemRdY6LXEmCoTFVhi5skeuo8YylUxlOOAx7hjJO3FlETwMOw==", + "license": "MIT", + "peerDependencies": { + "react": "^18.0.0 || ^19.0.0", + "system-canvas": "^0.2.57", + "yjs": "^13.6.0" + }, + "peerDependenciesMeta": { + "react": { + "optional": true + } + } + }, "node_modules/system-canvas-react": { "version": "0.2.57", "resolved": "https://registry.npmjs.org/system-canvas-react/-/system-canvas-react-0.2.57.tgz", diff --git a/package.json b/package.json index 3f1ddd2d61..105e74875d 100644 --- a/package.json +++ b/package.json @@ -162,6 +162,7 @@ "sonner": "^2.0.7", "stripe": "^21.0.1", "system-canvas": "^0.2.57", + "system-canvas-collab": "file:../system-canvas/packages/collab/system-canvas-collab-0.2.57.tgz", "system-canvas-react": "^0.2.57", "tailwind-merge": "^3.3.1", "task-master-ai": "^0.19.0", diff --git a/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx b/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx index 1bf3851e7e..33c502f4a4 100644 --- a/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx +++ b/src/app/org/[githubLogin]/connections/OrgCanvasBackground.tsx @@ -737,10 +737,10 @@ export function OrgCanvasBackground({ refreshRootHiddenLive, }); - // Real-time canvas presence — cursors, selection halos, conflict flash — - // plus live node-position sync (moves broadcast over the same socket). - const { collaborators, positionOverrides, publishNodePositions } = - useCanvasCollaborationViaRelay({ + // Real-time canvas presence (cursors/selection) + conflict-free document + // sync (nodes+edges) over the same relay socket. One Y.Doc per sub-canvas + // ref is the live source, seeded from the Postgres projection. + const { collaborators, crdt } = useCanvasCollaborationViaRelay({ githubLogin, canvasRef: currentRef, userId: session?.user?.id ?? "", @@ -752,54 +752,26 @@ export function OrgCanvasBackground({ selectedNodeId: selectedNodeIdForPresence, enabled: !!(session?.user?.id), }); + const { + reconstruct: crdtReconstruct, + seed: crdtSeed, + version: crdtVersion, + addNode: crdtAddNode, + updateNode: crdtUpdateNode, + removeNode: crdtRemoveNode, + addEdge: crdtAddEdge, + updateEdge: crdtUpdateEdge, + removeEdge: crdtRemoveEdge, + } = crdt; - // Broadcast node moves to collaborators (positions overlay), then run the - // normal handler (local state + Postgres autosave). `canvasRef` undefined - // means the root canvas ("" key). - const handleNodeUpdateSync = useCallback( - (id: string, patch: NodeUpdate, canvasRef: string | undefined) => { - if (typeof patch.x === "number" || typeof patch.y === "number") { - publishNodePositions([ - { ref: canvasRef ?? "", id, x: patch.x, y: patch.y }, - ]); - } - handleNodeUpdate(id, patch, canvasRef); - }, - [handleNodeUpdate, publishNodePositions], - ); - - const handleNodesUpdateSync = useCallback( - ( - updates: { id: string; patch: NodeUpdate }[], - canvasRef: string | undefined, - ) => { - const moves = updates - .filter( - (u) => typeof u.patch.x === "number" || typeof u.patch.y === "number", - ) - .map((u) => ({ ref: canvasRef ?? "", id: u.id, x: u.patch.x, y: u.patch.y })); - if (moves.length > 0) publishNodePositions(moves); - handleNodesUpdate(updates, canvasRef); - }, - [handleNodesUpdate, publishNodePositions], - ); - - // Merge collaborators' live positions onto a canvas at render. Unmoved - // nodes keep their projection position; overrides win per-coordinate. - const applyPositionOverrides = useCallback( - (data: CanvasData, ref: string): CanvasData => { - if (positionOverrides.size === 0) return data; - let changed = false; - const nodes = (data.nodes ?? []).map((n) => { - const o = positionOverrides.get(`${ref}:${n.id}`); - if (!o) return n; - changed = true; - return { ...n, x: o.x ?? n.x, y: o.y ?? n.y }; - }); - return changed ? { ...data, nodes } : data; - }, - [positionOverrides], - ); + // Seed each ref's doc from the projection as it loads. Idempotent, and a + // no-op unless we're the first client in the room (others sync from peers). + useEffect(() => { + if (root) crdtSeed("", root); + }, [root, crdtSeed]); + useEffect(() => { + for (const [ref, data] of Object.entries(subCanvases)) crdtSeed(ref, data); + }, [subCanvases, crdtSeed]); const { @@ -815,6 +787,69 @@ export function OrgCanvasBackground({ canvasHandleRef, subCanvasesRef, }); + + // Route every structural edit through the CRDT doc (live sync to peers) AND + // the existing handler (Postgres persist + domain bridges). `canvasRef` + // undefined = the root canvas ("" key). + const handleNodeAddSync = useCallback( + (node: CanvasNode, canvasRef: string | undefined) => { + crdtAddNode(canvasRef ?? "", node); + handleNodeAdd(node, canvasRef); + }, + [crdtAddNode, handleNodeAdd], + ); + const handleNodeUpdateSync = useCallback( + (id: string, patch: NodeUpdate, canvasRef: string | undefined) => { + crdtUpdateNode(canvasRef ?? "", id, patch); + handleNodeUpdate(id, patch, canvasRef); + }, + [crdtUpdateNode, handleNodeUpdate], + ); + const handleNodesUpdateSync = useCallback( + ( + updates: { id: string; patch: NodeUpdate }[], + canvasRef: string | undefined, + ) => { + for (const u of updates) crdtUpdateNode(canvasRef ?? "", u.id, u.patch); + handleNodesUpdate(updates, canvasRef); + }, + [crdtUpdateNode, handleNodesUpdate], + ); + const handleNodeDeleteSync = useCallback( + (id: string, canvasRef: string | undefined) => { + crdtRemoveNode(canvasRef ?? "", id); + handleNodeDelete(id, canvasRef); + }, + [crdtRemoveNode, handleNodeDelete], + ); + const handleNodesDeleteSync = useCallback( + (ids: string[], canvasRef: string | undefined) => { + for (const id of ids) crdtRemoveNode(canvasRef ?? "", id); + handleNodesDelete(ids, canvasRef); + }, + [crdtRemoveNode, handleNodesDelete], + ); + const handleEdgeAddSync = useCallback( + (edge: CanvasEdge, canvasRef: string | undefined) => { + crdtAddEdge(canvasRef ?? "", edge); + handleEdgeAdd(edge, canvasRef); + }, + [crdtAddEdge, handleEdgeAdd], + ); + const handleEdgeUpdateSync = useCallback( + (id: string, patch: EdgeUpdate, canvasRef: string | undefined) => { + crdtUpdateEdge(canvasRef ?? "", id, patch); + handleEdgeUpdate(id, patch, canvasRef); + }, + [crdtUpdateEdge, handleEdgeUpdate], + ); + const handleEdgeDeleteSync = useCallback( + (id: string, canvasRef: string | undefined) => { + crdtRemoveEdge(canvasRef ?? "", id); + handleEdgeDelete(id, canvasRef); + }, + [crdtRemoveEdge, handleEdgeDelete], + ); /** * Visually highlight edges that have a linked Connection doc. The * lib renders edges using `theme.edge.stroke` by default; setting @@ -853,13 +888,16 @@ export function OrgCanvasBackground({ [], ); + // The live CRDT doc is the render source once seeded/synced; until then we + // fall back to the raw projection. `crdtVersion` bumps on any doc change so + // remote edits re-render without a refetch. const canvasForRender = useMemo( () => - applyPositionOverrides( - decorateEdgesWithLinkVisual(root ?? { nodes: [], edges: [] }), - "", + decorateEdgesWithLinkVisual( + crdtReconstruct("") ?? root ?? { nodes: [], edges: [] }, ), - [root, decorateEdgesWithLinkVisual, applyPositionOverrides], + // eslint-disable-next-line react-hooks/exhaustive-deps + [crdtVersion, root, decorateEdgesWithLinkVisual, crdtReconstruct], ); // Sub-canvases also need decoration so links are highlighted on @@ -868,10 +906,11 @@ export function OrgCanvasBackground({ const subCanvasesForRender = useMemo>(() => { const out: Record = {}; for (const [ref, data] of Object.entries(subCanvases)) { - out[ref] = applyPositionOverrides(decorateEdgesWithLinkVisual(data), ref); + out[ref] = decorateEdgesWithLinkVisual(crdtReconstruct(ref) ?? data); } return out; - }, [subCanvases, decorateEdgesWithLinkVisual, applyPositionOverrides]); + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [crdtVersion, subCanvases, decorateEdgesWithLinkVisual, crdtReconstruct]); // Set of connection ids referenced by at least one edge across all // canvases we've loaded this session. Walked off the same `root` + @@ -1245,14 +1284,14 @@ export function OrgCanvasBackground({ onResolveCanvas={onResolveCanvas} onBreadcrumbsChange={handleBreadcrumbsChange} onSelectionChange={handleSelectionChange} - onNodeAdd={handleNodeAdd} + onNodeAdd={handleNodeAddSync} onNodeUpdate={handleNodeUpdateSync} onNodesUpdate={handleNodesUpdateSync} - onNodeDelete={handleNodeDelete} - onNodesDelete={handleNodesDelete} - onEdgeAdd={handleEdgeAdd} - onEdgeUpdate={handleEdgeUpdate} - onEdgeDelete={handleEdgeDelete} + onNodeDelete={handleNodeDeleteSync} + onNodesDelete={handleNodesDeleteSync} + onEdgeAdd={handleEdgeAddSync} + onEdgeUpdate={handleEdgeUpdateSync} + onEdgeDelete={handleEdgeDeleteSync} canDropNodeOn={canDropNodeOn} onNodeDrop={handleNodeDrop} nodeContextMenu={nodeContextMenu} diff --git a/src/hooks/useCanvasCollaborationViaRelay.ts b/src/hooks/useCanvasCollaborationViaRelay.ts index df29f0d441..0844c5204c 100644 --- a/src/hooks/useCanvasCollaborationViaRelay.ts +++ b/src/hooks/useCanvasCollaborationViaRelay.ts @@ -3,48 +3,52 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { io, type Socket } from "socket.io-client"; import * as Y from "yjs"; +import { + LOCAL_ORIGIN, + seedYDoc, + yDocToCanvasData, + addNode as docAddNode, + updateNode as docUpdateNode, + removeNode as docRemoveNode, + addEdge as docAddEdge, + updateEdge as docUpdateEdge, + removeEdge as docRemoveEdge, +} from "system-canvas-collab"; +import type { CanvasData, CanvasNode, CanvasEdge } from "system-canvas"; import type { CanvasCollaboratorInfo, UseCanvasCollaborationOptions, } from "./useCanvasCollaboration"; -/** A live position override for a node, keyed `${canvasRef}:${nodeId}`. */ -export interface NodePositionOverride { - x?: number; - y?: number; -} - -/** A node move to publish: `ref` is the sub-canvas ("" = root). */ -export interface NodePositionUpdate { - ref: string; - id: string; - x?: number; - y?: number; -} - -const LOCAL_ORIGIN = "local"; -const REMOTE_ORIGIN = "remote"; +/** Origins that must NOT re-broadcast (only user edits, tagged LOCAL_ORIGIN, do). */ +const REMOTE_ORIGIN = Symbol("remote"); +const SEED_ORIGIN = Symbol("seed"); /** - * Relay (Socket.IO) drop-in replacement for `useCanvasCollaboration`. - * - * Same options + return shape, but presence rides the per-swarm hive-relay - * over a persistent WebSocket instead of Pusher + a POST-per-cursor-move. - * The room is per-org (`canvas:`); each cursor carries its - * `canvasRef` so peers only render cursors/selection for the sub-canvas - * they're currently viewing. + * Conflict-free document layer for the org canvas, keyed by sub-canvas ref. + * Each ref gets its own Y.Doc (system-canvas-collab's binding); the ref that + * is first in the room seeds from the Postgres projection, late-joiners sync + * the live doc from peers (no re-seed). Edits + the binary deltas ride the + * SAME relay socket as presence. */ +export interface CanvasCrdt { + /** Reconstruct a ref's live CanvasData, or null until seeded/synced. */ + reconstruct: (ref: string) => CanvasData | null; + /** Seed a ref's doc from the projection (takes effect only if first in room). */ + seed: (ref: string, data: CanvasData) => void; + addNode: (ref: string, node: CanvasNode) => void; + updateNode: (ref: string, id: string, patch: Partial) => void; + removeNode: (ref: string, id: string) => void; + addEdge: (ref: string, edge: CanvasEdge) => void; + updateEdge: (ref: string, id: string, patch: Partial) => void; + removeEdge: (ref: string, id: string) => void; + /** Bumps on any doc change (local or remote) — use as a render memo dep. */ + version: number; +} interface UseCanvasCollaborationResult { collaborators: CanvasCollaboratorInfo[]; - /** - * Live node-position overrides from collaborators, keyed - * `${canvasRef}:${nodeId}` ("" ref = root). Merge these onto the - * projection at render so remote moves show without a refetch. - */ - positionOverrides: Map; - /** Broadcast local node moves to collaborators (conflict-free, LWW per node). */ - publishNodePositions: (updates: NodePositionUpdate[]) => void; + crdt: CanvasCrdt; } const CLIENT_TTL_MS = 60_000; @@ -109,21 +113,118 @@ export function useCanvasCollaborationViaRelay({ selectedNodeIdRef.current = selectedNodeId; const lastCursorRef = useRef<{ x: number; y: number } | null>(null); - // ---- Document sync (node positions) over the SAME socket ------------- - // A single Y.Doc per org canvas holds a positions map keyed by - // `${canvasRef}:${nodeId}`. Positions are LWW per node, so peers that - // seed independently from the same Postgres projection converge. The - // map starts empty — unmoved nodes fall back to the projection. - const docRef = useRef(null); - if (!docRef.current) docRef.current = new Y.Doc(); - const doc = docRef.current; - const positions = useMemo( - () => doc.getMap("positions"), - [doc], + // ---- Document CRDT (full nodes+edges, one Y.Doc per ref) -------------- + const docsRef = useRef>(new Map()); + const readyRef = useRef>(new Set()); + const seededRef = useRef>(new Set()); + const pendingSeedsRef = useRef>(new Map()); + // "pending" until room:roster tells us if we're first (seed) or not (sync). + const roomStateRef = useRef<"pending" | "first" | "existing">("pending"); + const [docVersion, setDocVersion] = useState(0); + const bumpVersion = useCallback(() => setDocVersion((v) => v + 1), []); + + // Get-or-create a ref's doc, wiring its broadcast/render handler once. + const getDoc = useCallback( + (ref: string): Y.Doc => { + const existing = docsRef.current.get(ref); + if (existing) return existing; + const doc = new Y.Doc(); + doc.on("update", (update: Uint8Array, origin: unknown) => { + if (origin === LOCAL_ORIGIN) { + socketRef.current?.emit("yupdate", { ref, update }); + } + bumpVersion(); + }); + docsRef.current.set(ref, doc); + return doc; + }, + [bumpVersion], + ); + + const applyRemote = useCallback( + (ref: string, update: Uint8Array) => { + try { + Y.applyUpdate(getDoc(ref), update, REMOTE_ORIGIN); + readyRef.current.add(ref); + } catch (err) { + console.warn("[canvas-relay] bad yupdate:", err); + } + }, + [getDoc], + ); + + const seedRef = useCallback( + (ref: string, data: CanvasData) => { + if (seededRef.current.has(ref) || readyRef.current.has(ref)) return; + if (roomStateRef.current === "pending") { + pendingSeedsRef.current.set(ref, data); + return; + } + if (roomStateRef.current !== "first") return; // existing → sync from peers + seededRef.current.add(ref); + readyRef.current.add(ref); + seedYDoc(getDoc(ref), data, SEED_ORIGIN); + }, + [getDoc], + ); + + const reconstruct = useCallback((ref: string): CanvasData | null => { + if (!readyRef.current.has(ref)) return null; + const doc = docsRef.current.get(ref); + return doc ? yDocToCanvasData(doc) : null; + }, []); + + const crdtAddNode = useCallback( + (ref: string, node: CanvasNode) => docAddNode(getDoc(ref), node, LOCAL_ORIGIN), + [getDoc], + ); + const crdtUpdateNode = useCallback( + (ref: string, id: string, patch: Partial) => + docUpdateNode(getDoc(ref), id, patch, LOCAL_ORIGIN), + [getDoc], + ); + const crdtRemoveNode = useCallback( + (ref: string, id: string) => docRemoveNode(getDoc(ref), id, LOCAL_ORIGIN), + [getDoc], + ); + const crdtAddEdge = useCallback( + (ref: string, edge: CanvasEdge) => docAddEdge(getDoc(ref), edge, LOCAL_ORIGIN), + [getDoc], + ); + const crdtUpdateEdge = useCallback( + (ref: string, id: string, patch: Partial) => + docUpdateEdge(getDoc(ref), id, patch, LOCAL_ORIGIN), + [getDoc], + ); + const crdtRemoveEdge = useCallback( + (ref: string, id: string) => docRemoveEdge(getDoc(ref), id, LOCAL_ORIGIN), + [getDoc], + ); + + const crdt = useMemo( + () => ({ + reconstruct, + seed: seedRef, + addNode: crdtAddNode, + updateNode: crdtUpdateNode, + removeNode: crdtRemoveNode, + addEdge: crdtAddEdge, + updateEdge: crdtUpdateEdge, + removeEdge: crdtRemoveEdge, + version: docVersion, + }), + [ + reconstruct, + seedRef, + crdtAddNode, + crdtUpdateNode, + crdtRemoveNode, + crdtAddEdge, + crdtUpdateEdge, + crdtRemoveEdge, + docVersion, + ], ); - const [positionOverrides, setPositionOverrides] = useState< - Map - >(new Map()); const upsertPeer = useCallback( (senderId: string, patch: Partial) => { @@ -155,7 +256,17 @@ export function useCanvasCollaborationViaRelay({ }); }, []); - // Connect + handle presence events. + // Broadcast this client's current doc state (all refs) to the room, so a + // late-joiner converges without independently seeding. + const broadcastDocState = useCallback(() => { + const sock = socketRef.current; + if (!sock) return; + for (const [ref, doc] of docsRef.current) { + sock.emit("yupdate", { ref, update: Y.encodeStateAsUpdate(doc) }); + } + }, []); + + // Connect + handle presence + doc sync. useEffect(() => { if (!enabled || !userId) return; let cancelled = false; @@ -181,6 +292,19 @@ export function useCanvasCollaborationViaRelay({ socketRef.current = socket; socket.on("room:roster", (data: { collaborators: RosterEntry[] }) => { + // Decide seed-vs-sync: an empty room means we own the initial state. + roomStateRef.current = + data.collaborators.length > 0 ? "existing" : "first"; + if (roomStateRef.current === "first") { + for (const [ref, seedData] of pendingSeedsRef.current) { + if (seededRef.current.has(ref) || readyRef.current.has(ref)) continue; + seededRef.current.add(ref); + readyRef.current.add(ref); + seedYDoc(getDoc(ref), seedData, SEED_ORIGIN); + } + } + pendingSeedsRef.current.clear(); + for (const c of data.collaborators) { upsertPeer(c.senderId, { userId: c.odinguserId, @@ -198,6 +322,8 @@ export function useCanvasCollaborationViaRelay({ color: data.user.color, image: data.user.image, }); + // A newcomer arrived; hand them our live doc so they don't re-seed. + broadcastDocState(); }); socket.on("user:leave", (data: { senderId: string }) => { @@ -229,18 +355,17 @@ export function useCanvasCollaborationViaRelay({ }, ); - // Binary Yjs position deltas from collaborators. - socket.on("yupdate", (data: { update: ArrayBuffer | Uint8Array }) => { - try { + // Binary Yjs deltas from collaborators, routed to the ref's doc. + socket.on( + "yupdate", + (data: { ref?: string; update: ArrayBuffer | Uint8Array }) => { const u = data.update instanceof Uint8Array ? data.update : new Uint8Array(data.update); - Y.applyUpdate(doc, u, REMOTE_ORIGIN); - } catch (err) { - console.warn("[canvas-relay] bad yupdate:", err); - } - }); + applyRemote(data.ref ?? "", u); + }, + ); })(); return () => { @@ -249,7 +374,16 @@ export function useCanvasCollaborationViaRelay({ socketRef.current = null; setPeers(new Map()); }; - }, [githubLogin, userId, enabled, upsertPeer, removePeer]); + }, [ + githubLogin, + userId, + enabled, + upsertPeer, + removePeer, + getDoc, + applyRemote, + broadcastDocState, + ]); // Broadcast cursor via pointermove (throttled), in canvas space. useEffect(() => { @@ -323,49 +457,15 @@ export function useCanvasCollaborationViaRelay({ return () => clearInterval(interval); }, [enabled]); - // Broadcast local position deltas + mirror the Y.Map into React state. - // Yjs `update` fires once per transaction with that transaction's binary - // delta — exactly what we send. Local deltas go out; remote ones don't - // echo (guarded by origin). + // Destroy all docs on unmount (frees observers + memory). useEffect(() => { - if (!enabled) return; - const onUpdate = (update: Uint8Array, origin: unknown) => { - if (origin === LOCAL_ORIGIN) { - socketRef.current?.emit("yupdate", { update }); - } - const next = new Map(); - positions.forEach((v, k) => next.set(k, v)); - setPositionOverrides(next); - }; - doc.on("update", onUpdate); + const docs = docsRef.current; return () => { - doc.off("update", onUpdate); - }; - }, [doc, positions, enabled]); - - // Destroy the doc on unmount (frees observers + memory). - useEffect(() => { - const d = docRef.current; - return () => { - d?.destroy(); - docRef.current = null; + for (const doc of docs.values()) doc.destroy(); + docs.clear(); }; }, []); - const publishNodePositions = useCallback( - (updates: NodePositionUpdate[]) => { - if (updates.length === 0) return; - doc.transact(() => { - for (const u of updates) { - const key = `${u.ref}:${u.id}`; - const cur = positions.get(key) ?? {}; - positions.set(key, { x: u.x ?? cur.x, y: u.y ?? cur.y }); - } - }, LOCAL_ORIGIN); - }, - [doc, positions], - ); - // Derive collaborators (self-filtered; cursor/selection only for THIS ref). const collaborators: CanvasCollaboratorInfo[] = Array.from(peers.values()) .filter((p) => p.userId !== userId) @@ -379,5 +479,5 @@ export function useCanvasCollaborationViaRelay({ p.cursorRef === canvasRef ? p.selectedNodeId ?? undefined : undefined, })); - return { collaborators, positionOverrides, publishNodePositions }; + return { collaborators, crdt }; }