diff --git a/Doku/flaws.md b/Doku/flaws.md index 13b33a8..05855fc 100644 --- a/Doku/flaws.md +++ b/Doku/flaws.md @@ -10,7 +10,7 @@ Diese Datei dokumentiert eine ehrliche und schonungslose technische Analyse alle |---|---|---|---|---| | **FLAW-01** | Workspace Sync | Kein echtes CRDT/OT – Lock-Heartbeat-Modell mit Last-Write-Wins | **Gelöst (v2.1)** | Behoben durch Yjs CRDT-Engine & blockbasierte kollaborative Bearbeitung | | **FLAW-02** | Storage | Dual-Storage Drift (SQLite vs. Dateisystem) | **Obsolet / Gelöst** | Kein Risiko: MindSift nutzt Pure-DB (keine `.md`-Dateien auf Platte) | -| **FLAW-03** | Netzwerk / WS | Proxy-Inkompatibilität bei WebSocket-Upgrades & Polling-Fallback | **Mittel** | Verzögerte Synchronisierung (1–3s Polling) in Firmennetzwerken | +| **FLAW-03** | Netzwerk / WS | Proxy-Inkompatibilität bei WebSocket-Upgrades & Polling-Fallback | **Gelöst (v2.2)** | Behoben durch Dual-Transport SSE-Fallback & globale Singleton-Verbindung | | **FLAW-04** | Datenbank / Suche | Volltextsuche via `LIKE '%query%'` statt SQLite FTS5 | **Mittel** | Performance-Einbrüche bei wachsenden Notizbeständen (>10.000 Blöcke) | | **FLAW-05** | Frontend DOM | Fehlende Virtualisierung im Outliner-Baum | **Mittel** | Frame-Drops und Lags bei Dokumenten mit >500 Blöcken auf Mobilgeräten | | **FLAW-06** | SQLite Concurrency | Single-Writer-Bottleneck bei hohem parallelem Schreibaufkommen | **Mittel** | `SQLITE_BUSY` Fehler bei vielen gleichzeitigen Team-Nutzern | diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index a0bfd16..0cc26a1 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,4 +1,4 @@ -import React, { useEffect, useRef, useState } from 'react'; +import React, { useCallback, useEffect, useRef, useState } from 'react'; import { Layers, CheckSquare, @@ -414,7 +414,7 @@ export const App: React.FC = () => { setActiveThemeState(themeId); }; - const refreshAll = async () => { + const refreshAll = useCallback(async () => { try { const [meRes, deckRes, listsRes] = await Promise.all([ api.getMe(), @@ -428,7 +428,7 @@ export const App: React.FC = () => { } catch { // If token expired or invalid, api.ts dispatches mindsift:unauthorized } - }; + }, []); const checkBootAndAuth = async () => { setBootLoading(true); diff --git a/frontend/src/features/lists/SharedListsView.tsx b/frontend/src/features/lists/SharedListsView.tsx index 8fc62f0..03b737d 100644 --- a/frontend/src/features/lists/SharedListsView.tsx +++ b/frontend/src/features/lists/SharedListsView.tsx @@ -105,7 +105,6 @@ export const SharedListsView: React.FC = ({ onRefreshGloba const [dragOverItemTarget, setDragOverItemTarget] = useState<{ itemId: string; position: 'before' | 'after' } | null>(null); const [dragOverCategoryTarget, setDragOverCategoryTarget] = useState(null); - const wsRef = useRef(null); const realtimeConnRef = useRef(null); const loadLists = async () => { diff --git a/frontend/src/features/workspace/WorkspaceView.tsx b/frontend/src/features/workspace/WorkspaceView.tsx index 4b46cf6..d0f7760 100644 --- a/frontend/src/features/workspace/WorkspaceView.tsx +++ b/frontend/src/features/workspace/WorkspaceView.tsx @@ -50,7 +50,7 @@ import { Radio, } from 'lucide-react'; import { BlockType, Document, DocumentBlock, DocBlocksPatchEvent, DocLockInfo, DocShare, FolderShare, SharedFolder, TagDetails, TagSummary, Team, User } from '../../lib/types'; -import { api, getToken, triggerHaptic, onWebSocketStatus, isWebSocketConnected, buildWebSocketURL, createRealtimeConnection, RealtimeTransport } from '../../lib/api'; +import { api, getToken, triggerHaptic, onWebSocketStatus, isWebSocketConnected, sendWebSocketMessage, createRealtimeConnection, RealtimeTransport } from '../../lib/api'; import { CodeBlockItem } from './CodeBlockItem'; import { UserAutocomplete } from '../UserAutocomplete'; import { UnifiedShareModal } from '../../components/UnifiedShareModal'; @@ -130,14 +130,15 @@ export const WorkspaceView: React.FC = ({ const [showIconPicker, setShowIconPicker] = useState(false); const [showWikiLinkMenu, setShowWikiLinkMenu] = useState(false); - // Dedicated WebSocket/SSE connection & status for live indicator - const wsRef = useRef(null); + // Realtime connection & status for live indicator const realtimeConnRef = useRef(null); - const wsReconnectTimerRef = useRef(null); - const wsReconnectDelayRef = useRef(2000); const [wsConnected, setWsConnected] = useState(false); const [realtimeTransport, setRealtimeTransport] = useState('none'); const [wsDisconnectReason, setWsDisconnectReason] = useState(null); + const onStatsChangeRef = useRef(onStatsChange); + onStatsChangeRef.current = onStatsChange; + const lastLocalEditTimestampRef = useRef(0); + const isSavingToBackendRef = useRef(false); // Folder state (synchronized with server) const [customFolders, setCustomFolders] = useState([]); @@ -441,25 +442,11 @@ export const WorkspaceView: React.FC = ({ color: getCollaboratorColor(currentUser?.id || ''), lastActive: Date.now(), }; - if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { - realtimeConnRef.current.send({ - event: 'CRDT_AWARENESS', - docId: selectedDoc.id, - payload, - }); - } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { - try { - wsRef.current.send( - JSON.stringify({ - event: 'CRDT_AWARENESS', - docId: selectedDoc.id, - payload, - }) - ); - } catch {} - } else { - api.sendDocumentAwareness(selectedDoc.id, payload).catch(() => {}); - } + sendWebSocketMessage({ + event: 'CRDT_AWARENESS', + docId: selectedDoc.id, + payload, + }); }; const handleBlockFocus = (blockIdOrEvent?: string | React.FocusEvent) => { @@ -501,6 +488,7 @@ export const WorkspaceView: React.FC = ({ const executeSaveToBackend = async (docToSave: Document) => { if (isReadOnly) return; let saveSucceeded = false; + isSavingToBackendRef.current = true; try { const docId = docToSave.id; const dirtyMap = localDirtyBlocksRef.current.get(docId); @@ -582,7 +570,7 @@ export const WorkspaceView: React.FC = ({ } saveSucceeded = true; - onStatsChange(); + onStatsChangeRef.current(); api.getWorkspaceTags().then(setWorkspaceTags).catch(() => {}); if (selectedTag) { api.getTagDetails(selectedTag).then(setTagDetails).catch(() => {}); @@ -590,6 +578,7 @@ export const WorkspaceView: React.FC = ({ } catch (e: any) { console.error('Failed to save document to backend', e); } finally { + isSavingToBackendRef.current = false; if (saveSucceeded) { if (pendingSaveDocRef.current === null || pendingSaveDocRef.current === docToSave) { pendingSaveDocRef.current = null; @@ -626,6 +615,7 @@ export const WorkspaceView: React.FC = ({ const persistDocument = async (updatedDoc: Document, immediate = false) => { if (isReadOnly) return; ensureLock(); + lastLocalEditTimestampRef.current = Date.now(); // Record block-level delta against previous local state and server baseline const docId = updatedDoc.id; @@ -792,15 +782,7 @@ export const WorkspaceView: React.FC = ({ if (prevId && prevId !== selectedDoc?.id) { flushPendingSave(); releaseMyLock(prevId); - if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { - realtimeConnRef.current.send({ event: 'CRDT_LEAVE', docId: prevId }); - } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { - try { - wsRef.current.send(JSON.stringify({ event: 'CRDT_LEAVE', docId: prevId })); - } catch {} - } else { - api.clearDocumentAwareness(prevId).catch(() => {}); - } + sendWebSocketMessage({ event: 'CRDT_LEAVE', docId: prevId }); if (crdtSessionRef.current) { crdtSessionRef.current.destroy(); crdtSessionRef.current = null; @@ -814,39 +796,24 @@ export const WorkspaceView: React.FC = ({ const session = new DocumentCRDTSession(targetDoc.id); session.initFromBlocks(targetDoc.blocks || []); session.onLocalUpdate = (b64Update) => { - if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { - realtimeConnRef.current.send({ - event: 'CRDT_UPDATE', - docId: targetDoc.id, - payload: { update: b64Update }, - }); - } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { - try { - wsRef.current.send( - JSON.stringify({ - event: 'CRDT_UPDATE', - docId: targetDoc.id, - payload: { update: b64Update }, - }) - ); - } catch {} - } else { - api.saveDocumentCRDT(targetDoc.id, { update: b64Update }).catch(() => {}); - } + sendWebSocketMessage({ + event: 'CRDT_UPDATE', + docId: targetDoc.id, + payload: { update: b64Update }, + }); }; session.onRemoteChange = (crdtBlocks) => { - setSelectedDoc((prev) => (prev && prev.id === targetDoc.id ? { ...prev, blocks: crdtBlocks } : prev)); - setDocs((prev) => prev.map((d) => (d.id === targetDoc.id ? { ...d, blocks: crdtBlocks } : d))); + const recentlyEdited = Date.now() - lastLocalEditTimestampRef.current < 2500; + const hasPendingSave = pendingSaveDocRef.current !== null || isSavingToBackendRef.current; + const hasDirtyBlocks = (localDirtyBlocksRef.current.get(targetDoc.id)?.size || 0) > 0; + if (!recentlyEdited && !hasPendingSave && !hasDirtyBlocks) { + setSelectedDoc((prev) => (prev && prev.id === targetDoc.id ? { ...prev, blocks: crdtBlocks } : prev)); + setDocs((prev) => prev.map((d) => (d.id === targetDoc.id ? { ...d, blocks: crdtBlocks } : d))); + } }; crdtSessionRef.current = session; - if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { - realtimeConnRef.current.send({ event: 'CRDT_JOIN', docId: targetDoc.id }); - } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { - try { - wsRef.current.send(JSON.stringify({ event: 'CRDT_JOIN', docId: targetDoc.id })); - } catch {} - } + sendWebSocketMessage({ event: 'CRDT_JOIN', docId: targetDoc.id }); api .getDocumentCRDT(targetDoc.id) @@ -859,7 +826,10 @@ export const WorkspaceView: React.FC = ({ crdtSessionRef.current.applyRemoteUpdates(res.updates); } const blocks = crdtSessionRef.current.getBlocks(); - if (blocks.length > 0) { + const recentlyEdited = Date.now() - lastLocalEditTimestampRef.current < 2500; + const hasPendingSave = pendingSaveDocRef.current !== null || isSavingToBackendRef.current; + const hasDirtyBlocks = (localDirtyBlocksRef.current.get(targetDoc.id)?.size || 0) > 0; + if (blocks.length > 0 && !recentlyEdited && !hasPendingSave && !hasDirtyBlocks) { setSelectedDoc((prev) => (prev && prev.id === targetDoc.id ? { ...prev, blocks } : prev)); setDocs((prev) => prev.map((d) => (d.id === targetDoc.id ? { ...d, blocks } : d))); } @@ -884,15 +854,7 @@ export const WorkspaceView: React.FC = ({ return () => { flushPendingSave(); if (currentDocIdRef.current) { - if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { - realtimeConnRef.current.send({ event: 'CRDT_LEAVE', docId: currentDocIdRef.current }); - } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { - try { - wsRef.current.send(JSON.stringify({ event: 'CRDT_LEAVE', docId: currentDocIdRef.current })); - } catch {} - } else { - api.clearDocumentAwareness(currentDocIdRef.current).catch(() => {}); - } + sendWebSocketMessage({ event: 'CRDT_LEAVE', docId: currentDocIdRef.current }); if (crdtSessionRef.current) { crdtSessionRef.current.destroy(); crdtSessionRef.current = null; @@ -967,7 +929,10 @@ export const WorkspaceView: React.FC = ({ crdtSessionRef.current.applyRemoteUpdates(payload.updates); } const blocks = crdtSessionRef.current.getBlocks(); - if (blocks.length > 0) { + const recentlyEdited = Date.now() - lastLocalEditTimestampRef.current < 2500; + const hasPendingSave = pendingSaveDocRef.current !== null || isSavingToBackendRef.current; + const hasDirtyBlocks = (localDirtyBlocksRef.current.get(payload.docId)?.size || 0) > 0; + if (blocks.length > 0 && !recentlyEdited && !hasPendingSave && !hasDirtyBlocks) { setSelectedDoc((prev) => (prev && prev.id === payload.docId ? { ...prev, blocks } : prev)); setDocs((prev) => prev.map((d) => (d.id === payload.docId ? { ...d, blocks } : d))); } @@ -978,7 +943,10 @@ export const WorkspaceView: React.FC = ({ if (payload.fromUserId !== currentUserIdRef.current) { crdtSessionRef.current.applyRemoteUpdate(payload.update); const blocks = crdtSessionRef.current.getBlocks(); - if (blocks && blocks.length > 0) { + const recentlyEdited = Date.now() - lastLocalEditTimestampRef.current < 2500; + const hasPendingSave = pendingSaveDocRef.current !== null || isSavingToBackendRef.current; + const hasDirtyBlocks = (localDirtyBlocksRef.current.get(payload.docId)?.size || 0) > 0; + if (blocks && blocks.length > 0 && !recentlyEdited && !hasPendingSave && !hasDirtyBlocks) { setSelectedDoc((prev) => (prev && prev.id === payload.docId ? { ...prev, blocks } : prev)); setDocs((prev) => prev.map((d) => (d.id === payload.docId ? { ...d, blocks } : d))); } @@ -1046,7 +1014,7 @@ export const WorkspaceView: React.FC = ({ if (created && created.id) { recordSyncedDoc(created); setDocs((prev) => (prev.some((d) => d.id === created.id) ? prev : [created, ...prev])); - onStatsChange(); + onStatsChangeRef.current(); } } else if (msg.event === 'DOC_UPDATED') { const updated = msg.payload as Document; @@ -1089,7 +1057,7 @@ export const WorkspaceView: React.FC = ({ return nextDoc; }); } - onStatsChange(); + onStatsChangeRef.current(); } } else if (msg.event === 'DOC_DELETED') { const deletedId = typeof msg.payload === 'string' ? msg.payload : msg.payload?.id; @@ -1098,7 +1066,7 @@ export const WorkspaceView: React.FC = ({ if (selectedDocIdRef.current === deletedId) { setSelectedDoc(null); } - onStatsChange(); + onStatsChangeRef.current(); } } else if (msg.event === 'WORKSPACE_FOLDERS_UPDATED') { api.getFolders().then(setCustomFolders).catch(() => {}); @@ -1177,7 +1145,7 @@ export const WorkspaceView: React.FC = ({ } window.removeEventListener('mindsift:ws', handleGlobalCustomEvent); }; - }, [onStatsChange]); + }, []); // Fallback HTTP polling when WebSocket is disconnected (e.g. corporate proxy / firewall blocking WebSockets) useEffect(() => { @@ -1200,7 +1168,16 @@ export const WorkspaceView: React.FC = ({ } // 2. Poll document content if not actively typing and document has changed on server - if (selectedDocIdRef.current && !isEditorFocusedRef.current && !pendingSaveDocRef.current) { + const recentlyEdited = Date.now() - lastLocalEditTimestampRef.current < 4000; + const hasPendingSave = pendingSaveDocRef.current !== null || isSavingToBackendRef.current; + const hasDirtyBlocks = (localDirtyBlocksRef.current.get(selectedDocIdRef.current || '')?.size || 0) > 0; + if ( + selectedDocIdRef.current && + !isEditorFocusedRef.current && + !hasPendingSave && + !hasDirtyBlocks && + !recentlyEdited + ) { try { const fresh = await api.getDocument(selectedDocIdRef.current); if (fresh && (!selectedDoc || fresh.updatedAt !== selectedDoc.updatedAt)) { @@ -1889,20 +1866,9 @@ export const WorkspaceView: React.FC = ({ : b ); nextBlocks.splice(afterIndex + 1, 0, newBlock); + lastLocalEditTimestampRef.current = Date.now(); persistDocument({ ...selectedDoc, blocks: nextBlocks }); - setLastFocusedBlockId(newBlock.id); - setBlockCursor(newBlock.id, 0); - - // Focus into the new element next tick - setTimeout(() => { - const el = blockInputRefs.current[newBlock.id]; - if (el) { - el.focus(); - if ('setSelectionRange' in el && typeof (el as any).value === 'string') { - (el as any).setSelectionRange(0, 0); - } - } - }, 20); + focusBlock(newBlock.id, 0); }; // Indent block (Tab): increase level by 1 and move all descendant sub-bullets along diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index be47083..765ffc2 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1140,6 +1140,11 @@ export function onRealtimeStatus(callback: (connected: boolean, transport: Realt }; } +let isWsBlockedByProxy = false; +let sseReconnectTimer: any = null; +let sseReconnectDelay = 2000; +const globalSubscribedRooms = new Set(); + function notifyRealtimeStatusChange(connected: boolean, transport: RealtimeTransport) { isRealtimeConnectedState = connected; activeTransport = transport; @@ -1154,20 +1159,32 @@ function notifyRealtimeStatusChange(connected: boolean, transport: RealtimeTrans } export function connectSSE(rooms?: string[]): void { + if (rooms) { + rooms.forEach((r) => globalSubscribedRooms.add(r)); + } + clearTimeout(sseReconnectTimer); + clearTimeout(wsReconnectTimer); + clearTimeout(wsConnectionWatchdog); + if (activeGlobalSSE) { try { activeGlobalSSE.close(); } catch {} activeGlobalSSE = null; } try { - const url = buildSSEURL(undefined, undefined, rooms); + const url = buildSSEURL(undefined, undefined, Array.from(globalSubscribedRooms)); const es = new EventSource(url); activeGlobalSSE = es; es.onopen = () => { console.info('[MindSift Realtime] Server-Sent Events (SSE) verbunden (Proxy-Fallback aktiv).'); - wsReconnectDelay = 2000; + sseReconnectDelay = 2000; notifyRealtimeStatusChange(true, 'sse'); + if (activeSSEId && globalSubscribedRooms.size > 0) { + globalSubscribedRooms.forEach((room) => { + api.joinSSERoom(activeSSEId!, room, 'join').catch(() => {}); + }); + } }; es.onmessage = (event) => { @@ -1175,6 +1192,11 @@ export function connectSSE(rooms?: string[]): void { const msg = JSON.parse(event.data) as WSMessage; if (msg.event === 'SSE_CONNECTED') { activeSSEId = msg.payload?.sseId || null; + if (activeSSEId && globalSubscribedRooms.size > 0) { + globalSubscribedRooms.forEach((room) => { + api.joinSSERoom(activeSSEId!, room, 'join').catch(() => {}); + }); + } } wsListeners.forEach((fn) => { try { @@ -1190,17 +1212,24 @@ export function connectSSE(rooms?: string[]): void { }; es.onerror = (e) => { - console.warn('[MindSift Realtime] SSE getrennt oder Fehler aufgetreten:', e); if (activeTransport === 'sse') { notifyRealtimeStatusChange(false, 'none'); } try { es.close(); } catch {} activeGlobalSSE = null; if (!manualDisconnect && getToken()) { - clearTimeout(wsReconnectTimer); - wsReconnectTimer = setTimeout(() => { - connectWebSocket(); - }, 5000); + clearTimeout(sseReconnectTimer); + const delay = sseReconnectDelay; + sseReconnectDelay = Math.min(30000, Math.round(sseReconnectDelay * 1.5)); + sseReconnectTimer = setTimeout(() => { + if (!manualDisconnect && getToken()) { + if (isWsBlockedByProxy) { + connectSSE(); + } else { + connectWebSocket(); + } + } + }, delay); } }; } catch (err) { @@ -1214,11 +1243,19 @@ export function connectWebSocket(): void { if (!token) return; manualDisconnect = false; + // If proxy blocked WebSocket previously, do not attempt failing WebSocket; stay on SSE + if (isWsBlockedByProxy) { + connectSSE(); + return; + } + if (activeGlobalWS && (activeGlobalWS.readyState === WebSocket.OPEN || activeGlobalWS.readyState === WebSocket.CONNECTING)) { return; } clearTimeout(wsConnectionWatchdog); + clearTimeout(wsReconnectTimer); + clearTimeout(sseReconnectTimer); try { const url = buildWebSocketURL(); @@ -1229,6 +1266,7 @@ export function connectWebSocket(): void { wsConnectionWatchdog = setTimeout(() => { if (ws && ws.readyState !== WebSocket.OPEN) { console.warn('[MindSift Realtime] WebSocket connection timeout (Proxy Upgrade Block). Wechsle automatisch zu SSE...'); + isWsBlockedByProxy = true; try { ws.close(); } catch {} activeGlobalWS = null; connectSSE(); @@ -1243,6 +1281,13 @@ export function connectWebSocket(): void { } wsReconnectDelay = 2000; notifyRealtimeStatusChange(true, 'ws'); + if (globalSubscribedRooms.size > 0) { + globalSubscribedRooms.forEach((room) => { + try { + ws.send(JSON.stringify({ event: 'JOIN_ROOM', room })); + } catch {} + }); + } }; ws.onmessage = (event) => { @@ -1264,11 +1309,11 @@ export function connectWebSocket(): void { ws.onclose = (evt: CloseEvent) => { clearTimeout(wsConnectionWatchdog); activeGlobalWS = null; - console.warn('[MindSift Global WS] Closed with code:', evt?.code, 'reason:', evt?.reason); const isProxyIssue = evt?.code === 1006 || evt?.code === 1002 || evt?.code === 1008; if (isProxyIssue && !manualDisconnect && getToken()) { - console.warn('[MindSift Realtime] Proxy/Firewall Inkompatibilität erkannt (Code ' + evt.code + '). Wechsle zu SSE...'); + console.warn('[MindSift Realtime] Proxy/Firewall Inkompatibilität erkannt (Code ' + evt.code + '). Wechsle dauerhaft zu SSE...'); + isWsBlockedByProxy = true; connectSSE(); return; } @@ -1282,21 +1327,22 @@ export function connectWebSocket(): void { const delay = wsReconnectDelay; wsReconnectDelay = Math.min(30000, Math.round(wsReconnectDelay * 1.5)); wsReconnectTimer = setTimeout(() => { - connectWebSocket(); + if (!manualDisconnect && getToken()) { + connectWebSocket(); + } }, delay); } }; ws.onerror = (e) => { clearTimeout(wsConnectionWatchdog); - console.warn('[MindSift Global WS] Error:', e); try { ws.close(); } catch {} }; } catch (e) { clearTimeout(wsConnectionWatchdog); - console.warn('Failed to connect WebSocket, switching to SSE:', e); + isWsBlockedByProxy = true; connectSSE(); } } @@ -1305,12 +1351,20 @@ export function connectWebSocket(): void { if (typeof window !== 'undefined') { window.addEventListener('online', () => { if (getToken() && !isRealtimeConnected()) { - connectWebSocket(); + if (isWsBlockedByProxy) { + connectSSE(); + } else { + connectWebSocket(); + } } }); document.addEventListener('visibilitychange', () => { if (document.visibilityState === 'visible' && getToken() && !isRealtimeConnected()) { - connectWebSocket(); + if (isWsBlockedByProxy) { + connectSSE(); + } else { + connectWebSocket(); + } } }); } @@ -1319,6 +1373,7 @@ export function disconnectWebSocket(): void { manualDisconnect = true; clearTimeout(wsReconnectTimer); clearTimeout(wsConnectionWatchdog); + clearTimeout(sseReconnectTimer); if (activeGlobalWS) { try { activeGlobalWS.close(); @@ -1344,18 +1399,22 @@ export function sendWebSocketMessage(msg: any): void { try { const parsed = typeof msg === 'string' ? JSON.parse(msg) : msg; if (parsed.event === 'CRDT_UPDATE' && parsed.docId) { - api.saveDocumentCRDT(parsed.docId, { update: parsed.payload?.update }); + api.saveDocumentCRDT(parsed.docId, { update: parsed.payload?.update }).catch(() => {}); } else if (parsed.event === 'CRDT_AWARENESS' && parsed.docId) { - api.sendDocumentAwareness(parsed.docId, parsed.payload); + api.sendDocumentAwareness(parsed.docId, parsed.payload).catch(() => {}); } else if (parsed.event === 'CRDT_LEAVE' && parsed.docId) { - api.clearDocumentAwareness(parsed.docId); - if (activeSSEId) api.joinSSERoom(activeSSEId, `doc:${parsed.docId}`, 'leave'); + api.clearDocumentAwareness(parsed.docId).catch(() => {}); + if (activeSSEId) api.joinSSERoom(activeSSEId, `doc:${parsed.docId}`, 'leave').catch(() => {}); } else if (parsed.event === 'CRDT_JOIN' && parsed.docId) { - if (activeSSEId) api.joinSSERoom(activeSSEId, `doc:${parsed.docId}`, 'join'); + if (activeSSEId) api.joinSSERoom(activeSSEId, `doc:${parsed.docId}`, 'join').catch(() => {}); + } else if (parsed.event === 'JOIN_ROOM' && parsed.room) { + if (activeSSEId) api.joinSSERoom(activeSSEId, parsed.room, 'join').catch(() => {}); + } else if (parsed.event === 'LEAVE_ROOM' && parsed.room) { + if (activeSSEId) api.joinSSERoom(activeSSEId, parsed.room, 'leave').catch(() => {}); } else if (parsed.event === 'LOCK_DOC' && parsed.docId) { - api.lockDocument(parsed.docId); + api.lockDocument(parsed.docId).catch(() => {}); } else if (parsed.event === 'UNLOCK_DOC' && parsed.docId) { - api.unlockDocument(parsed.docId); + api.unlockDocument(parsed.docId).catch(() => {}); } } catch (err) { console.error('Error dispatching fallback REST message for SSE:', err); @@ -1377,196 +1436,79 @@ export function createRealtimeConnection(options: { onStatusChange?: (status: { connected: boolean; transport: RealtimeTransport; reason?: string }) => void; }) { let isClosed = false; - let currentTransport: RealtimeTransport = 'none'; - let ws: WebSocket | null = null; - let es: EventSource | null = null; - let watchdogTimer: any = null; - let reconnectTimer: any = null; - let reconnectDelay = 2000; - let sseId: string | null = null; - const notify = (connected: boolean, transport: RealtimeTransport, reason?: string) => { - currentTransport = transport; - options.onStatusChange?.({ connected, transport, reason }); - }; + const joinedRooms: string[] = []; + if (options.docId) joinedRooms.push(`doc:${options.docId}`); + if (options.listId) joinedRooms.push(options.listId); + if (options.rooms) { + options.rooms.forEach((r) => { + if (!joinedRooms.includes(r)) joinedRooms.push(r); + }); + } - const connectSSE = () => { + joinedRooms.forEach((r) => globalSubscribedRooms.add(r)); + + // Subscribe to status changes + const unsubStatus = onRealtimeStatus((connected, transport) => { if (isClosed) return; - if (es) { - try { es.close(); } catch {} - es = null; - } + options.onStatusChange?.({ + connected, + transport, + reason: connected ? undefined : 'Verbindung getrennt', + }); + }); - try { - const allRooms = [...(options.rooms || [])]; - if (options.docId) allRooms.push(`doc:${options.docId}`); - if (options.listId) allRooms.push(options.listId); - const url = buildSSEURL(options.listId, options.docId, allRooms); - es = new EventSource(url); + // Subscribe to messages + let unsubMsg: (() => void) | null = null; + if (options.onMessage) { + unsubMsg = onWebSocketMessage((msg) => { + if (isClosed) return; + options.onMessage?.(msg); + }); + } - es.onopen = () => { - if (isClosed) return; - reconnectDelay = 2000; - notify(true, 'sse'); - }; - - es.onmessage = (evt) => { - if (isClosed) return; - try { - const msg = JSON.parse(evt.data) as WSMessage; - if (msg.event === 'SSE_CONNECTED') { - sseId = msg.payload?.sseId || null; - } - options.onMessage?.(msg); - } catch (e) { - console.error('SSE JSON error', e); - } - }; - - es.onerror = (e) => { - if (isClosed) return; - notify(false, 'none', 'SSE-Verbindungsabbruch'); - try { es?.close(); } catch {} - es = null; - clearTimeout(reconnectTimer); - reconnectTimer = setTimeout(() => { - if (!isClosed) connectWS(); - }, 5000); - }; - } catch (e: any) { - notify(false, 'none', e?.message || 'SSE Initialisierungsfehler'); - } + // Join rooms on the active connection + const joinAll = () => { + joinedRooms.forEach((r) => { + if (r.startsWith('doc:')) { + sendWebSocketMessage({ event: 'CRDT_JOIN', docId: r.slice(4) }); + } else { + sendWebSocketMessage({ event: 'JOIN_ROOM', room: r }); + } + }); }; - const connectWS = () => { + if (isRealtimeConnected()) { + joinAll(); + } + + const unsubRejoin = onRealtimeStatus((connected) => { if (isClosed) return; - const token = getToken(); - if (!token) { - notify(false, 'none', 'Kein Authentifizierungs-Token vorhanden'); - return; + if (connected) { + joinAll(); } - - if (ws) { - try { ws.close(); } catch {} - ws = null; - } - - clearTimeout(watchdogTimer); - - try { - const url = buildWebSocketURL(options.listId); - ws = new WebSocket(url); - - // 3.5s watchdog for corporate proxies silently discarding Upgrade header - watchdogTimer = setTimeout(() => { - if (ws && ws.readyState !== WebSocket.OPEN && !isClosed) { - console.warn('[MindSift Realtime] WebSocket connection timeout (Proxy Upgrade block). Falling back to SSE...'); - try { ws.close(); } catch {} - ws = null; - connectSSE(); - } - }, 3500); - - ws.onopen = () => { - if (isClosed) return; - clearTimeout(watchdogTimer); - reconnectDelay = 2000; - notify(true, 'ws'); - if (options.docId) { - try { - ws?.send(JSON.stringify({ event: 'CRDT_JOIN', docId: options.docId })); - } catch {} - } - }; - - ws.onmessage = (evt) => { - if (isClosed) return; - try { - const msg = JSON.parse(evt.data) as WSMessage; - options.onMessage?.(msg); - } catch (e) { - console.error('WS JSON error', e); - } - }; - - ws.onclose = (evt: CloseEvent) => { - if (isClosed) return; - clearTimeout(watchdogTimer); - ws = null; - - const isProxyError = evt?.code === 1006 || evt?.code === 1002 || evt?.code === 1008; - if (isProxyError) { - console.warn(`[MindSift Realtime] Proxy WebSocket rejection detected (Code ${evt.code}). Falling back to SSE...`); - connectSSE(); - return; - } - - const reason = evt.reason || (evt.code === 1000 ? 'Normal geschlossen' : `Code ${evt.code}`); - notify(false, 'none', reason); - - clearTimeout(reconnectTimer); - const delay = reconnectDelay; - reconnectDelay = Math.min(30000, Math.round(reconnectDelay * 1.5)); - reconnectTimer = setTimeout(connectWS, delay); - }; - - ws.onerror = (e) => { - if (isClosed) return; - clearTimeout(watchdogTimer); - try { ws?.close(); } catch {} - }; - } catch (e: any) { - clearTimeout(watchdogTimer); - console.warn('[MindSift Realtime] WebSocket constructor failed, trying SSE fallback:', e); - connectSSE(); - } - }; - - connectWS(); + }); return { - getTransport: () => currentTransport, - isConnected: () => currentTransport !== 'none', + getTransport: () => activeTransport, + isConnected: () => isRealtimeConnected(), send: (msg: any) => { - const str = typeof msg === 'string' ? msg : JSON.stringify(msg); - if (ws && ws.readyState === WebSocket.OPEN) { - ws.send(str); - return; - } - // SSE Fallback: dispatch via REST endpoints - try { - const parsed = typeof msg === 'string' ? JSON.parse(msg) : msg; - if (parsed.event === 'CRDT_UPDATE' && parsed.docId) { - api.saveDocumentCRDT(parsed.docId, { update: parsed.payload?.update }); - } else if (parsed.event === 'CRDT_AWARENESS' && parsed.docId) { - api.sendDocumentAwareness(parsed.docId, parsed.payload); - } else if (parsed.event === 'CRDT_LEAVE' && parsed.docId) { - api.clearDocumentAwareness(parsed.docId); - if (sseId) api.joinSSERoom(sseId, `doc:${parsed.docId}`, 'leave'); - } else if (parsed.event === 'CRDT_JOIN' && parsed.docId) { - if (sseId) api.joinSSERoom(sseId, `doc:${parsed.docId}`, 'join'); - } else if (parsed.event === 'LOCK_DOC' && parsed.docId) { - api.lockDocument(parsed.docId); - } else if (parsed.event === 'UNLOCK_DOC' && parsed.docId) { - api.unlockDocument(parsed.docId); - } - } catch (err) { - console.error('Error dispatching fallback REST message for connection:', err); - } + if (isClosed) return; + sendWebSocketMessage(msg); }, close: () => { isClosed = true; - clearTimeout(watchdogTimer); - clearTimeout(reconnectTimer); - if (ws) { - try { ws.close(); } catch {} - ws = null; - } - if (es) { - try { es.close(); } catch {} - es = null; - } - notify(false, 'none'); + unsubStatus(); + unsubRejoin(); + if (unsubMsg) unsubMsg(); + joinedRooms.forEach((r) => { + globalSubscribedRooms.delete(r); + if (r.startsWith('doc:')) { + sendWebSocketMessage({ event: 'CRDT_LEAVE', docId: r.slice(4) }); + } else { + sendWebSocketMessage({ event: 'LEAVE_ROOM', room: r }); + } + }); }, }; } diff --git a/internal/ws/hub.go b/internal/ws/hub.go index 6863cbc..56cf61a 100644 --- a/internal/ws/hub.go +++ b/internal/ws/hub.go @@ -761,10 +761,11 @@ func (c *Client) readPump() { break } - // Handle client-initiated lock, CRDT, and presence messages + // Handle client-initiated lock, CRDT, room subscription, and presence messages var incoming struct { Event string `json:"event"` DocID string `json:"docId"` + Room string `json:"room"` Payload json.RawMessage `json:"payload"` } if json.Unmarshal(message, &incoming) == nil && c.userID != "" && c.userID != "anonymous" { @@ -777,6 +778,25 @@ func (c *Client) readPump() { docID = p.DocID } + // Room subscription without requiring docID + if incoming.Event == "JOIN_ROOM" { + room := incoming.Room + if room == "" { + room = docID + } + if room != "" { + c.hub.JoinRoom(c, room) + } + } else if incoming.Event == "LEAVE_ROOM" { + room := incoming.Room + if room == "" { + room = docID + } + if room != "" { + c.hub.LeaveRoom(c, room) + } + } + if docID != "" { switch incoming.Event { case "LOCK_DOC":