/** * One live-editing session on a server document: the Yjs document, its * connection to the server (ActionCable, see app/channels/document_channel.rb), * per-person undo and the presence of the others (awareness: who is here, * what they have selected). * * The store stays the single source for the UI. Local edits flow * store → Binding.sync → Yjs → server; the others' edits flow * server → Yjs → readDocument → store.applyExternalState. Before a remote * change is applied, any not-yet-synced local change goes into Yjs first, so * neither side can overwrite the other. */ import * as Y from 'yjs'; import { Awareness, applyAwarenessUpdate, encodeAwarenessUpdate, removeAwarenessStates } from 'y-protocols/awareness'; import { fromBase64, toBase64 } from 'lib0/buffer'; import { createConsumer, type Consumer, type Subscription } from '@rails/actioncable'; import { reactive, ref, watch, type WatchStopHandle } from 'vue'; import { useDocument, setCollabHooks, contentFingerprint, type UndoProvider } from '../store/document'; import type { DocumentState } from '../store/types'; import type { Role } from '../io/api'; import { Binding, LOCAL, REMOTE, allAxes, initialUpdate, isInitialized, readDocument, type Axes } from './ydoc'; import { validateDocument } from '../io/json'; const SYNC_DELAY = 40; // ms: coalesce a burst of store changes into one Yjs transaction const SNAPSHOT_EVERY = 30_000; // JSON snapshot for the server (export, duplicate) const COMPACT_AFTER = 200; // stored updates before they're merged into one let consumer: Consumer | null = null; const cable = () => (consumer ??= createConsumer('/cable')); export interface Peer { clientId: number; name: string; color: string; sheetId: string | null; tableId: string | null; /** selection rectangle in current positions, or null */ rect: { c0: number; r0: number; c1: number; r1: number } | null; } export interface SessionEvents { onRole(role: Role): void; onRevoked(): void; onTitle(title: string): void; } interface InitMsg { type: 'init'; role: Role; title: string; updates: { id: number; data: string }[]; content?: string; } type Msg = | InitMsg | { type: 'update'; id: number; data: string; from: string } | { type: 'awareness'; data: string; from: string } | { type: 'compacted'; removed: number[]; id: number } | { type: 'meta'; title: string } | { type: 'role'; role: Role } | { type: 'revoked' } | { type: 'error'; error: string }; type SelIds = { tableId: string; a: [string, string]; f: [string, string] } | null; export class CollabSession { readonly ydoc = new Y.Doc(); readonly awareness = new Awareness(this.ydoc); readonly peers = reactive([]); readonly online = ref(false); role: Role = 'viewer'; private store = useDocument(); private binding = new Binding(this.ydoc); private undoManager: Y.UndoManager | null = null; private sub: Subscription | null = null; private key = Math.random().toString(36).slice(2) + Date.now().toString(36); private syncTimer: ReturnType | undefined; private snapshotTimer: ReturnType | undefined; private compactTimer: ReturnType | undefined; private queue: Msg[] = []; private queued = false; private knownIds = new Set(); private axes = new Map(); private lastSnapshot = ''; private initialized = false; private stopWatch: WatchStopHandle | null = null; private attached = false; private ready!: { resolve: (d: DocumentState) => void; reject: (e: Error) => void }; readonly canUndo = ref(false); readonly canRedo = ref(false); constructor( private docId: string, private token: string | null, private user: { name: string; color: string }, private events: SessionEvents, ) { this.ydoc.on('update', (update: Uint8Array, origin: unknown) => { if (origin === REMOTE || !this.canEdit) return; this.sub?.perform('update', { data: toBase64(update) }); }); this.awareness.on('update', ({ added, updated, removed }: Record, origin: unknown) => { if (origin !== 'local') return; const mine = [...added, ...updated, ...removed].filter((c) => c === this.ydoc.clientID); if (mine.length) this.sub?.perform('awareness', { data: toBase64(encodeAwarenessUpdate(this.awareness, mine)) }); }); this.awareness.on('change', () => this.updatePeers()); } get canEdit() { return this.role !== 'viewer'; } /** Connect; resolves with the document once the server sent it. */ start(): Promise { const done = new Promise((resolve, reject) => (this.ready = { resolve, reject })); this.sub = cable().subscriptions.create( { channel: 'DocumentChannel', id: this.docId, token: this.token, key: this.key }, { received: (m: Msg) => this.receive(m), connected: () => (this.online.value = true), disconnected: () => (this.online.value = false), rejected: () => this.ready.reject(new Error('Kein Zugriff auf dieses Dokument.')), }, ); return done; } /** Take over the store: sync hooks, per-person undo, own presence. */ attach(): UndoProvider { this.attached = true; this.axes = allAxes(this.ydoc); setCollabHooks({ onChange: () => this.scheduleSync(), onStructure: (tableId, axis, at, delta) => this.binding.structure(tableId, axis, at, delta), }); const root = this.ydoc.getMap('numolus'); this.undoManager = new Y.UndoManager(root, { trackedOrigins: new Set([LOCAL]), captureTimeout: 400 }); const refresh = () => { this.canUndo.value = this.undoManager!.canUndo(); this.canRedo.value = this.undoManager!.canRedo(); }; this.undoManager.on('stack-item-added', refresh); this.undoManager.on('stack-item-popped', refresh); this.undoManager.on('stack-cleared', refresh); this.awareness.setLocalState({ user: this.user, sheetId: null, sel: null }); this.stopWatch = watch( () => [this.store.doc.selection, this.store.doc.activeSheetId] as const, () => this.publishSelection(), { deep: true, immediate: true }, ); this.snapshotTimer = setInterval(() => this.sendSnapshot(), SNAPSHOT_EVERY); this.lastSnapshot = contentFingerprint(this.store.doc); this.rebuild(); // whatever arrived since start() resolved return { undo: () => this.undoRedo('undo'), redo: () => this.undoRedo('redo'), canUndo: this.canUndo, canRedo: this.canRedo, }; } /** Leave: send what's pending, say goodbye to the others, disconnect. */ destroy() { this.flushLocal(); this.sendSnapshot(); if (this.attached) setCollabHooks(null); this.stopWatch?.(); clearInterval(this.snapshotTimer); clearTimeout(this.compactTimer); removeAwarenessStates(this.awareness, [this.ydoc.clientID], 'local'); this.awareness.destroy(); this.undoManager?.destroy(); this.sub?.unsubscribe(); this.sub = null; this.ydoc.destroy(); } // ---- local → Yjs --------------------------------------------------------- private scheduleSync() { if (!this.canEdit) return; clearTimeout(this.syncTimer); this.syncTimer = setTimeout(() => this.flushLocal(), SYNC_DELAY); } /** Put not-yet-synced store changes into Yjs now. */ flushLocal() { if (this.syncTimer === undefined && !this.binding.hasPending) return; clearTimeout(this.syncTimer); this.syncTimer = undefined; if (!this.canEdit || !isInitialized(this.ydoc)) return; this.binding.sync(this.store.doc); this.axes = allAxes(this.ydoc); } private undoRedo(which: 'undo' | 'redo') { this.flushLocal(); if (which === 'undo') this.undoManager?.undo(); else this.undoManager?.redo(); this.rebuild(); } // ---- server → Yjs ---------------------------------------------------------- private receive(m: Msg) { // bursts are applied together, after the current task this.queue.push(m); if (this.queued) return; this.queued = true; queueMicrotask(() => { this.queued = false; const msgs = this.queue.splice(0); this.process(msgs); }); } private process(msgs: Msg[]) { let changed = false; this.flushLocal(); for (const m of msgs) { switch (m.type) { case 'init': this.init(m); changed = this.initialized; break; case 'update': this.knownIds.add(m.id); if (m.from !== this.key) { Y.applyUpdate(this.ydoc, fromBase64(m.data), REMOTE); changed = true; } break; case 'awareness': applyAwarenessUpdate(this.awareness, fromBase64(m.data), 'remote'); break; case 'compacted': for (const id of m.removed) this.knownIds.delete(id); this.knownIds.add(m.id); clearTimeout(this.compactTimer); this.compactTimer = undefined; break; case 'meta': this.events.onTitle(m.title); break; case 'role': this.role = m.role; this.events.onRole(m.role); break; case 'revoked': this.events.onRevoked(); return; } } if (changed && this.attached) this.rebuild(); this.maybeCompact(); } private init(m: InitMsg) { const first = !this.initialized; this.role = m.role; // what the server has, to find out what it's missing from us (reconnect) const server = new Y.Doc(); for (const u of m.updates) { this.knownIds.add(u.id); const bytes = fromBase64(u.data); Y.applyUpdate(server, bytes); Y.applyUpdate(this.ydoc, bytes, REMOTE); } if (!m.updates.length && m.content && !isInitialized(this.ydoc)) { // not yet converted: everyone converts the same way (see initialUpdate) let json: DocumentState; try { json = validateDocument(JSON.parse(m.content)); } catch (err) { this.ready.reject(err as Error); return; } const u = initialUpdate(json); Y.applyUpdate(this.ydoc, u, REMOTE); if (this.canEdit) this.sub?.perform('update', { data: toBase64(u) }); } else if (!first && this.canEdit) { const missing = Y.encodeStateAsUpdate(this.ydoc, Y.encodeStateVector(server)); if (missing.length > 2) this.sub?.perform('update', { data: toBase64(missing) }); } server.destroy(); if (!isInitialized(this.ydoc)) { this.ready.reject(new Error('Das Dokument ist leer.')); return; } this.initialized = true; if (first) { this.events.onTitle(m.title); this.ready.resolve(readDocument(this.ydoc)); } else { this.events.onRole(m.role); } } /** Show the current Yjs state in the store, keeping this person's view. */ private rebuild() { const selIds = this.selectionIds(); this.store.applyExternalState(readDocument(this.ydoc)); this.axes = allAxes(this.ydoc); this.restoreSelection(selIds); this.updatePeers(); } // ---- selection by ids (survives rows inserted by others) -------------------- private idsAt(tableId: string, c: number, r: number): [string, string] | null { const ax = this.axes.get(tableId); return ax && ax.cols[c] !== undefined && ax.rows[r] !== undefined ? [ax.cols[c], ax.rows[r]] : null; } private posOf(tableId: string, ids: [string, string]): { col: number; row: number } | null { const ax = this.axes.get(tableId); if (!ax) return null; const col = ax.cols.indexOf(ids[0]); const row = ax.rows.indexOf(ids[1]); return col < 0 || row < 0 ? null : { col, row }; } private selectionIds(): SelIds { const s = this.store.doc.selection; if (!s) return null; const a = this.idsAt(s.tableId, s.anchor.col, s.anchor.row); const f = this.idsAt(s.tableId, s.focus.col, s.focus.row); return a && f ? { tableId: s.tableId, a, f } : null; } private restoreSelection(ids: SelIds) { const s = this.store.doc.selection; if (!ids || !s || s.tableId !== ids.tableId) return; const anchor = this.posOf(ids.tableId, ids.a); const focus = this.posOf(ids.tableId, ids.f); if (!anchor || !focus) return; // its row/column is gone: stay where it was (clamped) if (anchor.col === s.anchor.col && anchor.row === s.anchor.row && focus.col === s.focus.col && focus.row === s.focus.row) return; this.store.doc.selection = { tableId: ids.tableId, anchor, focus }; } // ---- presence --------------------------------------------------------------- private publishSelection() { const sel = this.selectionIds(); this.awareness.setLocalStateField('sheetId', this.store.doc.activeSheetId); this.awareness.setLocalStateField('sel', sel); } private updatePeers() { const out: Peer[] = []; for (const [clientId, st] of this.awareness.getStates()) { if (clientId === this.ydoc.clientID || !st.user) continue; const sel = st.sel as SelIds; let rect: Peer['rect'] = null; if (sel) { const a = this.posOf(sel.tableId, sel.a); const f = this.posOf(sel.tableId, sel.f); if (a && f) { rect = { c0: Math.min(a.col, f.col), c1: Math.max(a.col, f.col), r0: Math.min(a.row, f.row), r1: Math.max(a.row, f.row), }; } } out.push({ clientId, name: String(st.user.name ?? '?'), color: String(st.user.color ?? '#888'), sheetId: (st.sheetId as string) ?? null, tableId: sel?.tableId ?? null, rect, }); } this.peers.splice(0, this.peers.length, ...out); } // ---- housekeeping ------------------------------------------------------------ /** JSON of the document for the server (export, duplicate, list), when changed. */ private sendSnapshot() { if (!this.canEdit || !this.initialized || !this.attached) return; const fp = contentFingerprint(this.store.doc); if (fp === this.lastSnapshot) return; this.lastSnapshot = fp; this.sub?.perform('snapshot', { content: JSON.stringify(this.store.doc) }); } /** Many stored updates → one; whoever notices first does it. */ private maybeCompact() { if (!this.canEdit || this.compactTimer || this.knownIds.size < COMPACT_AFTER) return; this.compactTimer = setTimeout(() => { this.flushLocal(); this.sub?.perform('compact', { state: toBase64(Y.encodeStateAsUpdate(this.ydoc)), ids: [...this.knownIds], snapshot: JSON.stringify(this.store.doc), }); // if no confirmation arrives, try again later this.compactTimer = setTimeout(() => (this.compactTimer = undefined), 30_000); }, 1000 + Math.random() * 3000); } }