From 8b7a53daaba4117d8209e19668b3eeeb8b69c6f7 Mon Sep 17 00:00:00 2001 From: Sandro Date: Wed, 28 Oct 2026 14:10:29 +0100 Subject: [PATCH] feat(notes): push, long polling and fallback detection for note sync Phase 4 of the collaboration plan: - The sync engine listens for DOC_SEQ in the note's realtime room and syncs right away. Without WebSocket/SSE it long-polls /docs/{id}/changes (25 s); after three failures it falls back to polling every 3 s and retries long polling when the browser comes back online. - With push or long polling the engine only syncs every 7 s while holding leases (renewal), otherwise every 30 s as a safety net. Focusing a block announces its lease immediately (otherwise it could take up to 30 s). - Realtime connection: SSE counts as connected only after its first event; a stream that stays silent for 5 s (buffering proxy) switches SSE off for the tab. App-level PING/PONG detects WebSockets that are accepted but forward nothing and switches to SSE after 35 s. - The editor shows the active channel for shared notes. - Tests: engine channel tests (push, heartbeat, lease on focus, long polling, interval fallback, switching), realtime fallback tests with fake WebSocket/EventSource, Go tests for PONG and DOC_SEQ delivery, and e2e/channels.e2e.mjs measuring latency behind six simulated proxy setups (0.2-0.3 s on all live channels, ~3 s with plain polling). Co-Authored-By: Claude Opus 5.5 --- Doku/api/websocket.md | 11 +- Doku/modules/workspace.md | 8 +- Doku/plans/kollaboration-notizen.md | 2 +- Doku/reverse-proxy.md | 4 + e2e/README.md | 6 +- e2e/channels.e2e.mjs | 119 +++++++++++ e2e/lock.e2e.mjs | 2 +- e2e/run.sh | 2 +- .../src/features/workspace/DocumentEditor.tsx | 25 ++- frontend/src/lib/__tests__/realtime.test.ts | 132 ++++++++++++ frontend/src/lib/api.ts | 56 ++++- .../src/lib/docSync/__tests__/engine.test.ts | 202 +++++++++++++++++- .../src/lib/docSync/__tests__/undo.test.ts | 1 + frontend/src/lib/docSync/engine.ts | 136 +++++++++++- frontend/src/lib/docSync/push.ts | 25 +++ frontend/src/lib/docSync/transport.ts | 14 +- internal/api/realtime_test.go | 70 ++++++ internal/ws/hub.go | 5 +- 18 files changed, 783 insertions(+), 37 deletions(-) create mode 100644 e2e/channels.e2e.mjs create mode 100644 frontend/src/lib/__tests__/realtime.test.ts create mode 100644 frontend/src/lib/docSync/push.ts diff --git a/Doku/api/websocket.md b/Doku/api/websocket.md index e905102..52a3c88 100644 --- a/Doku/api/websocket.md +++ b/Doku/api/websocket.md @@ -11,7 +11,14 @@ Der Server verteilt Änderungen als JSON-Ereignisse an verbundene Clients. Die E | WebSocket | `GET /api/v1/ws` (auch `/ws/lists/{id}`, `/ws/spaces/{id}`) | Ping alle 30 s, Lesefrist 60 s, Nachrichten bis 4 MB | | SSE | `GET /api/v1/events` | Kommentar-Ping alle 15 s, Header `X-Accel-Buffering: no` | -Der Client versucht zuerst WebSocket. Gelingt der Aufbau nicht innerhalb von 2,5 s oder bricht die Verbindung unsauber ab, wechselt er auf SSE und bleibt dabei (Firmen-Proxys). Dort liefert der Server `SSE_CONNECTED` mit einer `sseId`; Räume werden über `POST /api/v1/events/rooms` (`sseId`, `room`, `action`: `join`/`leave`) abonniert. +Der Client versucht zuerst WebSocket und wechselt in folgenden Fällen für die Sitzung des Tabs auf SSE: + +- Aufbau gelingt nicht innerhalb von 2,5 s, oder die Verbindung bricht unsauber ab (z. B. Code 1006/1008). +- **Herzschlag:** Alle 15 s schickt der Client `{"event":"PING"}`, der Server antwortet `PONG`. Kommt 35 s lang keine Nachricht, gilt der WebSocket als stumm (Proxy nimmt das Upgrade an, leitet aber nichts weiter). + +Über SSE liefert der Server zuerst `SSE_CONNECTED` mit einer `sseId`; erst dann gilt SSE als verbunden. Bleibt dieses erste Ereignis 5 s nach dem Öffnen aus, puffert ein Proxy den Stream: SSE wird für den Tab abgeschaltet. Räume werden über `POST /api/v1/events/rooms` (`sseId`, `room`, `action`: `join`/`leave`) abonniert. + +Ohne WebSocket und SSE nutzt die Notiz-Synchronisation Long-Polling (`GET /docs/{id}/changes`, bis 25 s), und wenn auch das dreimal scheitert, eine Abfrage alle 3 s (siehe [modules/workspace.md](../modules/workspace.md)). **Authentifizierung:** Token als `?token=` oder `Authorization`-Header. Ohne gültiges Token antworten `/ws` und `/events` mit `401`. Das Token wird nur beim Verbinden geprüft ([FLAW-22](../flaws.md)). @@ -23,7 +30,7 @@ Der Client versucht zuerst WebSocket. Gelingt der Aufbau nicht innerhalb von 2,5 { "event": "TASK_UPDATED", "listId": "", "payload": { … }, "timestamp": "2026-10-27T12:00:00Z" } ``` -Client → Server versteht der Server nur: `{"event":"JOIN_ROOM","room":""}` und `{"event":"LEAVE_ROOM","room":""}`. Räume heißen `doc:` für Notizen oder `` für geteilte Listen. +Client → Server versteht der Server nur: `{"event":"JOIN_ROOM","room":""}`, `{"event":"LEAVE_ROOM","room":""}` und `{"event":"PING"}` (Antwort `PONG`). Räume heißen `doc:` für Notizen oder `` für geteilte Listen. ## Zustellung diff --git a/Doku/modules/workspace.md b/Doku/modules/workspace.md index 7d06be3..502dcb9 100644 --- a/Doku/modules/workspace.md +++ b/Doku/modules/workspace.md @@ -18,7 +18,13 @@ Code: [WorkspaceView.tsx](../../frontend/src/features/workspace/WorkspaceView.ts - **Links:** `[[Notiztitel]]`, Markdown-Links `[Titel](url)` und reine URLs werden als Token dargestellt; `LinkModal` hilft beim Verknüpfen von Notizen. - **Einfügen:** Mehrzeiliger Text wird anhand von Markdown-Präfixen und Einrückung in Blöcke zerlegt; Überschriften gruppieren darunterliegenden Text und Code. - **Speichern:** Jede Notiz hat im Browser eine `DocSyncEngine` ([lib/docSync/](../../frontend/src/lib/docSync/)). Der Editor meldet jede Änderung der Block-Liste; die Engine übersetzt sie in Operationen, speichert sie zuerst in IndexedDB und sendet sie dann gesammelt (300 ms) an `POST /docs/{id}/sync`. Beim Öffnen zeigt sie sofort den gespeicherten Stand, auch offline. -- **Abgleich:** Bei geteilten oder mehrfach geöffneten Notizen fragt die Engine alle 3 s nach Änderungen, sonst alle 30 s, sowie sofort bei Rückkehr ins Netz oder in den Tab. Beim Verlassen sendet sie Ausstehendes mit `keepalive` und gibt ihre Sperren frei. +- **Abgleich bei geteilten oder mehrfach geöffneten Notizen**, nach Verfügbarkeit: + 1. **Push** über WebSocket oder SSE: Der Server meldet `DOC_SEQ`, die Engine gleicht sofort ab (im Test 0,2–0,3 s). + 2. **Long-Polling** (`/docs/{id}/changes`), wenn ein Proxy WebSocket und SSE blockiert oder SSE puffert: ebenso schnell. + 3. **Abfrage alle 3 s**, wenn auch Long-Polling dreimal scheitert (bei `online` wird Long-Polling erneut versucht). + + Solange 1. oder 2. läuft, gleicht die Engine zusätzlich alle 7 s ab, wenn sie Sperren hält (Verlängerung), sonst alle 30 s zur Sicherheit. Ein Fokuswechsel meldet die Sperre sofort an. Der Editor zeigt den Weg an („Live“, „Live (Long-Polling)“, „Abgleich alle 3 s“). +- **Private Notizen**, die nur einmal geöffnet sind, gleichen alle 30 s ab, außerdem sofort bei Rückkehr ins Netz oder in den Tab. Beim Verlassen sendet die Engine Ausstehendes mit `keepalive` und gibt ihre Sperren frei. - **Textkonflikte:** Jede Textänderung trägt die Version, die der Nutzer beim Tippen vor sich hatte. Hat jemand anderes den Bullet inzwischen geändert oder gesperrt, legt der Server eine Konfliktkopie an; es wird nie still überschrieben. - **Sperren:** Der Bullet mit dem Cursor, verlassene Bullets für 20 s und neu angelegte Bullets werden gehalten. Bullets, die jemand anderes hält, zeigen ein Schloss mit Namen und sind schreibgeschützt; Enter darin legt einen neuen Bullet darunter an. Löschen, Zusammenführen, Ein-/Ausrücken und Ziehen werden bei gesperrten Bullets nicht ausgeführt (Hinweis unten im Editor). - **Präsenz:** Im Kopf der Notiz steht, wer sie gerade sonst noch geöffnet hat. diff --git a/Doku/plans/kollaboration-notizen.md b/Doku/plans/kollaboration-notizen.md index e7d194b..d790cc6 100644 --- a/Doku/plans/kollaboration-notizen.md +++ b/Doku/plans/kollaboration-notizen.md @@ -281,7 +281,7 @@ Damit private Notizen keinen unnötigen Aufwand erzeugen, liefert der Server im | 1 | Operations-Log, `/snapshot`, `/sync`, `/changes`, `/leave`, Konfliktregeln, Leases, zentrale `DocRole`, Simulationstests. **Erledigt.** Abweichungen: Die bestehenden Tabellen `blocks`/`documents` wurden erweitert statt ersetzt (Baum per `parent_id`/`order_key`, alte Daten werden umgewandelt); der bisherige Speicherweg (`PUT`/`PATCH`) läuft als Adapter durch dieselbe Engine. `DocRole` arbeitet auf den vorhandenen Freigabetabellen; deren Zusammenlegung zu `workspace_access` erfolgt mit dem Freigabe-Dialog in Phase 5. `/changes` liefert nur `{headSeq, changed}`, die Daten kommen immer über `/sync`. | Server fertig und abgesichert | | 2 | `SyncEngine` mit IndexedDB-Warteschlange, nur Intervall-Abfrage als Transport, Editor umgestellt. **Erledigt.** Abweichung: Die Editor-Komponenten arbeiten weiter mit der flachen Liste (Ebene + Reihenfolge); die Engine übersetzt jede Änderung per Vergleich in Operationen (`lib/docSync/diff.ts`). Sperren werden bereits gehalten (Fokus, 20 s Nachlauf, neue Bullets), aber noch nicht angezeigt. Ende-zu-Ende-Test mit zwei Browsern: `e2e/`. | Bearbeiten funktioniert überall, auch offline | | 3 | Leases und Präsenz in UI, Konfliktblöcke mit Übernehmen/Verwerfen, Rückgängig/Wiederholen, FLAW-25/26. **Erledigt.** Ergänzungen: Enter in einem fremd gesperrten Bullet legt einen neuen Bullet darunter an (der gesperrte bleibt unberührt); „Übernehmen“/„Verwerfen“ prüfen auch die Sperre der Kopie selbst. | Gemeinsames Bearbeiten ohne Kollisionen, Strg+Z | -| 4 | Push-Stufen WebSocket → SSE → Long-Polling mit automatischer Erkennung | Live-Gefühl, wo das Netz es zulässt | +| 4 | Push-Stufen WebSocket → SSE → Long-Polling → Intervall mit automatischer Erkennung. **Erledigt.** Ergänzungen: Herzschlag `PING`/`PONG` erkennt WebSockets, die angenommen werden, aber nichts weiterleiten (Wechsel auf SSE nach 35 s ohne Antwort); Sperren werden beim Fokuswechsel sofort angemeldet. Abweichung: Kein automatischer erneuter WebSocket-Versuch alle 5 Minuten – die Erkennung „WebSocket bzw. SSE blockiert“ gilt für die Sitzung des Tabs, ein neuer Tab versucht es erneut. Browser-Test: `e2e/channels.e2e.mjs`. | Live-Gefühl, wo das Netz es zulässt | | 5 | Freigabe-Dialog mit Bearbeiter-Rolle (**vorgezogen und erledigt**, für Notizen und geteilte Ordner), Ordnerfreigaben, Altlasten entfernen (`PUT`-Blockersatz, `PATCH /blocks`, `doc_shares`, `doc_team_shares`, Zwangsmigration auf „read“) | Feature komplett | | 6 | Härtung: Playwright-Tests, Log-Kompaktierung (Snapshot statt Log für Clients mit sehr altem `lastSeq`), Aufräumen der Grabsteine | Langzeitbetrieb | diff --git a/Doku/reverse-proxy.md b/Doku/reverse-proxy.md index ebf6a6d..864449c 100644 --- a/Doku/reverse-proxy.md +++ b/Doku/reverse-proxy.md @@ -193,6 +193,10 @@ VirtualHost-Konfiguration (`/etc/apache2/sites-available/mindsift.conf`): --- +## Wenn der Proxy nicht angepasst werden kann + +MindSift funktioniert auch hinter Proxys, die WebSocket blocken, Streams puffern oder nur kurze Anfragen erlauben. Die Notiz-Synchronisation wählt automatisch den besten verfügbaren Weg (WebSocket → SSE → Long-Polling → Abfrage alle 3 s); der Editor zeigt ihn an. Long-Polling-Anfragen dauern höchstens 25 s und bleiben damit unter üblichen Proxy-Grenzen. Details: [api/websocket.md](api/websocket.md). + ## 🔍 Diagnose & Verifikation Mit den folgenden Befehlen lässt sich die Erreichbarkeit und Streaming-Funktion prüfen: diff --git a/e2e/README.md b/e2e/README.md index e485079..4a030fe 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -32,4 +32,8 @@ Der Eigentümer teilt eine Notiz über den Freigabe-Dialog mit Rolle „Bearbeit Sperren, Präsenz, Rückgängig und Konfliktkopien mit zwei Browsern: Schloss und Schreibschutz am fremd bearbeiteten Bullet, Hinweis statt halb ausgeführter Aktion, `Strg+Z`/`Strg+Umschalt+Z`, Tippen direkt nach Enter, Markierung von Konfliktkopien sowie „Übernehmen“ (verweigert bei fremder Sperre, erfolgreich für den Sperrinhaber). -`collab.e2e.mjs` und `lock.e2e.mjs` dauern wegen des Sperrablaufs je etwa eine Minute. Er läuft nicht in der CI, weil er Browser-Images braucht. +## channels.e2e.mjs + +Der Eigentümer tippt mit normaler Verbindung, der Bearbeiter sitzt hinter einem simulierten Proxy. Für jede Situation wird der gewählte Weg und die Zeit bis zur Anzeige gemessen: nichts blockiert (WebSocket), WebSocket abgelehnt (SSE), WebSocket angenommen aber stumm (SSE nach Herzschlag-Ausfall), WebSocket und SSE blockiert sowie SSE hängend (Long-Polling), nur normale Anfragen (Abfrage alle 3 s). + +`collab.e2e.mjs`, `lock.e2e.mjs` und `channels.e2e.mjs` dauern je etwa eine Minute. Die Tests laufen nicht in der CI, weil sie Browser-Images brauchen. diff --git a/e2e/channels.e2e.mjs b/e2e/channels.e2e.mjs new file mode 100644 index 0000000..b1f0ae0 --- /dev/null +++ b/e2e/channels.e2e.mjs @@ -0,0 +1,119 @@ +// How fast changes arrive behind different (simulated) proxies. The admin edits with a normal connection; +// the editor's browser blocks WebSocket, SSE and/or long polling. Needs a server in demo mode. +import { chromium } from 'playwright'; + +const BASE = process.env.MINDSIFT_URL || 'http://localhost:18080'; +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); +let failures = 0; +const check = (cond, msg) => { + console.log(`${cond ? 'OK ' : 'FAIL'} ${msg}`); + if (!cond) failures++; +}; + +async function api(token, method, path, body) { + const res = await fetch(BASE + path, { + method, + headers: { 'Content-Type': 'application/json', ...(token ? { Authorization: `Bearer ${token}` } : {}) }, + body: body ? JSON.stringify(body) : undefined, + }); + const text = await res.text(); + if (!res.ok) throw new Error(`${method} ${path}: ${res.status} ${text}`); + return text ? JSON.parse(text) : null; +} + +async function waitFor(fn, ms) { + const end = Date.now() + ms; + while (Date.now() < end) { + try { + const v = await fn(); + if (v) return v; + } catch {} + await sleep(100); + } + return null; +} + +const editorValues = (page) => + page.$$eval('textarea', (els) => els.filter((e) => !e.placeholder?.startsWith('Titel')).map((e) => e.value)); +const channelOf = (page) => page.getAttribute('[data-sync-channel]', 'data-sync-channel'); + +async function openDoc(browser, token, title, block = {}) { + const ctx = await browser.newContext({ viewport: { width: 1280, height: 900 } }); + // A blocking proxy refuses the connection (policy violation); a clean close would just mean "reconnect" + if (block.ws === 'reject') await ctx.routeWebSocket(/\/api\/v1\/ws/, (ws) => ws.close({ code: 1008, reason: 'blocked by proxy' })); + if (block.ws === 'silent') await ctx.routeWebSocket(/\/api\/v1\/ws/, () => {}); // accepted, nothing forwarded + if (block.sse === 'abort') await ctx.route('**/api/v1/events**', (r) => r.abort()); + if (block.sse === 'hang') await ctx.route('**/api/v1/events**', () => {}); // never answered + if (block.longpoll) await ctx.route('**/changes**', (r) => r.abort()); + await ctx.addInitScript((t) => localStorage.setItem('mindsift_jwt_token', t), token); + const page = await ctx.newPage(); + await page.goto(BASE); + await page.getByRole('button', { name: /Workspace/ }).first().click(); + await page.getByText(title, { exact: true }).first().click(); + await page.waitForFunction(() => document.querySelectorAll('textarea').length > 1, null, { timeout: 10000 }); + return { ctx, page }; +} + +async function typeAtEnd(page, text) { + await page.bringToFront(); + // Place the cursor via script: Playwright's actionability checks wait for animation frames, which + // pages in the background do not get while the other browser window is in front + const ok = await page.evaluate(() => { + const el = Array.from(document.querySelectorAll('textarea')).find((e) => e.value.startsWith('Live-Test')); + if (!el) return false; + el.focus(); + el.setSelectionRange(el.value.length, el.value.length); + return true; + }); + if (!ok) throw new Error('Live-Test field not found'); + await page.keyboard.type(text); +} + +const TITLE = `Kanal-Test ${Date.now()}`; + +const MODES = [ + { name: 'nichts blockiert', block: {}, channel: 'ws', maxMs: 2500 }, + { name: 'WebSocket-Handshake abgelehnt', block: { ws: 'reject' }, channel: 'sse', maxMs: 2500 }, + { name: 'WebSocket angenommen, leitet nichts weiter', block: { ws: 'silent' }, channel: 'sse', maxMs: 2500, switchMs: 60000 }, + { name: 'WebSocket und SSE blockiert', block: { ws: 'reject', sse: 'abort' }, channel: 'longpoll', maxMs: 2500 }, + { name: 'SSE-Verbindung hängt', block: { ws: 'reject', sse: 'hang' }, channel: 'longpoll', maxMs: 2500 }, + { name: 'nur normale Anfragen möglich', block: { ws: 'reject', sse: 'abort', longpoll: true }, channel: 'interval', maxMs: 6000, switchMs: 15000 }, +]; + +const main = async () => { + const admin = (await api(null, 'POST', '/api/v1/auth/login', { login: 'admin', password: 'admin' })).token; + const user = (await api(null, 'POST', '/api/v1/auth/login', { login: 'user', password: 'user' })).token; + const dir = await api(admin, 'GET', '/api/v1/users/directory'); + const userId = dir.find((u) => u.username === 'user' || u.email?.startsWith('user')).id; + const doc = await api(admin, 'POST', '/api/v1/documents', { + title: TITLE, + blocks: [{ id: `ch_${Date.now()}`, type: 'bullet', content: 'Live-Test', level: 0 }], + }); + await api(admin, 'POST', `/api/v1/documents/${doc.id}/shares`, { userId, permission: 'write' }); + + const browser = await chromium.launch(); + const A = await openDoc(browser, admin, TITLE); + + for (const [i, mode] of MODES.entries()) { + const B = await openDoc(browser, user, TITLE, mode.block); + const channel = await waitFor(async () => ((await channelOf(B.page)) === mode.channel ? mode.channel : null), mode.switchMs || 10000); + check(channel === mode.channel, `${mode.name}: Weg ${(await channelOf(B.page)) || '?'} (erwartet ${mode.channel})`); + + const marker = ` r${i}`; + const start = Date.now(); + await typeAtEnd(A.page, marker); + const arrived = await waitFor(async () => (await editorValues(B.page)).some((v) => v.endsWith(marker)), mode.maxMs + 4000); + const ms = Date.now() - start; + check(!!arrived && ms <= mode.maxMs, `${mode.name}: Änderung kam nach ${arrived ? ms : '–'} ms an (Ziel ≤ ${mode.maxMs} ms)`); + await B.ctx.close(); + } + + await browser.close(); + console.log(failures ? `${failures} FEHLGESCHLAGEN` : 'ALLE PRÜFUNGEN BESTANDEN'); + process.exit(failures ? 1 : 0); +}; + +main().catch((e) => { + console.error(e); + process.exit(2); +}); diff --git a/e2e/lock.e2e.mjs b/e2e/lock.e2e.mjs index f998d74..d1d9d47 100644 --- a/e2e/lock.e2e.mjs +++ b/e2e/lock.e2e.mjs @@ -43,7 +43,7 @@ async function focusField(page, text) { const handle = await page.waitForFunction( (t) => Array.from(document.querySelectorAll('textarea')).find((e) => e.value === t) || null, text, - { timeout: 5000 } + { timeout: 5000, polling: 100 } // rAF polling stalls in background pages ); const el = handle.asElement(); await el.click(); diff --git a/e2e/run.sh b/e2e/run.sh index 64efd7b..5152375 100755 --- a/e2e/run.sh +++ b/e2e/run.sh @@ -15,4 +15,4 @@ for _ in $(seq 1 60); do curl -s -m 1 http://localhost:18080/api/v1/health >/dev echo "Starte Browser-Tests …" "$ENGINE" run --rm --network host -v "$ROOT/e2e":/e2e:Z -w /e2e mcr.microsoft.com/playwright:v1.48.2-jammy \ - sh -c 'npm install --no-audit --no-fund >/dev/null 2>&1 && node collab.e2e.mjs && node share.e2e.mjs && node lock.e2e.mjs' + sh -c 'npm install --no-audit --no-fund >/dev/null 2>&1 && node collab.e2e.mjs && node share.e2e.mjs && node lock.e2e.mjs && node channels.e2e.mjs' diff --git a/frontend/src/features/workspace/DocumentEditor.tsx b/frontend/src/features/workspace/DocumentEditor.tsx index e2af139..71fc774 100644 --- a/frontend/src/features/workspace/DocumentEditor.tsx +++ b/frontend/src/features/workspace/DocumentEditor.tsx @@ -687,7 +687,7 @@ export const DocumentEditor: React.FC = ({ } return ( -
+
{/* Top Document Action Bar */}
@@ -766,6 +766,29 @@ export const DocumentEditor: React.FC = ({ {/* Right Action Buttons */}
+ {/* How changes of others arrive (only relevant for shared notes) */} + {engineState && (engineState.shared || engineState.sessions > 1) && ( + + + {engineState.channel === 'longpoll' ? 'Live (Long-Polling)' : engineState.channel === 'interval' ? 'Abgleich alle 3 s' : 'Live'} + + )} + {/* Who else has this note open */} {otherPeople.length > 0 && ( void) | null = null; + onmessage: ((e: { data: string }) => void) | null = null; + onclose: ((e: { wasClean: boolean; code: number }) => void) | null = null; + onerror: ((e: unknown) => void) | null = null; + constructor(public url: string) { + FakeWebSocket.instances.push(this); + } + open() { + this.readyState = FakeWebSocket.OPEN; + this.onopen?.(); + } + send(data: string) { + this.sent.push(data); + } + close() { + this.readyState = FakeWebSocket.CLOSED; + } +} + +class FakeEventSource { + static instances: FakeEventSource[] = []; + closed = false; + onopen: (() => void) | null = null; + onmessage: ((e: { data: string }) => void) | null = null; + onerror: ((e: unknown) => void) | null = null; + constructor(public url: string) { + FakeEventSource.instances.push(this); + } + open() { + this.onopen?.(); + } + emit(event: string, payload: unknown = {}) { + this.onmessage?.({ data: JSON.stringify({ event, payload }) }); + } + close() { + this.closed = true; + } +} + +async function loadApi() { + vi.resetModules(); // fresh module state (transport flags) for every test + return import('../api'); +} + +describe('realtime fallback', () => { + beforeEach(() => { + vi.useFakeTimers(); + FakeWebSocket.instances = []; + FakeEventSource.instances = []; + vi.stubGlobal('WebSocket', FakeWebSocket); + vi.stubGlobal('EventSource', FakeEventSource); + sessionStorage.clear(); + localStorage.setItem('mindsift_jwt_token', 'test-token'); + vi.stubGlobal('fetch', vi.fn(async () => new Response('{}', { status: 200 }))); + }); + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + it('a WebSocket that stays silent is replaced by SSE', async () => { + const api = await loadApi(); + const states: string[] = []; + api.onRealtimeStatus((connected, transport) => states.push(`${connected}:${transport}`)); + api.connectWebSocket(); + FakeWebSocket.instances[0].open(); + expect(api.getRealtimeTransport()).toBe('ws'); + + await vi.advanceTimersByTimeAsync(15_000); + expect(FakeWebSocket.instances[0].sent).toContain(JSON.stringify({ event: 'PING' })); + await vi.advanceTimersByTimeAsync(40_000); // no message at all, not even PONG + expect(FakeEventSource.instances).toHaveLength(1); + FakeEventSource.instances[0].open(); + FakeEventSource.instances[0].emit('SSE_CONNECTED', { sseId: 'sse_1' }); + expect(api.getRealtimeTransport()).toBe('sse'); + api.disconnectWebSocket(); + }); + + it('a WebSocket answering PONG stays in use', async () => { + const api = await loadApi(); + api.connectWebSocket(); + const ws = FakeWebSocket.instances[0]; + ws.open(); + for (let i = 0; i < 6; i++) { + await vi.advanceTimersByTimeAsync(15_000); + ws.onmessage?.({ data: JSON.stringify({ event: 'PONG' }) }); + } + expect(FakeEventSource.instances).toHaveLength(0); + expect(api.getRealtimeTransport()).toBe('ws'); + api.disconnectWebSocket(); + }); + + it('SSE only counts as connected once the first event arrives', async () => { + sessionStorage.setItem('mindsift_ws_blocked', 'true'); + const api = await loadApi(); + api.connectWebSocket(); + const es = FakeEventSource.instances[0]; + es.open(); + expect(api.isRealtimeConnected()).toBe(false); + es.emit('SSE_CONNECTED', { sseId: 'sse_1' }); + expect(api.isRealtimeConnected()).toBe(true); + await vi.advanceTimersByTimeAsync(10_000); + expect(api.isRealtimeConnected()).toBe(true); + api.disconnectWebSocket(); + }); + + it('a buffering proxy (stream opens, no events) switches SSE off for this tab', async () => { + sessionStorage.setItem('mindsift_ws_blocked', 'true'); + const api = await loadApi(); + api.connectWebSocket(); + const es = FakeEventSource.instances[0]; + es.open(); + await vi.advanceTimersByTimeAsync(5_500); + expect(es.closed).toBe(true); + expect(api.isRealtimeConnected()).toBe(false); + expect(sessionStorage.getItem('mindsift_sse_buffered')).toBe('true'); + await vi.advanceTimersByTimeAsync(60_000); + expect(FakeEventSource.instances).toHaveLength(1); // no reconnect loop: note sync uses long polling + api.disconnectWebSocket(); + }); +}); diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index c213f18..bb37b3a 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1134,6 +1134,16 @@ export function onRealtimeStatus(callback: (connected: boolean, transport: Realt } let isWsBlockedByProxy = typeof window !== 'undefined' && sessionStorage.getItem('mindsift_ws_blocked') === 'true'; +// Some proxies buffer event streams: the connection opens, but events only arrive much later (or never). +// Then SSE is useless and the note sync engines fall back to long polling. +let isSseBufferedByProxy = typeof window !== 'undefined' && sessionStorage.getItem('mindsift_sse_buffered') === 'true'; +let sseFirstEventWatchdog: any = null; +const SSE_FIRST_EVENT_TIMEOUT_MS = 5000; +// Some proxies accept the WebSocket upgrade but then forward nothing. An app-level ping detects that. +let wsHeartbeatTimer: any = null; +let wsLastMessageAt = 0; +const WS_PING_INTERVAL_MS = 15000; +const WS_DEAD_AFTER_MS = 35000; let sseReconnectTimer: any = null; let sseReconnectDelay = 2000; const globalSubscribedRooms = new Set(); @@ -1153,6 +1163,7 @@ function notifyRealtimeStatusChange(connected: boolean, transport: RealtimeTrans function switchToSSE(reason: string) { console.warn(`[MindSift Realtime] ${reason} -> Schalte sofort und dauerhaft auf Server-Sent Events (SSE) um.`); + clearInterval(wsHeartbeatTimer); isWsBlockedByProxy = true; try { sessionStorage.setItem('mindsift_ws_blocked', 'true'); @@ -1177,6 +1188,12 @@ export function connectSSE(rooms?: string[]): void { clearTimeout(sseReconnectTimer); clearTimeout(wsReconnectTimer); clearTimeout(wsConnectionWatchdog); + clearTimeout(sseFirstEventWatchdog); + + if (isSseBufferedByProxy) { + notifyRealtimeStatusChange(false, 'none'); + return; + } if (activeGlobalSSE) { try { activeGlobalSSE.close(); } catch {} @@ -1190,18 +1207,28 @@ export function connectSSE(rooms?: string[]): void { es.onopen = () => { sseReconnectDelay = 2000; - notifyRealtimeStatusChange(true, 'sse'); - if (activeSSEId && globalSubscribedRooms.size > 0) { - globalSubscribedRooms.forEach((room) => { - api.joinSSERoom(activeSSEId!, room, 'join').catch(() => {}); - }); - } + // Connected only counts once the first event really arrives (see SSE_CONNECTED below) + clearTimeout(sseFirstEventWatchdog); + sseFirstEventWatchdog = setTimeout(() => { + console.warn('[MindSift Realtime] SSE liefert keine Ereignisse (Proxy puffert) -> Long-Polling'); + isSseBufferedByProxy = true; + try { + sessionStorage.setItem('mindsift_sse_buffered', 'true'); + } catch {} + try { + es.close(); + } catch {} + if (activeGlobalSSE === es) activeGlobalSSE = null; + notifyRealtimeStatusChange(false, 'none'); + }, SSE_FIRST_EVENT_TIMEOUT_MS); }; es.onmessage = (event) => { try { const msg = JSON.parse(event.data) as WSMessage; if (msg.event === 'SSE_CONNECTED') { + clearTimeout(sseFirstEventWatchdog); + notifyRealtimeStatusChange(true, 'sse'); activeSSEId = msg.payload?.sseId || null; if (activeSSEId && globalSubscribedRooms.size > 0) { globalSubscribedRooms.forEach((room) => { @@ -1223,6 +1250,7 @@ export function connectSSE(rooms?: string[]): void { }; es.onerror = (e) => { + clearTimeout(sseFirstEventWatchdog); console.warn('[MindSift Realtime] SSE Verbindungsfehler:', e); if (activeTransport === 'sse') { notifyRealtimeStatusChange(false, 'none'); @@ -1279,6 +1307,17 @@ export function connectWebSocket(): void { ws.onopen = () => { clearTimeout(wsConnectionWatchdog); + wsLastMessageAt = Date.now(); + clearInterval(wsHeartbeatTimer); + wsHeartbeatTimer = setInterval(() => { + if (Date.now() - wsLastMessageAt > WS_DEAD_AFTER_MS) { + switchToSSE('WebSocket liefert keine Antworten mehr (Proxy leitet nichts weiter)'); + return; + } + try { + ws.send(JSON.stringify({ event: 'PING' })); + } catch {} + }, WS_PING_INTERVAL_MS); if (activeGlobalSSE) { try { activeGlobalSSE.close(); } catch {} activeGlobalSSE = null; @@ -1295,8 +1334,10 @@ export function connectWebSocket(): void { }; ws.onmessage = (event) => { + wsLastMessageAt = Date.now(); try { const msg = JSON.parse(event.data) as WSMessage; + if (msg.event === 'PONG') return; wsListeners.forEach((fn) => { try { fn(msg); @@ -1312,6 +1353,7 @@ export function connectWebSocket(): void { ws.onclose = (evt: CloseEvent) => { clearTimeout(wsConnectionWatchdog); + clearInterval(wsHeartbeatTimer); activeGlobalWS = null; // In corporate environments, blocked WebSockets close abruptly without clean handshake @@ -1368,6 +1410,8 @@ if (typeof window !== 'undefined') { } export function disconnectWebSocket(): void { + clearInterval(wsHeartbeatTimer); + clearTimeout(sseFirstEventWatchdog); manualDisconnect = true; clearTimeout(wsReconnectTimer); clearTimeout(wsConnectionWatchdog); diff --git a/frontend/src/lib/docSync/__tests__/engine.test.ts b/frontend/src/lib/docSync/__tests__/engine.test.ts index c3c44de..e9a612f 100644 --- a/frontend/src/lib/docSync/__tests__/engine.test.ts +++ b/frontend/src/lib/docSync/__tests__/engine.test.ts @@ -4,6 +4,7 @@ import type { DocumentBlock } from '../../types'; import { DocSyncEngine, formBatch } from '../engine'; import { applyOp, cloneDoc, Doc, emptyDoc, Op, Result } from '../model'; import { IndexedDBStorage, MemoryStorage } from '../persist'; +import { PushChannel, PushKind } from '../push'; import { LogEntry, SnapshotResponse, SyncRequest, SyncResponse, Transport, TransportError } from '../transport'; /** In-memory stand-in for the Go server: same operation rules, log and idempotency. */ @@ -13,6 +14,12 @@ class FakeServer { results = new Map(); roles = new Map(); // clientId -> role applyCount = new Map(); + private waiters: (() => void)[] = []; + + /** Resolves on the next change (used by the long-poll stand-in). */ + nextChange(): Promise { + return new Promise((r) => this.waiters.push(r)); + } constructor(initial: { id: string; content: string }[] = [{ id: 'b1', content: 'Start' }]) { let key = 'a0'; @@ -30,6 +37,11 @@ class FakeServer { for (const e of effects) this.log.push({ seq: this.log.length + 1, userId: clientId, clientId, opId: op.opId, effect: e }); if (effects.length > 0) result.seq = this.log.length; this.results.set(op.opId, result); + if (effects.length > 0) { + const w = this.waiters; + this.waiters = []; + w.forEach((fn) => fn()); + } return result; } @@ -66,6 +78,8 @@ class FakeServer { /** Network between one client and the fake server, with switchable failures. */ class FakeNet implements Transport { mode: 'ok' | 'down' | 'lose-response' = 'ok'; + longPollDown = false; + changesCalls = 0; duringRequest: (() => void) | null = null; // runs while the request is "on the wire" constructor(private server: FakeServer, private role = 'owner') {} @@ -84,6 +98,40 @@ class FakeNet implements Transport { return res; } async leave() {} + async changes(_docId: string, since: number, waitSec: number, signal?: AbortSignal) { + this.changesCalls++; + if (this.mode === 'down' || this.longPollDown) throw new TransportError('blocked', 0); + if (this.server.log.length === since) { + await Promise.race([ + this.server.nextChange(), + new Promise((r) => setTimeout(r, waitSec * 1000)), + new Promise((_, reject) => signal?.addEventListener('abort', () => reject(new Error('aborted')))), + ]); + } + return { headSeq: this.server.log.length, changed: this.server.log.length !== since }; + } +} + +/** Push channel controlled by the test. */ +class FakePush implements PushChannel { + kind: PushKind = 'none'; + private poke: (() => void) | null = null; + private status: ((k: PushKind) => void) | null = null; + subscribe(_docId: string, onPoke: () => void, onStatus: (k: PushKind) => void) { + this.poke = onPoke; + this.status = onStatus; + onStatus(this.kind); + return () => { + this.poke = null; + }; + } + setKind(k: PushKind) { + this.kind = k; + this.status?.(k); + } + notify() { + this.poke?.(); + } } function flatOf(engine: DocSyncEngine): DocumentBlock[] { @@ -254,27 +302,23 @@ describe('DocSyncEngine', () => { expect(serverContents(server)).toEqual(['Start']); }); - it('polls every 3 s right after reopening a shared note from local storage', async () => { + it('changes of others arrive within seconds right after reopening a shared note from local storage', async () => { vi.useFakeTimers(); try { const server = new FakeServer(); const net = new FakeNet(server); - let calls = 0; - const sync = net.sync.bind(net); - net.sync = async (...args: Parameters) => { - calls++; - return sync(...args); - }; + net.longPollDown = true; // worst case: only plain polling works // Opened before: the note is reopened from local storage (like after a reload) const storage = new MemoryStorage(); const first = new DocSyncEngine('doc', { transport: net, storage, manual: true }); await first.start(); first.stop(); - const engine = new DocSyncEngine('doc', { transport: net, storage }); + const engine = new DocSyncEngine('doc', { transport: net, storage, push: new FakePush() }); await engine.start(); - const afterStart = calls; + await vi.advanceTimersByTimeAsync(5_000); // long polling gives up after three attempts + server.apply({ opId: 'remote', kind: 'set_content', blockId: 'b1', baseVersion: 1, content: 'Von anderen' }, 'other'); await vi.advanceTimersByTimeAsync(3_500); - expect(calls).toBeGreaterThan(afterStart); // not only after 30 s + expect(contents(engine)).toEqual(['Von anderen']); // not only after 30 s engine.stop(); } finally { vi.useRealTimers(); @@ -303,3 +347,141 @@ describe('DocSyncEngine', () => { }); }); + +describe('how changes of others arrive', () => { + async function live(push: FakePush, net: FakeNet) { + const engine = new DocSyncEngine('doc', { transport: net, storage: new MemoryStorage(), push }); + await engine.start(); + await vi.advanceTimersByTimeAsync(100); // let the start-up round trips settle + return engine; + } + const otherEdit = (server: FakeServer, text: string) => + server.apply({ opId: `other-${text}`, kind: 'set_content', blockId: 'b1', baseVersion: server.doc.blocks.get('b1')!.version, content: text }, 'other'); + + it('push: syncs on a notification and otherwise only checks every 30 s', async () => { + vi.useFakeTimers(); + try { + const server = new FakeServer(); + const net = new FakeNet(server); + const push = new FakePush(); + push.kind = 'ws'; + const engine = await live(push, net); + expect(engine.getState().channel).toBe('ws'); + otherEdit(server, 'Von anderen'); + await vi.advanceTimersByTimeAsync(20_000); + expect(contents(engine)).toEqual(['Start']); // no notification, no poll yet + push.notify(); + await vi.advanceTimersByTimeAsync(100); + expect(contents(engine)).toEqual(['Von anderen']); + expect(net.changesCalls).toBe(0); + engine.stop(); + } finally { + vi.useRealTimers(); + } + }); + + it('push: renews leases every 7 s while a block is held', async () => { + vi.useFakeTimers(); + try { + const server = new FakeServer(); + const net = new FakeNet(server); + const push = new FakePush(); + push.kind = 'sse'; + const engine = await live(push, net); + let syncs = 0; + const orig = net.sync.bind(net); + net.sync = async (...a: Parameters) => { + syncs++; + return orig(...a); + }; + engine.focusBlock('b1'); + await vi.advanceTimersByTimeAsync(15_000); + expect(syncs).toBeGreaterThanOrEqual(2); + engine.stop(); + } finally { + vi.useRealTimers(); + } + }); + + it('without push: long polling delivers changes right away', async () => { + vi.useFakeTimers(); + try { + const server = new FakeServer(); + const net = new FakeNet(server); + const engine = await live(new FakePush(), net); + await vi.advanceTimersByTimeAsync(10); + expect(engine.getState().channel).toBe('longpoll'); + otherEdit(server, 'Per Long-Polling'); + await vi.advanceTimersByTimeAsync(200); + expect(contents(engine)).toEqual(['Per Long-Polling']); + engine.stop(); + } finally { + vi.useRealTimers(); + } + }); + + it('falls back to polling every 3 s when long polling is blocked too', async () => { + vi.useFakeTimers(); + try { + const server = new FakeServer(); + const net = new FakeNet(server); + net.longPollDown = true; + const engine = await live(new FakePush(), net); + await vi.advanceTimersByTimeAsync(5_000); // three failed attempts with back-off + expect(engine.getState().channel).toBe('interval'); + otherEdit(server, 'Per Abfrage'); + await vi.advanceTimersByTimeAsync(3_500); + expect(contents(engine)).toEqual(['Per Abfrage']); + engine.stop(); + } finally { + vi.useRealTimers(); + } + }); + + it('push: requests the lease right after focusing a block', async () => { + vi.useFakeTimers(); + try { + const server = new FakeServer(); + const net = new FakeNet(server); + const push = new FakePush(); + push.kind = 'ws'; + const engine = await live(push, net); + const holds: string[][] = []; + const orig = net.sync.bind(net); + net.sync = async (id: string, body: SyncRequest, o?: { keepalive?: boolean }) => { + holds.push(body.leases?.hold || []); + return orig(id, body, o); + }; + engine.focusBlock('b1'); + await vi.advanceTimersByTimeAsync(1_000); + expect(holds.some((h) => h.includes('b1'))).toBe(true); + engine.stop(); + } finally { + vi.useRealTimers(); + } + }); + + it('switches between push and long polling when the connection changes', async () => { + vi.useFakeTimers(); + try { + const server = new FakeServer(); + const net = new FakeNet(server); + const push = new FakePush(); + push.kind = 'ws'; + const engine = await live(push, net); + push.setKind('none'); + await vi.advanceTimersByTimeAsync(10); + expect(engine.getState().channel).toBe('longpoll'); + push.setKind('ws'); + await vi.advanceTimersByTimeAsync(10); + expect(engine.getState().channel).toBe('ws'); + otherEdit(server, 'Wieder per Push'); + push.notify(); + await vi.advanceTimersByTimeAsync(100); + expect(contents(engine)).toEqual(['Wieder per Push']); + engine.stop(); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/frontend/src/lib/docSync/__tests__/undo.test.ts b/frontend/src/lib/docSync/__tests__/undo.test.ts index 3bf9bd2..557f0e4 100644 --- a/frontend/src/lib/docSync/__tests__/undo.test.ts +++ b/frontend/src/lib/docSync/__tests__/undo.test.ts @@ -51,6 +51,7 @@ class Server { }; }, leave: async () => {}, + changes: async () => ({ headSeq: this.log.length, changed: false }), }; } } diff --git a/frontend/src/lib/docSync/engine.ts b/frontend/src/lib/docSync/engine.ts index 6cb4cca..e3168e1 100644 --- a/frontend/src/lib/docSync/engine.ts +++ b/frontend/src/lib/docSync/engine.ts @@ -8,9 +8,13 @@ import { applyEffect, applyOp, cloneDoc, Doc, docFromSnapshot, emptyDoc, Op, Res import { defaultStorage, SyncStorage } from './persist'; import { httpTransport, LeaseInfo, Presence, Transport, TransportError } from './transport'; import { buildStep, failedChecks, touchedBlocks, UndoStep } from './undo'; +import { PushChannel, PushKind, realtimePush } from './push'; export type EngineStatus = 'loading' | 'saved' | 'saving' | 'offline' | 'error'; +/** How changes of others reach this client: push (WebSocket/SSE), long polling or plain polling. */ +export type SyncChannel = 'ws' | 'sse' | 'longpoll' | 'interval'; + export interface EngineState { ready: boolean; status: EngineStatus; @@ -26,6 +30,7 @@ export interface EngineState { conflicts: Result[]; // conflicts of own operations since start canUndo: boolean; canRedo: boolean; + channel: SyncChannel; } export interface ActionResult { @@ -37,15 +42,21 @@ export interface EngineOptions { transport?: Transport; storage?: SyncStorage; now?: () => number; - /** Disables timers and window listeners (tests drive the engine with syncNow()). */ + /** Disables timers, window listeners, push and long polling (tests drive the engine with syncNow()). */ manual?: boolean; + /** Push notifications; defaults to the app's realtime connection (not used in manual mode). */ + push?: PushChannel; } const LEASE_GRACE_MS = 20_000; // keep a left block for this long const IDLE_RELEASE_MS = 60_000; // release even the focused block after this long without input const SYNC_DEBOUNCE_MS = 300; -const POLL_COLLAB_MS = 3_000; -const POLL_SOLO_MS = 30_000; +const POLL_COLLAB_MS = 3_000; // shared note without push or long polling +const POLL_SOLO_MS = 30_000; // private note, opened once +const HEARTBEAT_MS = 7_000; // renews block leases while changes arrive by push/long polling +const SAFETY_POLL_MS = 30_000; // in case a push notification got lost +const LONG_POLL_WAIT_S = 25; // below typical proxy timeouts (nginx 60 s, Cloudflare 100 s) +const LONG_POLL_MAX_FAILURES = 3; const BACKOFF_MAX_MS = 30_000; const UNDO_LIMIT = 200; const UNDO_GROUP_MS = 1_000; // keystrokes in the same block are one undo step until a pause this long @@ -131,6 +142,14 @@ export class DocSyncEngine { private undoStack: UndoStep[] = []; private redoStack: UndoStep[] = []; + private push: PushChannel | null; + private pushKind: PushKind = 'none'; + private unsubscribePush: (() => void) | null = null; + private pokeTimer: ReturnType | null = null; + private longPollRunning = false; + private longPollBroken = false; + private longPollAbort: AbortController | null = null; + private syncing = false; private syncAgain = false; private stopped = false; @@ -144,6 +163,7 @@ export class DocSyncEngine { this.storage = opts.storage || defaultStorage(); this.now = opts.now || (() => Date.now()); this.manual = !!opts.manual; + this.push = opts.push || (this.manual ? null : realtimePush); } // --- Public API --- @@ -165,6 +185,13 @@ export class DocSyncEngine { // Storage unavailable: continue with the server state only } if (!this.manual) this.attachWindowListeners(); + if (this.push) { + this.unsubscribePush = this.push.subscribe( + this.docId, + () => this.onPoke(), + (kind) => this.onPushStatus(kind) + ); + } if (!this.ready) { try { await this.loadSnapshot(); @@ -185,6 +212,10 @@ export class DocSyncEngine { this.flushOnExit(); this.stopped = true; if (this.timer) clearTimeout(this.timer); + if (this.pokeTimer) clearTimeout(this.pokeTimer); + this.longPollAbort?.abort(); + this.unsubscribePush?.(); + this.unsubscribePush = null; this.cleanup.forEach((fn) => fn()); this.cleanup = []; this.listeners.clear(); @@ -212,6 +243,7 @@ export class DocSyncEngine { conflicts: this.conflicts, canUndo: this.undoStack.length > 0, canRedo: this.redoStack.length > 0, + channel: this.channel(), }; } @@ -314,9 +346,15 @@ export class DocSyncEngine { /** Tells the engine which block the cursor is in (null = none); used for block leases. */ focusBlock(blockId: string | null): void { const t = this.now(); + const changed = blockId !== this.focused; if (this.focused) this.activity.set(this.focused, t); this.focused = blockId; if (blockId) this.activity.set(blockId, t); + // Announce the lease right away: with push the next regular sync may be up to 30 s away + if (changed && blockId && this.canEdit && this.collaborative()) { + if (this.syncing) this.syncAgain = true; + else this.schedule(SYNC_DEBOUNCE_MS); + } } /** Runs one sync round trip now (or right after the running one). */ @@ -331,19 +369,26 @@ export class DocSyncEngine { this.timer = null; } this.syncing = true; - let delay: number; + let failed = false; try { await this.roundTrip(); this.backoff = 0; - // Decided only now: the response tells whether the note is shared (3 s) or private (30 s) - delay = this.syncAgain || this.pending.length > 0 ? 0 : this.pollDelay(); } catch (err) { this.handleError(err); - delay = this.status === 'error' ? this.pollDelay() : this.nextBackoff(); + failed = true; } finally { this.syncing = false; - this.syncAgain = false; this.recompute(); + if (!failed) this.ensureLongPoll(); + // Decided only now: the response tells whether the note is shared and how changes arrive + const delay = failed + ? this.status === 'error' + ? this.pollDelay() + : this.nextBackoff() + : this.syncAgain || this.pending.length > 0 + ? 0 + : this.pollDelay(); + this.syncAgain = false; this.emit(); this.schedule(delay); } @@ -548,8 +593,76 @@ export class DocSyncEngine { this.errorMessage = err instanceof Error ? err.message : String(err); } + private channel(): SyncChannel { + if (this.pushKind !== 'none') return this.pushKind; + if (this.longPollRunning && !this.longPollBroken) return 'longpoll'; + return 'interval'; + } + private pollDelay(): number { - return this.collaborative() ? POLL_COLLAB_MS : POLL_SOLO_MS; + if (!this.collaborative()) return POLL_SOLO_MS; + if (this.channel() === 'interval') return POLL_COLLAB_MS; + return this.canEdit && this.holds().length > 0 ? HEARTBEAT_MS : SAFETY_POLL_MS; + } + + private onPoke(): void { + if (this.pokeTimer || this.stopped) return; + // Several notifications in a row (e.g. typing of others) lead to one sync + this.pokeTimer = setTimeout(() => { + this.pokeTimer = null; + void this.syncNow(); + }, 50); + } + + private onPushStatus(kind: PushKind): void { + const wasLive = this.pushKind !== 'none'; + this.pushKind = kind; + if (kind !== 'none') { + this.longPollAbort?.abort(); + if (!wasLive) void this.syncNow(); // catch up on what was missed while disconnected + } else { + this.ensureLongPoll(); + if (!this.syncing) this.schedule(this.pollDelay()); + } + this.emit(); + } + + /** Starts long polling when there is no push channel (e.g. a proxy blocks WebSocket and SSE). */ + private ensureLongPoll(): void { + if (this.manual || this.stopped || !this.ready || this.longPollRunning || this.longPollBroken) return; + if (this.pushKind !== 'none' || !this.collaborative() || this.status === 'error') return; + void this.longPollLoop(); + } + + private async longPollLoop(): Promise { + this.longPollRunning = true; + this.emit(); + let failures = 0; + try { + while (!this.stopped && this.pushKind === 'none' && this.collaborative() && this.status !== 'error') { + const ctrl = new AbortController(); + this.longPollAbort = ctrl; + try { + const res = await this.transport.changes(this.docId, this.lastSeq, LONG_POLL_WAIT_S, ctrl.signal); + failures = 0; + if (res.changed || res.headSeq !== this.lastSeq) await this.syncNow(); + } catch { + if (ctrl.signal.aborted || this.stopped) break; + failures++; + if (failures >= LONG_POLL_MAX_FAILURES) { + // Long polling does not get through either: plain polling every 3 s remains + this.longPollBroken = true; + break; + } + await new Promise((r) => setTimeout(r, 1000 * failures)); + } + } + } finally { + this.longPollRunning = false; + this.longPollAbort = null; + this.emit(); + if (!this.syncing && !this.stopped) this.schedule(this.pollDelay()); + } } private nextBackoff(): number { @@ -580,7 +693,10 @@ export class DocSyncEngine { } private attachWindowListeners(): void { - const onOnline = () => void this.syncNow(); + const onOnline = () => { + this.longPollBroken = false; // the network changed: long polling may work again + void this.syncNow(); + }; const onVisibility = () => { if (document.visibilityState === 'hidden') this.flushOnExit(); else void this.syncNow(); diff --git a/frontend/src/lib/docSync/push.ts b/frontend/src/lib/docSync/push.ts new file mode 100644 index 0000000..c3dad2d --- /dev/null +++ b/frontend/src/lib/docSync/push.ts @@ -0,0 +1,25 @@ +// Push notifications for note sync: the server sends DOC_SEQ into the note's room whenever the note, its +// leases or its presence change. Delivery uses the app-wide realtime connection (WebSocket, else SSE). +import { createRealtimeConnection, getRealtimeTransport, isRealtimeConnected } from '../api'; + +export type PushKind = 'ws' | 'sse' | 'none'; + +export interface PushChannel { + /** Calls onPoke for changes of docId and onStatus whenever the channel goes up or down. */ + subscribe(docId: string, onPoke: () => void, onStatus: (kind: PushKind) => void): () => void; +} + +export const realtimePush: PushChannel = { + subscribe(docId, onPoke, onStatus) { + const current = (): PushKind => (isRealtimeConnected() ? (getRealtimeTransport() as PushKind) : 'none'); + const conn = createRealtimeConnection({ + docId, + onMessage: (msg: any) => { + if (msg?.event === 'DOC_SEQ' && msg.payload?.docId === docId) onPoke(); + }, + onStatusChange: () => onStatus(current()), + }); + onStatus(current()); + return () => conn.close(); + }, +}; diff --git a/frontend/src/lib/docSync/transport.ts b/frontend/src/lib/docSync/transport.ts index e6de919..127f6ce 100644 --- a/frontend/src/lib/docSync/transport.ts +++ b/frontend/src/lib/docSync/transport.ts @@ -59,20 +59,28 @@ export class TransportError extends Error { } } +export interface ChangesResponse { + headSeq: number; + changed: boolean; +} + export interface Transport { snapshot(docId: string): Promise; + /** Long poll: resolves when the note changes, at the latest after waitSec seconds. */ + changes(docId: string, since: number, waitSec: number, signal?: AbortSignal): Promise; sync(docId: string, body: SyncRequest, opts?: { keepalive?: boolean }): Promise; leave(docId: string, clientId: string): Promise; } const TIMEOUT_MS = 15000; -async function call(path: string, init: RequestInit & { keepalive?: boolean } = {}): Promise { +async function call(path: string, init: RequestInit & { keepalive?: boolean } = {}, timeoutMs = TIMEOUT_MS): Promise { const headers: Record = { 'Content-Type': 'application/json' }; const token = getToken(); if (token) headers.Authorization = `Bearer ${token}`; const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(), TIMEOUT_MS); + const timer = setTimeout(() => controller.abort(), timeoutMs); + init.signal?.addEventListener('abort', () => controller.abort()); let res: Response; try { res = await fetch(path, { ...init, headers, cache: 'no-store', signal: init.keepalive ? undefined : controller.signal }); @@ -99,6 +107,8 @@ async function call(path: string, init: RequestInit & { keepalive?: boolean } export const httpTransport: Transport = { snapshot: (docId) => call(`/api/v1/docs/${encodeURIComponent(docId)}/snapshot`), + changes: (docId, since, waitSec, signal) => + call(`/api/v1/docs/${encodeURIComponent(docId)}/changes?since=${since}&wait=${waitSec}`, { signal }, (waitSec + 10) * 1000), sync: (docId, body, opts) => call(`/api/v1/docs/${encodeURIComponent(docId)}/sync`, { method: 'POST', body: JSON.stringify(body), keepalive: opts?.keepalive }), leave: async (docId, clientId) => { diff --git a/internal/api/realtime_test.go b/internal/api/realtime_test.go index 51520a0..0f70c48 100644 --- a/internal/api/realtime_test.go +++ b/internal/api/realtime_test.go @@ -11,7 +11,9 @@ import ( "testing" "time" + "github.com/gorilla/websocket" "github.com/sandrodach/mindsift/internal/auth" + "github.com/sandrodach/mindsift/internal/collab" "github.com/sandrodach/mindsift/internal/domain" "github.com/sandrodach/mindsift/internal/storage" ) @@ -266,3 +268,71 @@ func TestReorderRejectsForeignTasks(t *testing.T) { t.Fatalf("foreign task was modified: sortOrder=%d completed=%v", after.SortOrder, after.IsCompleted) } } + +func dialWS(t *testing.T, ts *httptest.Server, token string) *websocket.Conn { + t.Helper() + url := "ws" + strings.TrimPrefix(ts.URL, "http") + "/api/v1/ws?token=" + token + conn, _, err := websocket.DefaultDialer.Dial(url, nil) + if err != nil { + t.Fatalf("dial ws: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func readWSEvent(conn *websocket.Conn, event string, timeout time.Duration) *sseMsg { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + _ = conn.SetReadDeadline(deadline) + var m sseMsg + if err := conn.ReadJSON(&m); err != nil { + return nil + } + if m.Event == event { + return &m + } + } + return nil +} + +func TestWebSocketHeartbeat(t *testing.T) { + _, _, ts, _, token := startRealtimeServer(t) + conn := dialWS(t, ts, token) + if err := conn.WriteJSON(map[string]string{"event": "PING"}); err != nil { + t.Fatal(err) + } + if readWSEvent(conn, "PONG", 2*time.Second) == nil { + t.Fatal("no PONG for an app-level PING") + } +} + +func TestDocChangesArePushedToTheNoteRoom(t *testing.T) { + _, store, ts, owner, ownerToken := startRealtimeServer(t) + editor, editorToken := newTestUser(t, store, "push-editor@example.com") + doc, err := store.CreateDocument(owner.ID, "", "Push", "📄", "", []domain.DocumentBlock{{ID: "push_b1", Type: "bullet", Content: "Start"}}, nil, false) + if err != nil { + t.Fatal(err) + } + if _, err := store.CreateDocShare(owner.ID, doc.ID, editor.ID, "write"); err != nil { + t.Fatal(err) + } + + conn := dialWS(t, ts, editorToken) + if err := conn.WriteJSON(map[string]string{"event": "JOIN_ROOM", "room": "doc:" + doc.ID}); err != nil { + t.Fatal(err) + } + time.Sleep(100 * time.Millisecond) // the join is processed asynchronously + + text := "Geändert" + resp := doJSON(t, ts, http.MethodPost, "/api/v1/docs/"+doc.ID+"/sync", ownerToken, map[string]interface{}{ + "clientId": "owner-tab", + "ops": []collab.Op{{OpID: "push-1", Kind: collab.OpSetContent, BlockID: "push_b1", BaseVersion: 1, Content: &text}}, + }) + if resp.StatusCode != http.StatusOK { + t.Fatalf("sync: %d", resp.StatusCode) + } + msg := readWSEvent(conn, "DOC_SEQ", 2*time.Second) + if msg == nil || !strings.Contains(string(msg.Payload), doc.ID) { + t.Fatalf("no DOC_SEQ push for the note: %+v", msg) + } +} diff --git a/internal/ws/hub.go b/internal/ws/hub.go index 5f7ac91..02d11a0 100644 --- a/internal/ws/hub.go +++ b/internal/ws/hub.go @@ -631,7 +631,10 @@ func (c *Client) readPump() { } // Room subscription without requiring docID - if incoming.Event == "JOIN_ROOM" { + if incoming.Event == "PING" { + // App-level heartbeat: lets clients detect proxies that accept the upgrade but forward nothing + c.SendEvent("PONG", nil) + } else if incoming.Event == "JOIN_ROOM" { room := incoming.Room if room == "" { room = docID