refac

backend/open_webui/socket/main.py

@@ -950,8 +950,6 @@ async def ydoc_document_join(sid, data):
         # Join Socket.IO room
         await sio.enter_room(sid, f'doc_{document_id}')
 
-        active_session_ids = get_session_ids_from_room(f'doc_{document_id}')
-
         # Get the Yjs document state
         ydoc = Y.Doc()
         updates = await YDOC_MANAGER.get_updates(document_id)
@@ -966,7 +964,6 @@ async def ydoc_document_join(sid, data):
                 'document_id': document_id,
                 'state': list(state_update),  # Convert bytes to list for JSON
                 'content': note.data.get('content') if document_id.startswith('note:') and note.data else None,
-                'sessions': active_session_ids,
             },
             room=sid,
         )
@@ -1102,6 +1099,16 @@ async def yjs_document_update(sid, data):
 
         update = data.get('update')  # List of bytes from frontend
 
+        # The document lock makes initialization first-write-wins across workers.
+        # Return the stored state so losing clients never apply their own seed.
+        if data.get('initialize'):
+            updates = await YDOC_MANAGER.get_updates(document_id)
+            if updates:
+                ydoc = Y.Doc()
+                for stored_update in updates:
+                    ydoc.apply_update(bytes(stored_update))
+                return list(ydoc.get_update())
+
         if update:
             user_id = data.get('user_id', sid)
 
@@ -1125,6 +1132,9 @@ async def yjs_document_update(sid, data):
             else:
                 log.warning(f'Update for document {document_id} is invalid or over the size limit. Rejecting update.')
 
+            if data.get('initialize'):
+                return update if stored else None
+
         async def debounced_save():
             await asyncio.sleep(0.5)
             async with YDOC_MANAGER.lock(document_id):

src/lib/components/common/RichTextInput/Collaboration.ts

@@ -11,7 +11,7 @@ import {
 } from 'y-prosemirror';
 import type { Socket } from 'socket.io-client';
 import type { SessionUser } from '$lib/stores';
-import { Editor, Extension } from '@tiptap/core';
+import { createNodeFromContent, Editor, Extension } from '@tiptap/core';
 import { keymap } from 'prosemirror-keymap';
 import { Plugin } from 'prosemirror-state';
 import { tick } from 'svelte';
@@ -68,6 +68,15 @@ export class SocketIOCollaborationProvider {
 				const plugins = [
 					new Plugin({
 						filterTransaction: (tr) => {
+							// Wait for the saved content before accepting local edits.
+							if (
+								!this.synced &&
+								this.isConnected &&
+								tr.docChanged &&
+								!tr.getMeta(ySyncPluginKey)?.isChangeOrigin
+							) {
+								return false;
+							}
 							// Preserve literal URLs received from another editor.
 							if (tr.getMeta(ySyncPluginKey)?.isChangeOrigin) {
 								tr.setMeta('preventAutolink', true);
@@ -104,16 +113,26 @@ export class SocketIOCollaborationProvider {
 		}
 	}
 
-	private applyInitialContent(content = this.initialContent) {
-		if (!this.editor || !content) return;
+	private initializeContent(content = this.initialContent) {
+		if (!this.editor) return;
 
 		if (typeof content === 'string') {
-			this.editor.commands.setContent(content);
-			return;
+			content = createNodeFromContent(content, this.editor.schema, { slice: false }).toJSON();
 		}
 
 		const doc = prosemirrorJSONToYDoc(this.editor.schema, content);
-		Y.applyUpdate(this.doc, Y.encodeStateAsUpdate(doc));
+		const update = Array.from(Y.encodeStateAsUpdate(doc));
+		doc.destroy();
+		this.socket.emit(
+			'ydoc:document:update',
+			{ document_id: this.documentId, update, initialize: true },
+			(state: number[] | null) => {
+				if (!state || !this.editor) return;
+				Y.applyUpdate(this.doc, new Uint8Array(state), 'server');
+				this.synced = true;
+				if (!this.contentSnapshotTimer) this.sendContentSnapshot();
+			}
+		);
 	}
 
 	// Send the merged content; the remote sender had not seen our edits yet.
@@ -181,18 +200,16 @@ export class SocketIOCollaborationProvider {
 								!this.editor.getText().trim() &&
 								this.doc.getXmlFragment('prosemirror').length === 0
 							) {
-								if (
-									(data.content || this.initialContent) &&
-									[...(data.sessions ?? [])].sort()[0] === this.socket.id
-								) {
+								if (data.content || this.initialContent) {
 									// The HTTP snapshot may predate the last socket save.
-									this.applyInitialContent(
+									this.initializeContent(
 										data.content
 											? data.content.json ||
 													data.content.html ||
 													marked.parse(data.content.md ?? '', { async: false })
 											: this.initialContent
 									);
+									return;
 								}
 							} else {
 								// If the editor already has content, we don't need to send an empty state