sandrodach/MindSift
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(collab): fix corporate SSE fallback, realtime lock propagation and prevent stale overwrite
MindSift CI & Docker Release / Build, Test in Container & Push (push) Successful in 1m1s
Details

Browse Source
This commit is contained in:
sandrodach committed 2026-10-27 15:21:05 +01:00
1 parent 996f27133c
commit a1c7a65d84
5 files changed
+238 -63
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontend/src/features/workspace/DocumentEditor.tsx
+162 -26
View File
@@ -599,32 +599,148 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
}
}, [registerLiveToggle, handleToggleLive]);
// Load initial locks on mount
useEffect(() => {
let cancelled = false;
api.getDocLocks(docId)
.then((res) => {
if (cancelled) return;
if (res.liveSession) {
setLiveSession(res.liveSession);
if (currentUserId && res.liveSession.initiatorId === currentUserId) {
setIsLiveEnabled(true);
}
}
if (Array.isArray(res.blockLocks)) {
const map: Record<string, BlockLockInfo> = {};
for (const l of res.blockLocks) {
map[l.blockId] = l;
}
setActiveBlockLocks(map);
}
})
.catch(() => {});
// Handle remote document updates (from other users saving via HTTP or realtime)
const handleRemoteDocUpdated = useCallback(
(remoteDoc: Document) => {
if (!remoteDoc || remoteDoc.id !== docId || !Array.isArray(remoteDoc.blocks)) return;
return () => {
cancelled = true;
const isTitleActivelyEditing =
titleInputRef.current !== null &&
document.activeElement === titleInputRef.current &&
typeof document !== 'undefined' &&
document.hasFocus();
let targetTitle = currentTitleRef.current;
if (!isTitleActivelyEditing && remoteDoc.title !== undefined) {
setTitle(remoteDoc.title);
currentTitleRef.current = remoteDoc.title;
targetTitle = remoteDoc.title;
}
// Merge remote blocks: preserve any block currently being actively typed by local user,
// update all other blocks from the authoritative server state.
const localFocusedId = focusedBlockIdRef.current;
const localBlocks = currentBlocksRef.current;
const localFocusedBlock = localFocusedId ? localBlocks.find((b) => b.id === localFocusedId) : null;
const mergedBlocks = remoteDoc.blocks.map((remoteB) => {
if (localFocusedId && remoteB.id === localFocusedId && localFocusedBlock) {
return { ...remoteB, content: localFocusedBlock.content };
}
return remoteB;
});
// If local newly-created block isn't on server yet, retain it
if (localFocusedId && localFocusedBlock && !mergedBlocks.some((b) => b.id === localFocusedId)) {
mergedBlocks.push(localFocusedBlock);
}
const nonEmpties = ensureNonEmptyBlocks(docId, mergedBlocks);
setBlocks(nonEmpties);
currentBlocksRef.current = nonEmpties;
const newSnapshot = JSON.stringify({
title: targetTitle,
blocks: nonEmpties,
tags: remoteDoc.tags || currentTagsRef.current || [],
});
lastSavedSnapshotRef.current = newSnapshot;
currentSnapshotRef.current = newSnapshot;
if (docRef.current) {
docRef.current = { ...docRef.current, ...remoteDoc };
}
onDocMetadataChange(remoteDoc);
},
[docId, onDocMetadataChange]
);
const handleRemoteBlocksPatched = useCallback(
(patch: { upserted?: DocumentBlock[]; deletedIDs?: string[]; blockOrder?: string[] }) => {
console.log('[DocumentEditor] DOC_BLOCKS_PATCHED empfangen:', patch);
const localFocusedId = focusedBlockIdRef.current;
const localBlocks = currentBlocksRef.current;
const localFocusedBlock = localFocusedId ? localBlocks.find((b) => b.id === localFocusedId) : null;
const blockMap = new Map<string, DocumentBlock>();
for (const b of localBlocks) {
blockMap.set(b.id, b);
}
if (Array.isArray(patch.upserted)) {
for (const u of patch.upserted) {
if (localFocusedId && u.id === localFocusedId && localFocusedBlock) {
blockMap.set(u.id, { ...u, content: localFocusedBlock.content });
} else {
blockMap.set(u.id, u);
}
}
}
if (Array.isArray(patch.deletedIDs)) {
for (const delId of patch.deletedIDs) {
if (delId !== localFocusedId) {
blockMap.delete(delId);
}
}
}
let reconciledBlocks: DocumentBlock[] = [];
if (Array.isArray(patch.blockOrder) && patch.blockOrder.length > 0) {
for (const id of patch.blockOrder) {
const b = blockMap.get(id);
if (b) reconciledBlocks.push(b);
}
blockMap.forEach((b, id) => {
if (!patch.blockOrder!.includes(id)) {
reconciledBlocks.push(b);
}
});
} else {
reconciledBlocks = Array.from(blockMap.values());
}
const nonEmpties = ensureNonEmptyBlocks(docId, reconciledBlocks);
setBlocks(nonEmpties);
currentBlocksRef.current = nonEmpties;
const newSnapshot = JSON.stringify({
title: currentTitleRef.current,
blocks: nonEmpties,
tags: currentTagsRef.current,
});
lastSavedSnapshotRef.current = newSnapshot;
currentSnapshotRef.current = newSnapshot;
},
[docId]
);
// Periodic poll for locks & updates (guarantees synchronization even across flaky networks)
useEffect(() => {
if (!docId) return;
const syncLocks = () => {
api.getDocLocks(docId)
.then((res) => {
if (Array.isArray(res.blockLocks)) {
const map: Record<string, BlockLockInfo> = {};
for (const l of res.blockLocks) {
map[l.blockId] = l;
}
setActiveBlockLocks(map);
}
if (res.liveSession) {
setLiveSession(res.liveSession);
} else {
setLiveSession(null);
}
})
.catch(() => {});
};
}, [docId, currentUserId]);
syncLocks();
const interval = window.setInterval(syncLocks, 3000);
return () => window.clearInterval(interval);
}, [docId]);
// Subscribe to realtime lock & live session events
useEffect(() => {
@@ -633,6 +749,7 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
onMessage: (msg: WSMessage) => {
if (msg.event === 'DOC_BLOCK_LOCKED') {
const p = (msg.payload?.lock || msg.payload) as BlockLockInfo;
console.log('[DocumentEditor] DOC_BLOCK_LOCKED empfangen:', p);
if (p?.docId === docId && p?.blockId) {
setActiveBlockLocks((prev) => ({
...prev,
@@ -641,6 +758,7 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
}
} else if (msg.event === 'DOC_BLOCK_UNLOCKED') {
const p = msg.payload;
console.log('[DocumentEditor] DOC_BLOCK_UNLOCKED empfangen:', p);
if (p?.docId === docId && p?.blockId) {
setActiveBlockLocks((prev) => {
const next = { ...prev };
@@ -648,8 +766,21 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
return next;
});
}
} else if (msg.event === 'DOC_UPDATED') {
const p = msg.payload as Document;
console.log('[DocumentEditor] DOC_UPDATED von anderem Benutzer empfangen:', p);
if (p?.id === docId && Array.isArray(p?.blocks)) {
handleRemoteDocUpdated(p);
}
} else if (msg.event === 'DOC_BLOCKS_PATCHED') {
const p = msg.payload;
console.log('[DocumentEditor] DOC_BLOCKS_PATCHED von Server empfangen:', p);
if (p?.docId === docId) {
handleRemoteBlocksPatched(p);
}
} else if (msg.event === 'DOC_LIVE_STARTED') {
const s = (msg.payload?.session || msg.payload) as LiveSessionInfo;
console.log('[DocumentEditor] DOC_LIVE_STARTED empfangen:', s);
if (s?.docId === docId) {
setLiveSession(s);
if (currentUserId && s.initiatorId === currentUserId) {
@@ -657,6 +788,7 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
}
}
} else if (msg.event === 'DOC_LIVE_ENDED') {
console.log('[DocumentEditor] DOC_LIVE_ENDED empfangen');
if (msg.payload?.docId === docId) {
setLiveSession(null);
setIsLiveEnabled(false);
@@ -668,7 +800,7 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
return () => {
conn.close();
};
}, [docId, currentUserId]);
}, [docId, currentUserId, handleRemoteDocUpdated, handleRemoteBlocksPatched]);
// Block focus/blur locks
const localUserColor = useMemo(
@@ -682,7 +814,10 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
(blockId: string) => {
if (!effectiveCanEdit) return;
focusedBlockIdRef.current = blockId;
api.lockBlock(docId, blockId, localUserColor, 30).catch(() => {});
console.log(`[DocumentEditor] Fokus auf Block "${blockId}". Sende Lock an Server...`);
api.lockBlock(docId, blockId, localUserColor, 30).catch((err) => {
console.warn(`[DocumentEditor] Lock für Block "${blockId}" fehlgeschlagen:`, err);
});
if (lockKeepaliveTimerRef.current !== null) {
window.clearInterval(lockKeepaliveTimerRef.current);
@@ -698,6 +833,7 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
const handleBlurBlock = useCallback(
(blockId: string) => {
console.log(`[DocumentEditor] Blur von Block "${blockId}". Gebe Lock frei und sichere per AutoSave...`);
if (lockKeepaliveTimerRef.current !== null) {
window.clearInterval(lockKeepaliveTimerRef.current);
lockKeepaliveTimerRef.current = null;
frontend/src/features/workspace/OutlinerBlock.tsx
+2
View File
@@ -314,6 +314,7 @@ export const OutlinerBlock: React.FC<OutlinerBlockProps> = ({
};
const handleFocus = (e: React.FocusEvent<HTMLTextAreaElement>) => {
if (isLockedByOther) return;
isFocusedRef.current = true;
onFocusBlockRef.current(block.id);
sendCursorMovedRef.current?.(block.id, e.target.selectionStart, e.target.selectionEnd);
@@ -321,6 +322,7 @@ export const OutlinerBlock: React.FC<OutlinerBlockProps> = ({
const handleBlur = () => {
isFocusedRef.current = false;
if (isLockedByOther) return;
if (debounceTimerRef.current !== null) {
window.clearTimeout(debounceTimerRef.current);
debounceTimerRef.current = null;
frontend/src/lib/api.ts
+49 -27
View File
@@ -650,17 +650,22 @@ export const api = {
},
async lockBlock(docId: string, blockId: string, userColor?: string, ttlSeconds: number = 30): Promise<BlockLockInfo> {
console.log(`[MindSift Lock] Lock anfordern für Block "${blockId}" in Notiz "${docId}"...`);
const res = await request<any>(`/api/v1/documents/${docId}/blocks/${encodeURIComponent(blockId)}/lock`, {
method: 'POST',
body: JSON.stringify({ userColor, ttlSeconds }),
body: JSON.stringify({ userColor, ttlSeconds, ttl: ttlSeconds }),
});
console.log(`[MindSift Lock] Lock erfolgreich erhalten für "${blockId}":`, res?.lock || res);
return res?.lock || res;
},
async unlockBlock(docId: string, blockId: string): Promise<{ status: string }> {
return request<{ status: string }>(`/api/v1/documents/${docId}/blocks/${encodeURIComponent(blockId)}/unlock`, {
console.log(`[MindSift Lock] Lock freigeben für Block "${blockId}" in Notiz "${docId}"...`);
const res = await request<{ status: string }>(`/api/v1/documents/${docId}/blocks/${encodeURIComponent(blockId)}/unlock`, {
method: 'POST',
});
console.log(`[MindSift Lock] Lock freigegeben für "${blockId}"`);
return res;
},
async startLiveSession(docId: string): Promise<LiveSessionInfo> {
@@ -1140,7 +1145,7 @@ export function onRealtimeStatus(callback: (connected: boolean, transport: Realt
};
}
let isWsBlockedByProxy = false;
let isWsBlockedByProxy = typeof window !== 'undefined' && sessionStorage.getItem('mindsift_ws_blocked') === 'true';
let sseReconnectTimer: any = null;
let sseReconnectDelay = 2000;
const globalSubscribedRooms = new Set<string>();
@@ -1148,6 +1153,7 @@ const globalSubscribedRooms = new Set<string>();
function notifyRealtimeStatusChange(connected: boolean, transport: RealtimeTransport) {
isRealtimeConnectedState = connected;
activeTransport = transport;
console.log(`[MindSift Realtime] Verbindungsstatus geändert: connected=${connected}, transport=${transport}`);
realtimeStatusListeners.forEach((fn) => {
try {
fn(connected, transport);
@@ -1158,6 +1164,25 @@ function notifyRealtimeStatusChange(connected: boolean, transport: RealtimeTrans
window.dispatchEvent(new CustomEvent('mindsift:ws:status', { detail: { connected, transport } }));
}
function switchToSSE(reason: string) {
console.warn(`[MindSift Realtime] ${reason} -> Schalte sofort und dauerhaft auf Server-Sent Events (SSE) um.`);
isWsBlockedByProxy = true;
try {
sessionStorage.setItem('mindsift_ws_blocked', 'true');
} catch {}
clearTimeout(wsConnectionWatchdog);
clearTimeout(wsReconnectTimer);
if (activeGlobalWS) {
try {
activeGlobalWS.onclose = null;
activeGlobalWS.onerror = null;
activeGlobalWS.close();
} catch {}
activeGlobalWS = null;
}
connectSSE();
}
export function connectSSE(rooms?: string[]): void {
if (rooms) {
rooms.forEach((r) => globalSubscribedRooms.add(r));
@@ -1173,15 +1198,17 @@ export function connectSSE(rooms?: string[]): void {
try {
const url = buildSSEURL(undefined, undefined, Array.from(globalSubscribedRooms));
console.log(`[MindSift Realtime] Verbinde SSE zu ${url}...`);
const es = new EventSource(url);
activeGlobalSSE = es;
es.onopen = () => {
console.info('[MindSift Realtime] Server-Sent Events (SSE) verbunden (Proxy-Fallback aktiv).');
console.info('[MindSift Realtime] Server-Sent Events (SSE) erfolgreich verbunden (Corporate Fallback aktiv).');
sseReconnectDelay = 2000;
notifyRealtimeStatusChange(true, 'sse');
if (activeSSEId && globalSubscribedRooms.size > 0) {
globalSubscribedRooms.forEach((room) => {
console.log(`[MindSift Realtime] SSE Rejoin Room "${room}"`);
api.joinSSERoom(activeSSEId!, room, 'join').catch(() => {});
});
}
@@ -1190,10 +1217,13 @@ export function connectSSE(rooms?: string[]): void {
es.onmessage = (event) => {
try {
const msg = JSON.parse(event.data) as WSMessage;
console.log('[MindSift Realtime Event] [SSE]', msg.event, msg.payload);
if (msg.event === 'SSE_CONNECTED') {
activeSSEId = msg.payload?.sseId || null;
console.log('[MindSift Realtime] SSE-Client registriert mit ID:', activeSSEId);
if (activeSSEId && globalSubscribedRooms.size > 0) {
globalSubscribedRooms.forEach((room) => {
console.log(`[MindSift Realtime] Trete Raum per SSE bei: "${room}"`);
api.joinSSERoom(activeSSEId!, room, 'join').catch(() => {});
});
}
@@ -1212,6 +1242,7 @@ export function connectSSE(rooms?: string[]): void {
};
es.onerror = (e) => {
console.warn('[MindSift Realtime] SSE Verbindungsfehler:', e);
if (activeTransport === 'sse') {
notifyRealtimeStatusChange(false, 'none');
}
@@ -1223,11 +1254,7 @@ export function connectSSE(rooms?: string[]): void {
sseReconnectDelay = Math.min(30000, Math.round(sseReconnectDelay * 1.5));
sseReconnectTimer = setTimeout(() => {
if (!manualDisconnect && getToken()) {
if (isWsBlockedByProxy) {
connectSSE();
} else {
connectWebSocket();
}
connectSSE();
}
}, delay);
}
@@ -1245,6 +1272,7 @@ export function connectWebSocket(): void {
// If proxy blocked WebSocket previously, do not attempt failing WebSocket; stay on SSE
if (isWsBlockedByProxy) {
console.info('[MindSift Realtime] WebSocket im Firmennetzwerk blockiert (Cache aktiv). Verwende direkt SSE.');
connectSSE();
return;
}
@@ -1259,22 +1287,20 @@ export function connectWebSocket(): void {
try {
const url = buildWebSocketURL();
console.log(`[MindSift Realtime] Versuche WebSocket-Verbindung zu ${url}...`);
const ws = new WebSocket(url);
activeGlobalWS = ws;
// 3.5s watchdog for corporate proxies silently stripping Upgrade header
// 2.5s watchdog for corporate proxies silently stripping Upgrade header or hanging
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();
switchToSSE('WebSocket Timeout nach 2.5s (Proxy Upgrade Header blockiert)');
}
}, 3500);
}, 2500);
ws.onopen = () => {
clearTimeout(wsConnectionWatchdog);
console.info('[MindSift Realtime] WebSocket erfolgreich verbunden!');
if (activeGlobalSSE) {
try { activeGlobalSSE.close(); } catch {}
activeGlobalSSE = null;
@@ -1284,6 +1310,7 @@ export function connectWebSocket(): void {
if (globalSubscribedRooms.size > 0) {
globalSubscribedRooms.forEach((room) => {
try {
console.log(`[MindSift Realtime] Trete Raum per WS bei: "${room}"`);
ws.send(JSON.stringify({ event: 'JOIN_ROOM', room }));
} catch {}
});
@@ -1293,6 +1320,7 @@ export function connectWebSocket(): void {
ws.onmessage = (event) => {
try {
const msg = JSON.parse(event.data) as WSMessage;
console.log('[MindSift Realtime Event] [WS]', msg.event, msg.payload);
wsListeners.forEach((fn) => {
try {
fn(msg);
@@ -1310,11 +1338,9 @@ export function connectWebSocket(): void {
clearTimeout(wsConnectionWatchdog);
activeGlobalWS = null;
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 dauerhaft zu SSE...');
isWsBlockedByProxy = true;
connectSSE();
// In corporate environments, blocked WebSockets close abruptly without clean handshake
if (!evt.wasClean || evt.code === 1006 || evt.code === 1002 || evt.code === 1008 || evt.code === 1005) {
switchToSSE(`WebSocket unsauber beendet (Code ${evt.code}, wasClean=${evt.wasClean})`);
return;
}
@@ -1336,14 +1362,10 @@ export function connectWebSocket(): void {
ws.onerror = (e) => {
clearTimeout(wsConnectionWatchdog);
try {
ws.close();
} catch {}
switchToSSE('WebSocket Fehler (Handshake/Verbindung verweigert)');
};
} catch (e) {
clearTimeout(wsConnectionWatchdog);
isWsBlockedByProxy = true;
connectSSE();
switchToSSE('WebSocket Exception beim Initialisieren');
}
}
frontend/src/lib/useDocumentSync.ts
+6
View File
@@ -159,6 +159,12 @@ export function useDocumentSync(
const connect = () => {
if (isDisposed) return;
if (typeof window !== 'undefined' && sessionStorage.getItem('mindsift_ws_blocked') === 'true') {
console.info('[DocSync] WebSockets im Firmennetzwerk blockiert (Cache). Live-Transport übersprungen.');
setStatus('blocked');
return;
}
const token = getToken();
if (!token) {
setStatus('offline');