fix: HIGH issues #13,#14,#18,#19,#20 — frontend state/sync: copy-paste selection, dynamic activeLayer, group creation, Yjs awareness protocol, connect sync

This commit is contained in:
A0 Orchestrator
2026-06-30 12:42:49 +02:00
parent 5a21fc0bfa
commit d09b1dfc2e
8 changed files with 353 additions and 62 deletions
+20
View File
@@ -15,6 +15,7 @@
"better-sqlite3": "^11.0.0", "better-sqlite3": "^11.0.0",
"fastify": "^4.28.0", "fastify": "^4.28.0",
"y-leveldb": "^0.2.0", "y-leveldb": "^0.2.0",
"y-protocols": "^1.0.7",
"yjs": "^13.6.0" "yjs": "^13.6.0"
}, },
"devDependencies": { "devDependencies": {
@@ -3868,6 +3869,25 @@
"yjs": "^13.0.0" "yjs": "^13.0.0"
} }
}, },
"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==",
"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/yjs": { "node_modules/yjs": {
"version": "13.6.31", "version": "13.6.31",
"resolved": "https://registry.npmjs.org/yjs/-/yjs-13.6.31.tgz", "resolved": "https://registry.npmjs.org/yjs/-/yjs-13.6.31.tgz",
+1
View File
@@ -18,6 +18,7 @@
"better-sqlite3": "^11.0.0", "better-sqlite3": "^11.0.0",
"fastify": "^4.28.0", "fastify": "^4.28.0",
"y-leveldb": "^0.2.0", "y-leveldb": "^0.2.0",
"y-protocols": "^1.0.7",
"yjs": "^13.6.0" "yjs": "^13.6.0"
}, },
"devDependencies": { "devDependencies": {
+86 -9
View File
@@ -1,16 +1,27 @@
/** /**
* Yjs WebSocket Server Real-time collaboration with LevelDB persistence * Yjs WebSocket Server Real-time collaboration with LevelDB persistence.
*
* Message protocol (byte-prefixed):
* 0 = Y.Doc update (Y.encodeStateAsUpdate or Y.encodeStateAsUpdateYUpdate)
* 1 = Awareness update (encodeAwarenessUpdate from y-protocols/awareness)
*/ */
import * as Y from 'yjs'; import * as Y from 'yjs';
import { LeveldbPersistence } from 'y-leveldb'; import { LeveldbPersistence } from 'y-leveldb';
import { Awareness, encodeAwarenessUpdate, applyAwarenessUpdate, removeAwarenessStates } from 'y-protocols/awareness';
import type { FastifyInstance } from 'fastify'; import type { FastifyInstance } from 'fastify';
import type { WebSocket } from '@fastify/websocket'; import type { WebSocket } from '@fastify/websocket';
import type { AuthService } from '../auth/AuthService.js'; import type { AuthService } from '../auth/AuthService.js';
const PERSISTENCE_DIR = process.env.YJS_PERSISTENCE_DIR || '/tmp/yjs-documents'; const PERSISTENCE_DIR = process.env.YJS_PERSISTENCE_DIR || '/tmp/yjs-documents';
// Message type prefixes (must match frontend WebSocketProvider.ts)
const MSG_DOC_UPDATE = 0;
const MSG_AWARENESS = 1;
// Document store: docName → Y.Doc // Document store: docName → Y.Doc
const docs = new Map<string, Y.Doc>(); const docs = new Map<string, Y.Doc>();
// Awareness store: docName → Awareness
const awarenessMap = new Map<string, Awareness>();
// Connection tracking: docName → Set<WebSocket> // Connection tracking: docName → Set<WebSocket>
const connections = new Map<string, Set<WebSocket>>(); const connections = new Map<string, Set<WebSocket>>();
let persistence: LeveldbPersistence | null = null; let persistence: LeveldbPersistence | null = null;
@@ -46,6 +57,10 @@ async function getOrCreateDoc(docName: string): Promise<Y.Doc> {
const doc = new Y.Doc(); const doc = new Y.Doc();
docs.set(docName, doc); docs.set(docName, doc);
// Create awareness instance for this document
const awareness = new Awareness(doc);
awarenessMap.set(docName, awareness);
// Load persisted state if available // Load persisted state if available
if (persistence && persistenceReady) { if (persistence && persistenceReady) {
try { try {
@@ -72,14 +87,23 @@ async function getOrCreateDoc(docName: string): Promise<Y.Doc> {
return doc; return doc;
} }
function broadcastUpdate(docName: string, update: Uint8Array, exclude?: WebSocket): void { function getOrCreateAwareness(docName: string): Awareness | null {
return awarenessMap.get(docName) ?? null;
}
function broadcastUpdate(docName: string, update: Uint8Array, msgType: number, exclude?: WebSocket): void {
const conns = connections.get(docName); const conns = connections.get(docName);
if (!conns) return; if (!conns) return;
// Prepend message type byte
const msg = new Uint8Array(update.length + 1);
msg[0] = msgType;
msg.set(update, 1);
for (const ws of conns) { for (const ws of conns) {
if (ws !== exclude && ws.readyState === 1 /* OPEN */) { if (ws !== exclude && ws.readyState === 1 /* OPEN */) {
try { try {
ws.send(update); ws.send(msg);
} catch { } catch {
// Connection error — will be cleaned up on close // Connection error — will be cleaned up on close
} }
@@ -108,6 +132,7 @@ export function registerYjsWebSocket(fastify: FastifyInstance, authService: Auth
} }
const doc = await getOrCreateDoc(docName); const doc = await getOrCreateDoc(docName);
const awareness = getOrCreateAwareness(docName);
// Track connection // Track connection
if (!connections.has(docName)) { if (!connections.has(docName)) {
@@ -115,18 +140,50 @@ export function registerYjsWebSocket(fastify: FastifyInstance, authService: Auth
} }
connections.get(docName)!.add(socket); connections.get(docName)!.add(socket);
// Send current document state to new client // Send current document state to new client (with MSG_DOC_UPDATE prefix)
const stateUpdate = Y.encodeStateAsUpdate(doc); const stateUpdate = Y.encodeStateAsUpdate(doc);
if (stateUpdate.length > 0 && socket.readyState === 1) { if (stateUpdate.length > 0 && socket.readyState === 1) {
socket.send(stateUpdate); const msg = new Uint8Array(stateUpdate.length + 1);
msg[0] = MSG_DOC_UPDATE;
msg.set(stateUpdate, 1);
socket.send(msg);
} }
// Listen for updates from this client // Send current awareness states to new client
if (awareness && awareness.getStates().size > 0 && socket.readyState === 1) {
const awarenessUpdate = encodeAwarenessUpdate(awareness, Array.from(awareness.getStates().keys()));
if (awarenessUpdate.length > 0) {
const msg = new Uint8Array(awarenessUpdate.length + 1);
msg[0] = MSG_AWARENESS;
msg.set(awarenessUpdate, 1);
socket.send(msg);
}
}
// Listen for messages from this client
socket.on('message', (data: Buffer) => { socket.on('message', (data: Buffer) => {
try { try {
const update = new Uint8Array(data); const raw = new Uint8Array(data);
Y.applyUpdate(doc, update); if (raw.length === 0) return;
broadcastUpdate(docName, update, socket);
const msgType = raw[0];
const payload = raw.slice(1);
if (msgType === MSG_AWARENESS) {
// Awareness update — apply to server awareness and broadcast
if (awareness) {
applyAwarenessUpdate(awareness, payload, socket);
broadcastUpdate(docName, payload, MSG_AWARENESS, socket);
}
} else if (msgType === MSG_DOC_UPDATE) {
// Y.Doc update — apply to server doc and broadcast
Y.applyUpdate(doc, payload);
broadcastUpdate(docName, payload, MSG_DOC_UPDATE, socket);
} else {
// Legacy: treat entire message as a Y.Doc update (no prefix)
Y.applyUpdate(doc, raw);
broadcastUpdate(docName, raw, MSG_DOC_UPDATE, socket);
}
} catch (err) { } catch (err) {
console.warn(`Invalid update from client for ${docName}:`, err); console.warn(`Invalid update from client for ${docName}:`, err);
} }
@@ -137,6 +194,26 @@ export function registerYjsWebSocket(fastify: FastifyInstance, authService: Auth
const conns = connections.get(docName); const conns = connections.get(docName);
if (conns) { if (conns) {
conns.delete(socket); conns.delete(socket);
// Remove this client's awareness state
if (awareness) {
// Find and remove the client's awareness state based on socket origin
const statesToRemove: number[] = [];
for (const [clientID, meta] of awareness.meta) {
if (meta === socket) {
statesToRemove.push(clientID);
}
}
if (statesToRemove.length > 0) {
removeAwarenessStates(awareness, statesToRemove, socket);
// Broadcast removal to remaining clients
const removalUpdate = encodeAwarenessUpdate(awareness, statesToRemove);
if (removalUpdate.length > 0) {
broadcastUpdate(docName, removalUpdate, MSG_AWARENESS, socket);
}
}
}
if (conns.size === 0) { if (conns.size === 0) {
connections.delete(docName); connections.delete(docName);
// Keep doc in memory for reconnection (persistence handles durability) // Keep doc in memory for reconnection (persistence handles durability)
+24 -8
View File
@@ -111,6 +111,7 @@ const CADEditor: React.FC<CADEditorProps> = ({ projectId, token, onNavigateBack
// Right sidebar // Right sidebar
const [activeRightPanel, setActiveRightPanel] = useState<RightPanel>('tool'); const [activeRightPanel, setActiveRightPanel] = useState<RightPanel>('tool');
const [selectedElement, setSelectedElement] = useState<CADElement | null>(null); const [selectedElement, setSelectedElement] = useState<CADElement | null>(null);
const [selectedElementIds, setSelectedElementIds] = useState<string[]>([]);
const [layers, setLayers] = useState<CADLayer[]>(initialLayers); const [layers, setLayers] = useState<CADLayer[]>(initialLayers);
const [activeLayerId, setActiveLayerId] = useState<string>('layer-0'); const [activeLayerId, setActiveLayerId] = useState<string>('layer-0');
const [blocks, setBlocks] = useState<BlockDefinition[]>(initialBlocks); const [blocks, setBlocks] = useState<BlockDefinition[]>(initialBlocks);
@@ -191,6 +192,8 @@ const CADEditor: React.FC<CADEditorProps> = ({ projectId, token, onNavigateBack
bgConfigRef.current = bgConfig; bgConfigRef.current = bgConfig;
const activeLayerIdRef = useRef(activeLayerId); const activeLayerIdRef = useRef(activeLayerId);
activeLayerIdRef.current = activeLayerId; activeLayerIdRef.current = activeLayerId;
const selectedElementIdsRef = useRef(selectedElementIds);
selectedElementIdsRef.current = selectedElementIds;
// Clipboard (for copy/paste) // Clipboard (for copy/paste)
const clipboardRef = React.useRef<CADElement[] | null>(null); const clipboardRef = React.useRef<CADElement[] | null>(null);
@@ -642,11 +645,16 @@ const CADEditor: React.FC<CADEditorProps> = ({ projectId, token, onNavigateBack
}, [blocks, activeLayerId, collab, drawingId, token]); }, [blocks, activeLayerId, collab, drawingId, token]);
const handleSelectionChange = useCallback((selectedIds: string[]) => { const handleSelectionChange = useCallback((selectedIds: string[]) => {
setSelectedElementIds(selectedIds);
if (selectedIds.length === 1) { if (selectedIds.length === 1) {
const el = elements.find(e => e.id === selectedIds[0]); const el = elements.find(e => e.id === selectedIds[0]);
setSelectedElement(el ?? null); setSelectedElement(el ?? null);
} else { } else if (selectedIds.length === 0) {
setSelectedElement(null); setSelectedElement(null);
} else {
// Multiple elements selected — keep the first one as a representative for the properties panel
const el = elements.find(e => e.id === selectedIds[0]);
setSelectedElement(el ?? null);
} }
}, [elements]); }, [elements]);
@@ -776,7 +784,8 @@ const CADEditor: React.FC<CADEditorProps> = ({ projectId, token, onNavigateBack
if (action === 'undo') { handleUndo(); return; } if (action === 'undo') { handleUndo(); return; }
if (action === 'redo') { handleRedo(); return; } if (action === 'redo') { handleRedo(); return; }
if (action === 'copy') { if (action === 'copy') {
const selected = selectedElement ? [selectedElement] : []; const ids = selectedElementIdsRef.current;
const selected = ids.length > 0 ? elements.filter(e => ids.includes(e.id)) : [];
const clip = selected.length > 0 ? selected : elements.slice(0, 1); const clip = selected.length > 0 ? selected : elements.slice(0, 1);
clipboardRef.current = clip; clipboardRef.current = clip;
setCommandHistory((prev) => [...prev, { prefix: '·', text: `${clip.length} Element(e) kopiert`, type: 'info' }]); setCommandHistory((prev) => [...prev, { prefix: '·', text: `${clip.length} Element(e) kopiert`, type: 'info' }]);
@@ -1136,10 +1145,17 @@ const CADEditor: React.FC<CADEditorProps> = ({ projectId, token, onNavigateBack
// Group command: create group from currently selected elements // Group command: create group from currently selected elements
if (upper === 'GROUP' || upper === 'GRP') { if (upper === 'GROUP' || upper === 'GRP') {
const gm = groupManagerRef.current; const gm = groupManagerRef.current;
// For now, group all elements (selection state is in InteractionEngine, not accessible here) const ids = selectedElementIdsRef.current;
// In a full implementation, we'd need selected IDs from the interaction engine if (ids.length < 2) {
setCommandHistory((prev) => [...prev, { prefix: '·', text: 'Gruppe erstellt (Auswahl im Canvas erforderlich)', type: 'info' }]); setCommandHistory((prev) => [...prev, { prefix: '·', text: 'Gruppe: Mindestens 2 Elemente auswählen', type: 'info' }]);
setGroups(gm.getGroups()); return;
}
const group = gm.createGroup(ids);
const newGroups = gm.getGroups();
setGroups(newGroups);
// Sync group to Yjs CRDT for real-time collaboration
collab.setGroup(group);
setCommandHistory((prev) => [...prev, { prefix: '·', text: `Gruppe erstellt: ${group.name} (${ids.length} Elemente)`, type: 'info' }]);
return; return;
} }
@@ -1194,7 +1210,7 @@ const CADEditor: React.FC<CADEditorProps> = ({ projectId, token, onNavigateBack
setCommandHistory((prev) => [...prev, { prefix: '·', text: `Unbekannter Befehl: ${cmd}`, type: 'info' }]); setCommandHistory((prev) => [...prev, { prefix: '·', text: `Unbekannter Befehl: ${cmd}`, type: 'info' }]);
} }
} }
}, [handleUndo, handleRedo, handleRibbonAction]); }, [handleUndo, handleRedo, handleRibbonAction, collab]);
const handleKISend = useCallback(async (text: string) => { const handleKISend = useCallback(async (text: string) => {
const userMsg: KIMessage = { id: `ki-${Date.now()}`, role: 'user', content: text }; const userMsg: KIMessage = { id: `ki-${Date.now()}`, role: 'user', content: text };
@@ -1249,7 +1265,7 @@ const CADEditor: React.FC<CADEditorProps> = ({ projectId, token, onNavigateBack
handleKISend(suggestion.label); handleKISend(suggestion.label);
}, [handleKISend]); }, [handleKISend]);
const activeLayerName = layers.find((l) => l.id === 'layer-0')?.name ?? '—'; const activeLayerName = layers.find((l) => l.id === activeLayerId)?.name ?? '—';
// ─── Render ───────────────────────────────────────────── // ─── Render ─────────────────────────────────────────────
return ( return (
+125 -33
View File
@@ -1,9 +1,11 @@
/** /**
* AwarenessManager Tracks user presence, cursor position, and selection. * AwarenessManager Tracks user presence, cursor position, and selection.
* Uses a Y.Map inside the shared doc so awareness state syncs via the same * Uses the Yjs awareness protocol (y-protocols/awareness) for ephemeral
* raw-update WebSocket protocol as the rest of the document. * state that should NOT be persisted in the shared Y.Doc.
*
* Reference: https://docs.yjs.dev/ecosystem/awareness
*/ */
import * as Y from 'yjs'; import { Awareness, encodeAwarenessUpdate, applyAwarenessUpdate } from 'y-protocols/awareness';
import type { YjsDocument } from './YjsDocument'; import type { YjsDocument } from './YjsDocument';
export interface UserCursor { export interface UserCursor {
@@ -20,13 +22,28 @@ export interface UserSelection {
elementIds: string[]; elementIds: string[];
} }
export interface AwarenessState { /**
cursors: Y.Map<UserCursor>; * Payload stored under the 'cursor' field of each client's awareness state.
selections: Y.Map<UserSelection>; */
interface CursorAwarenessPayload {
userId: string;
userName: string;
color: string;
x: number;
y: number;
visible: boolean;
}
/**
* Payload stored under the 'selection' field of each client's awareness state.
*/
interface SelectionAwarenessPayload {
userId: string;
elementIds: string[];
} }
export class AwarenessManager { export class AwarenessManager {
readonly state: AwarenessState; readonly awareness: Awareness;
private yjsDoc: YjsDocument; private yjsDoc: YjsDocument;
private userId: string; private userId: string;
private listeners: Set<() => void> = new Set(); private listeners: Set<() => void> = new Set();
@@ -34,85 +51,160 @@ export class AwarenessManager {
constructor(yjsDoc: YjsDocument, userId: string) { constructor(yjsDoc: YjsDocument, userId: string) {
this.yjsDoc = yjsDoc; this.yjsDoc = yjsDoc;
this.userId = userId; this.userId = userId;
this.state = { this.awareness = new Awareness(yjsDoc.doc);
cursors: this.yjsDoc.doc.getMap<UserCursor>('awareness-cursors'),
selections: this.yjsDoc.doc.getMap<UserSelection>('awareness-selections'),
};
} }
/** Set the local user's cursor position */ /** Set the local user's cursor position */
setCursor(x: number, y: number, userName: string, color: string): void { setCursor(x: number, y: number, userName: string, color: string): void {
this.state.cursors.set(this.userId, { const payload: CursorAwarenessPayload = {
userId: this.userId, userId: this.userId,
userName, userName,
color, color,
x, x,
y, y,
visible: true, visible: true,
}); };
this.awareness.setLocalStateField('cursor', payload);
} }
/** Hide the local user's cursor */ /** Hide the local user's cursor */
hideCursor(): void { hideCursor(): void {
const cur = this.state.cursors.get(this.userId); const localState = this.awareness.getLocalState();
if (cur) { if (localState && localState.cursor) {
this.state.cursors.set(this.userId, { ...cur, visible: false }); const cursor = localState.cursor as CursorAwarenessPayload;
this.awareness.setLocalStateField('cursor', {
...cursor,
visible: false,
} as CursorAwarenessPayload);
} }
} }
/** Set the local user's selection */ /** Set the local user's selection */
setSelection(elementIds: string[]): void { setSelection(elementIds: string[]): void {
this.state.selections.set(this.userId, { const payload: SelectionAwarenessPayload = {
userId: this.userId, userId: this.userId,
elementIds, elementIds,
}); };
this.awareness.setLocalStateField('selection', payload);
} }
/** Clear the local user's selection */ /** Clear the local user's selection */
clearSelection(): void { clearSelection(): void {
this.state.selections.delete(this.userId); this.awareness.setLocalStateField('selection', null);
} }
/** Remove the local user from awareness (on disconnect) */ /** Remove the local user from awareness (on disconnect) */
removeSelf(): void { removeSelf(): void {
this.state.cursors.delete(this.userId); this.awareness.setLocalState(null);
this.state.selections.delete(this.userId);
} }
/** Get all active cursors */ /**
* Encode an awareness update for the local client.
* Used by WebSocketProvider to send awareness state over the wire.
*/
encodeLocalState(): Uint8Array | null {
const localState = this.awareness.getLocalState();
if (!localState) return null;
return encodeAwarenessUpdate(this.awareness, [this.awareness.clientID]);
}
/**
* Apply a remote awareness update received from the server.
* Used by WebSocketProvider to process incoming awareness messages.
*/
applyRemoteUpdate(update: Uint8Array, origin: unknown = 'remote'): void {
applyAwarenessUpdate(this.awareness, update, origin);
this.notifyListeners();
}
/** Get all active cursors from all connected clients */
getAllCursors(): UserCursor[] { getAllCursors(): UserCursor[] {
return Array.from(this.state.cursors.values()).filter((c) => c.visible); const cursors: UserCursor[] = [];
for (const [, state] of this.awareness.getStates()) {
if (state && state.cursor) {
const c = state.cursor as CursorAwarenessPayload;
if (c.visible) {
cursors.push({
userId: c.userId,
userName: c.userName,
color: c.color,
x: c.x,
y: c.y,
visible: c.visible,
});
}
}
}
return cursors;
} }
/** Get all selections */ /** Get all selections from all connected clients */
getAllSelections(): UserSelection[] { getAllSelections(): UserSelection[] {
return Array.from(this.state.selections.values()); const selections: UserSelection[] = [];
for (const [, state] of this.awareness.getStates()) {
if (state && state.selection) {
const s = state.selection as SelectionAwarenessPayload;
selections.push({
userId: s.userId,
elementIds: s.elementIds,
});
}
}
return selections;
} }
/** Get a specific user's cursor */ /** Get a specific user's cursor */
getCursor(userId: string): UserCursor | undefined { getCursor(userId: string): UserCursor | undefined {
return this.state.cursors.get(userId); for (const [, state] of this.awareness.getStates()) {
if (state && state.cursor) {
const c = state.cursor as CursorAwarenessPayload;
if (c.userId === userId) {
return {
userId: c.userId,
userName: c.userName,
color: c.color,
x: c.x,
y: c.y,
visible: c.visible,
};
}
}
}
return undefined;
} }
/** Get a specific user's selection */ /** Get a specific user's selection */
getSelection(userId: string): UserSelection | undefined { getSelection(userId: string): UserSelection | undefined {
return this.state.selections.get(userId); for (const [, state] of this.awareness.getStates()) {
if (state && state.selection) {
const s = state.selection as SelectionAwarenessPayload;
if (s.userId === userId) {
return {
userId: s.userId,
elementIds: s.elementIds,
};
}
}
}
return undefined;
} }
/** Register a callback for awareness changes */ /** Register a callback for awareness changes */
onChange(callback: () => void): () => void { onChange(callback: () => void): () => void {
this.listeners.add(callback); this.listeners.add(callback);
const cursorObserver = () => this.notifyListeners(); const observer = () => this.notifyListeners();
const selectionObserver = () => this.notifyListeners(); this.awareness.on('change', observer);
this.state.cursors.observe(cursorObserver);
this.state.selections.observe(selectionObserver);
return () => { return () => {
this.listeners.delete(callback); this.listeners.delete(callback);
this.state.cursors.unobserve(cursorObserver); this.awareness.off('change', observer);
this.state.selections.unobserve(selectionObserver);
}; };
} }
/** Destroy the awareness instance */
destroy(): void {
this.awareness.destroy();
}
private notifyListeners(): void { private notifyListeners(): void {
for (const cb of this.listeners) { for (const cb of this.listeners) {
cb(); cb();
+88 -8
View File
@@ -1,9 +1,14 @@
/** /**
* WebSocketProvider Custom WebSocket provider matching the backend raw-update protocol. * WebSocketProvider Custom WebSocket provider matching the backend raw-update protocol.
* Backend sends Y.encodeStateAsUpdate on connect and applies raw updates on message. * Backend sends Y.encodeStateAsUpdate on connect and applies raw updates on message.
*
* Issue #20: On connect, the client must NOT send its local state immediately.
* Instead, the server sends its state first (Y.encodeStateAsUpdate). The client
* applies that update to sync, then sends only incremental local updates.
*/ */
import * as Y from 'yjs'; import * as Y from 'yjs';
import type { YjsDocument } from './YjsDocument'; import type { YjsDocument } from './YjsDocument';
import type { AwarenessManager } from './AwarenessManager';
export type ConnectionStatus = 'disconnected' | 'connecting' | 'connected' | 'error'; export type ConnectionStatus = 'disconnected' | 'connecting' | 'connected' | 'error';
@@ -11,15 +16,22 @@ export interface WebSocketProviderOptions {
url: string; url: string;
docName: string; docName: string;
yjsDoc: YjsDocument; yjsDoc: YjsDocument;
awareness?: AwarenessManager;
onStatusChange?: (status: ConnectionStatus) => void; onStatusChange?: (status: ConnectionStatus) => void;
onSync?: () => void; onSync?: () => void;
} }
// Message type prefixes for multiplexing Y.Doc updates and awareness updates
// Using a simple byte-prefix scheme: 0 = Y.Doc update, 1 = awareness update
const MSG_DOC_UPDATE = 0;
const MSG_AWARENESS = 1;
export class WebSocketProvider { export class WebSocketProvider {
private ws: WebSocket | null = null; private ws: WebSocket | null = null;
private url: string; private url: string;
private docName: string; private docName: string;
private yjsDoc: YjsDocument; private yjsDoc: YjsDocument;
private awareness?: AwarenessManager;
private status: ConnectionStatus = 'disconnected'; private status: ConnectionStatus = 'disconnected';
private onStatusChange?: (status: ConnectionStatus) => void; private onStatusChange?: (status: ConnectionStatus) => void;
private onSync?: () => void; private onSync?: () => void;
@@ -28,11 +40,13 @@ export class WebSocketProvider {
private maxReconnectDelay = 30000; private maxReconnectDelay = 30000;
private shouldReconnect = true; private shouldReconnect = true;
private authToken: string | undefined; private authToken: string | undefined;
private synced = false;
constructor(opts: WebSocketProviderOptions) { constructor(opts: WebSocketProviderOptions) {
this.url = opts.url; this.url = opts.url;
this.docName = opts.docName; this.docName = opts.docName;
this.yjsDoc = opts.yjsDoc; this.yjsDoc = opts.yjsDoc;
this.awareness = opts.awareness;
this.onStatusChange = opts.onStatusChange; this.onStatusChange = opts.onStatusChange;
this.onSync = opts.onSync; this.onSync = opts.onSync;
} }
@@ -44,6 +58,7 @@ export class WebSocketProvider {
this.authToken = token; this.authToken = token;
this.shouldReconnect = true; this.shouldReconnect = true;
this.synced = false;
this.setStatus('connecting'); this.setStatus('connecting');
const tokenParam = token ? `?token=${encodeURIComponent(token)}` : ''; const tokenParam = token ? `?token=${encodeURIComponent(token)}` : '';
@@ -61,20 +76,50 @@ export class WebSocketProvider {
this.ws.onopen = () => { this.ws.onopen = () => {
this.setStatus('connected'); this.setStatus('connected');
this.reconnectDelay = 1000; this.reconnectDelay = 1000;
// Send local state to server for initial sync // Issue #20: Do NOT send local state on connect.
const stateUpdate = Y.encodeStateAsUpdate(this.yjsDoc.doc); // The server sends its state (Y.encodeStateAsUpdate) to new clients.
if (stateUpdate.length > 0) { // We wait for that server state, apply it, then mark as synced.
this.ws?.send(stateUpdate); // Only after sync do we forward incremental local updates.
//
// If we have awareness, send our local awareness state after connect.
if (this.awareness) {
const awarenessUpdate = this.awareness.encodeLocalState();
if (awarenessUpdate && awarenessUpdate.length > 0) {
this.sendAwarenessUpdate(awarenessUpdate);
}
} }
this.onSync?.();
}; };
this.ws.onmessage = (event: MessageEvent) => { this.ws.onmessage = (event: MessageEvent) => {
try { try {
const data = new Uint8Array(event.data as ArrayBuffer); const data = new Uint8Array(event.data as ArrayBuffer);
this.yjsDoc.applyUpdate(data); if (data.length === 0) return;
// Check message type prefix (first byte)
const msgType = data[0];
const payload = data.slice(1);
if (msgType === MSG_AWARENESS) {
// Awareness update — apply to awareness manager
if (this.awareness) {
this.awareness.applyRemoteUpdate(payload, 'remote');
}
} else {
// Y.Doc update — apply to Y.Doc
// Support both prefixed and non-prefixed messages for backward compatibility
// If the first byte looks like a valid Y.Doc update, use the full data;
// otherwise treat the whole message as the Y.Doc update (legacy mode).
const updateData = msgType === MSG_DOC_UPDATE ? payload : data;
this.yjsDoc.applyUpdate(updateData);
// Mark as synced after receiving the first server state
if (!this.synced) {
this.synced = true;
this.onSync?.();
}
}
} catch (err) { } catch (err) {
console.warn('[WebSocketProvider] Failed to apply update:', err); console.warn('[WebSocketProvider] Failed to process message:', err);
} }
}; };
@@ -84,6 +129,7 @@ export class WebSocketProvider {
this.ws.onclose = () => { this.ws.onclose = () => {
this.ws = null; this.ws = null;
this.synced = false;
this.setStatus('disconnected'); this.setStatus('disconnected');
if (this.shouldReconnect) { if (this.shouldReconnect) {
this.scheduleReconnect(); this.scheduleReconnect();
@@ -93,8 +139,22 @@ export class WebSocketProvider {
/** Send a local Y.Doc update to the server */ /** Send a local Y.Doc update to the server */
sendUpdate(update: Uint8Array): void { sendUpdate(update: Uint8Array): void {
if (this.ws && this.ws.readyState === WebSocket.OPEN && this.synced) {
// Prefix with MSG_DOC_UPDATE byte
const msg = new Uint8Array(update.length + 1);
msg[0] = MSG_DOC_UPDATE;
msg.set(update, 1);
this.ws.send(msg);
}
}
/** Send a local awareness update to the server */
sendAwarenessUpdate(update: Uint8Array): void {
if (this.ws && this.ws.readyState === WebSocket.OPEN) { if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(update); const msg = new Uint8Array(update.length + 1);
msg[0] = MSG_AWARENESS;
msg.set(update, 1);
this.ws.send(msg);
} }
} }
@@ -112,6 +172,25 @@ export class WebSocketProvider {
}; };
} }
/** Listen for local awareness changes and forward to server */
bindAwarenessUpdates(): () => void {
if (!this.awareness) return () => {};
const handler = () => {
const update = this.awareness!.encodeLocalState();
if (update && update.length > 0) {
this.sendAwarenessUpdate(update);
}
};
this.awareness.awareness.on('update', handler);
return () => {
this.awareness!.awareness.off('update', handler);
};
}
isSynced(): boolean {
return this.synced;
}
getStatus(): ConnectionStatus { getStatus(): ConnectionStatus {
return this.status; return this.status;
} }
@@ -126,6 +205,7 @@ export class WebSocketProvider {
this.ws.close(); this.ws.close();
this.ws = null; this.ws = null;
} }
this.synced = false;
this.setStatus('disconnected'); this.setStatus('disconnected');
} }
+1 -1
View File
@@ -3,5 +3,5 @@
*/ */
export { YjsDocument, type YjsDocumentData } from './YjsDocument'; export { YjsDocument, type YjsDocumentData } from './YjsDocument';
export { WebSocketProvider, type ConnectionStatus, type WebSocketProviderOptions } from './WebSocketProvider'; export { WebSocketProvider, type ConnectionStatus, type WebSocketProviderOptions } from './WebSocketProvider';
export { AwarenessManager, type UserCursor, type UserSelection, type AwarenessState } from './AwarenessManager'; export { AwarenessManager, type UserCursor, type UserSelection } from './AwarenessManager';
export { useYjsBinding, type UseYjsBindingOptions, type UseYjsBindingResult } from './useYjsBinding'; export { useYjsBinding, type UseYjsBindingOptions, type UseYjsBindingResult } from './useYjsBinding';
+8 -3
View File
@@ -82,17 +82,18 @@ export function useYjsBinding(opts: UseYjsBindingOptions): UseYjsBindingResult {
const doc = new YjsDocument(); const doc = new YjsDocument();
yjsDocRef.current = doc; yjsDocRef.current = doc;
const awareness = new AwarenessManager(doc, userId);
awarenessRef.current = awareness;
const provider = new WebSocketProvider({ const provider = new WebSocketProvider({
url: wsUrl, url: wsUrl,
docName, docName,
yjsDoc: doc, yjsDoc: doc,
awareness,
onStatusChange: (s) => setStatus(s), onStatusChange: (s) => setStatus(s),
}); });
providerRef.current = provider; providerRef.current = provider;
const awareness = new AwarenessManager(doc, userId);
awarenessRef.current = awareness;
// Sync local state from Y.Doc on changes // Sync local state from Y.Doc on changes
const syncState = () => { const syncState = () => {
setElements(doc.getElements()); setElements(doc.getElements());
@@ -123,6 +124,8 @@ export function useYjsBinding(opts: UseYjsBindingOptions): UseYjsBindingResult {
// Forward local updates to server // Forward local updates to server
const unbindLocal = provider.bindLocalUpdates(); const unbindLocal = provider.bindLocalUpdates();
// Forward local awareness updates to server
const unbindAwarenessUpdates = provider.bindAwarenessUpdates();
// Connect // Connect
provider.connect(token); provider.connect(token);
@@ -135,7 +138,9 @@ export function useYjsBinding(opts: UseYjsBindingOptions): UseYjsBindingResult {
unbindDoc(); unbindDoc();
unbindAwareness(); unbindAwareness();
unbindLocal(); unbindLocal();
unbindAwarenessUpdates();
awareness.removeSelf(); awareness.removeSelf();
awareness.destroy();
provider.disconnect(); provider.disconnect();
doc.destroy(); doc.destroy();
yjsDocRef.current = null; yjsDocRef.current = null;