From 2b7356d04344cc773d9e8d62c347aa5fd3a5a8eb Mon Sep 17 00:00:00 2001 From: Sandro Date: Mon, 2 Nov 2026 11:50:27 +0100 Subject: [PATCH] feat(realtime): long polling for app-wide events behind proxies When a proxy lets neither WebSocket nor SSE through, the app opens a poll connection: the server buffers its events (an SSEClient without a stream), each GET waits up to 25 s, and room joins work like for SSE. Tasks and lists now update live there too, and note sync takes its notifications from it instead of a second long poll. Idle connections close after two minutes. The profile shows such a connection as limited. Co-Authored-By: Claude Opus 5.5 --- Doku/api/rest.md | 2 + Doku/api/websocket.md | 5 +- Doku/architecture.md | 10 +- Doku/modules/collaboration.md | 2 +- frontend/src/App.tsx | 2 +- frontend/src/lib/__tests__/realtime.test.ts | 32 ++++++ .../lib/__tests__/useConnectionState.test.tsx | 13 ++- frontend/src/lib/api.ts | 79 ++++++++++++- frontend/src/lib/docSync/engine.ts | 2 + frontend/src/lib/docSync/push.ts | 5 +- frontend/src/lib/useConnectionState.ts | 12 +- internal/api/realtime.go | 28 +++++ internal/api/realtime_test.go | 60 ++++++++++ internal/api/server.go | 2 + internal/ws/hub.go | 108 ++++++++++++++++++ internal/ws/hub_test.go | 27 +++++ 16 files changed, 369 insertions(+), 20 deletions(-) diff --git a/Doku/api/rest.md b/Doku/api/rest.md index ffffefb..925a213 100644 --- a/Doku/api/rest.md +++ b/Doku/api/rest.md @@ -28,6 +28,8 @@ Basis: `/api/v1`. Antworten sind JSON; Fehler haben die Form `{"error": "Text"}` |---|---|---| | GET | `/auth/me` | Benutzer und Kennzahlen | | POST | `/events/rooms` | SSE-Raum beitreten/verlassen (`sseId`, `room`, `action`); nur eigene `sseId`, nur Räume mit Lesezugriff | +| POST | `/events/poll` | Long-Polling-Verbindung öffnen (`rooms` optional) → `{pollId}`; nur mit Sitzung | +| GET | `/events/poll/{pollId}?wait=25` | Gepufferte Ereignisse abholen, wartet bis zu `wait` s → `{events}`; `404` = abgelaufen | | POST | `/auth/stream-ticket` | Einmal-Ticket (60 s) für WebSocket/SSE | | POST | `/voice/transcribe` | Nur im Server-Modus der Spracherkennung: `multipart/form-data` mit `file` (Aufnahme, z. B. WebM/Opus oder MP4/AAC) → `{text}`. `404` ohne `STT_URL`, `413` zu groß, `429` Stundenlimit, `503` ausgelastet, `502` Whisper-Fehler (Details im Log) | | POST | `/auth/logout` | Sitzung dieses Geräts beenden | diff --git a/Doku/api/websocket.md b/Doku/api/websocket.md index e629721..1ef6388 100644 --- a/Doku/api/websocket.md +++ b/Doku/api/websocket.md @@ -1,4 +1,4 @@ -# Echtzeit: WebSocket und SSE +# Echtzeit: WebSocket, SSE und Long-Polling Der Server verteilt Änderungen als JSON-Ereignisse an verbundene Clients. Die Ereignisse entstehen **nach** einer REST-Änderung; es gibt keinen Schreibkanal über die Verbindung. Implementierung: [hub.go](../../internal/ws/hub.go), Client: [api.ts](../../frontend/src/lib/api.ts). @@ -10,6 +10,7 @@ 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` | +| Long-Polling | `POST /api/v1/events/poll` → `{pollId}`, dann `GET /api/v1/events/poll/{pollId}?wait=25` | Antwort `{events: [...]}`, sobald Ereignisse da sind, spätestens nach `wait` s leer; `404` = abgelaufen, neu öffnen | Der Client versucht zuerst WebSocket und wechselt in folgenden Fällen für die Sitzung des Tabs auf SSE: @@ -18,7 +19,7 @@ Der Client versucht zuerst WebSocket und wechselt in folgenden Fällen für die Ü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)). +Ohne WebSocket und SSE holt der Client die Ereignisse per **Long-Polling**: Er öffnet eine Poll-Verbindung (Räume als `rooms` gleich mitgeben oder später wie bei SSE über `POST /events/rooms` mit der `pollId` als `sseId`), der Server puffert ihre Ereignisse (bis 256), und jede Abfrage wartet bis zu 25 s. Wird eine Poll-Verbindung 2 Minuten nicht abgefragt, schließt der Server sie. Nur mit Sitzung, nicht mit API-Schlüssel. So kommen auch Aufgaben- und Listenänderungen hinter Proxys live an. Die Notiz-Synchronisation nutzt dann dieselbe Verbindung als Weckruf; ohne sie fällt sie auf ihr eigenes Long-Polling (`GET /docs/{id}/changes`) und zuletzt auf eine Abfrage alle 3 s zurück (siehe [modules/collaboration.md](../modules/collaboration.md)). **Authentifizierung:** Browser holen sich direkt vor dem Verbinden ein Einmal-Ticket (`POST /auth/stream-ticket` mit dem normalen `Authorization`-Header → `{ticket, expiresIn: 60}`) und hängen es als `?ticket=` an; es gilt 60 Sekunden und einmal und wird ungültig, sobald alle Sitzungen beendet werden. Andere Clients dürfen stattdessen den `Authorization`-Header schicken. Sitzungs-Tokens in der URL und API-Keys werden nicht angenommen (`401`). Bei Logout, „Alle Sitzungen beenden“, neuem Passwort und Deaktivierung trennt der Server alle Streams des Kontos; Clients mit gültiger Sitzung verbinden sich mit neuem Ticket neu. Mit gesetzter `BASE_URL` nimmt der WebSocket nur Verbindungen von dieser Adresse oder vom angefragten Host an. diff --git a/Doku/architecture.md b/Doku/architecture.md index e73f14d..d3c101f 100644 --- a/Doku/architecture.md +++ b/Doku/architecture.md @@ -5,10 +5,10 @@ ``` Browser (React SPA, PWA-Manifest) │ REST /api/v1/* Authorization: Bearer - │ WebSocket /api/v1/ws Fallback: SSE /api/v1/events + │ WebSocket /api/v1/ws Fallback: SSE /api/v1/events, dann Long-Polling /api/v1/events/poll ▼ Go-Server (chi) ── internal/api Handler, Middleware, Alarm-Ticker - │ internal/ws Hub (Räume, Benutzer-Kanäle, SSE) + │ internal/ws Hub (Räume, Benutzer-Kanäle, SSE, Long-Polling) │ internal/auth JWT, bcrypt, API-Key-Hash ▼ internal/config SQLite (modernc.org/sqlite, WAL) ── internal/storage @@ -30,7 +30,7 @@ Das Frontend liegt per `go:embed` im Binary ([cmd/server/main.go](../cmd/server/ | Pfad | Inhalt | |---|---| | `cmd/server/` | `main.go`, eingebettetes `dist/` | -| `internal/api/` | `server.go` (Server, Router mit allen Routen, Middleware, JSON-Helfer); Handler nach Bereich in `handlers_*.go` (`auth`, `account`, `thoughts`, `documents`, `folders`, `lists`, `tasklists`, `teams`, `admin`, `integrations`), `realtime.go` (WebSocket/SSE), `alarms.go` (Ticker und Versand), `docsync.go`, `mail.go`, `netguard.go`, `password_reset.go` | +| `internal/api/` | `server.go` (Server, Router mit allen Routen, Middleware, JSON-Helfer); Handler nach Bereich in `handlers_*.go` (`auth`, `account`, `thoughts`, `documents`, `folders`, `lists`, `tasklists`, `teams`, `admin`, `integrations`), `realtime.go` (WebSocket/SSE/Long-Polling), `alarms.go` (Ticker und Versand), `docsync.go`, `mail.go`, `netguard.go`, `password_reset.go` | | `internal/storage/` | `sqlite.go` (Store, Schema, Migrationen); Datenlogik nach Bereich: `users.go`, `apikeys.go`, `thoughts.go` (inkl. Alarme), `tasklists.go`, `documents.go`, `folders.go`, `tags.go`, `lists.go`, `teams.go`, `system.go` (Einstellungen, Statistik, Export), dazu `visibility.go`, `docsync.go`, `search.go`, `trash.go`, `backlinks.go`, `recurrence.go` u. a. | | `internal/domain/` | Strukturen (`models.go`), ID-Erzeugung, Tag-Extraktion | | `internal/ws/` | `hub.go` | @@ -38,7 +38,7 @@ Das Frontend liegt per `go:embed` im Binary ([cmd/server/main.go](../cmd/server/ | `internal/auth/` | Tokens und Passwörter | | `internal/config/` | Umgebungsvariablen | | `frontend/src/features/` | `triage`, `workspace`, `lists`, `teams`, `settings`, `capture`, `auth`, `help` | -| `frontend/src/lib/` | `api.ts` (REST-Client, Transport-Umschaltung WS/SSE), `types.ts`, `markdown.tsx`, `theme.ts` | +| `frontend/src/lib/` | `api.ts` (REST-Client, Transport-Umschaltung WS/SSE/Long-Polling), `types.ts`, `markdown.tsx`, `theme.ts` | | `frontend/src/lib/docSync/` | Notiz-Synchronisation im Browser: Portierung von `internal/collab` (`model.ts`, `orderKey.ts`), Vergleich Editor-Liste → Operationen (`diff.ts`), `engine.ts`, IndexedDB (`persist.ts`), HTTP (`transport.ts`) | | `e2e/` | Browser-Tests mit Playwright gegen einen echten Server (`e2e/run.sh`) | | `frontend/src/themes/` | CSS pro Theme | @@ -173,7 +173,7 @@ Die Tests laufen beim Image-Build (Frontend: `npm run test:run`, Backend: `go te - **Neue Version:** Jeder Build legt `version.json` (Version und Build-Kennung) neben `index.html`, nie gecacht. Die App vergleicht sie mit ihrer eigenen Kennung beim Zurückkehren in Tab oder App, alle 15 Minuten und nach einem Neustart des Servers (Echtzeit-Verbindung wieder da) und zeigt dann „Neue Version verfügbar“ mit „Jetzt aktualisieren“ ([lib/useUpdateCheck.ts](../frontend/src/lib/useUpdateCheck.ts)). - **Zustand pro Tab:** Der offene Bereich (Triage, Workspace, Aufgaben …), die offene Notiz und die gewählte Liste stehen in `sessionStorage` ([lib/viewSession.ts](../frontend/src/lib/viewSession.ts)) und bleiben beim Neuladen erhalten. - **Hinweis:** Ein Banner über der unteren Navigation zeigt „Offline“ und die Zahl wartender Änderungen bzw. „Synchronisiere…“. -- **Verbindungsstatus im Kopf** ([lib/useConnectionState.ts](../frontend/src/lib/useConnectionState.ts)): Maßgeblich ist die app-weite Echtzeit-Verbindung (WebSocket/SSE), nicht `navigator.onLine`; so fällt auch ein ausgefallener Server auf. Fehlt sie länger als 4 s (Start, kurze Reconnects überbrückt), fragt die App `/api/v1/health` (danach alle 15 s): Antwortet der Server, ist die Verbindung **eingeschränkt** (z. B. Proxy, der SSE puffert; Notizen laufen dann über Long Polling, andere Änderungen erst beim Zurückkehren in den Tab oder Neuladen), sonst **offline**. Ohne Gerätenetz sofort offline. Anzeige als Punkt grün/gelb/rot: Desktop mit Text unter dem Namen, Handy am Profilbild. +- **Verbindungsstatus im Kopf** ([lib/useConnectionState.ts](../frontend/src/lib/useConnectionState.ts)): Maßgeblich ist die app-weite Echtzeit-Verbindung, nicht `navigator.onLine`; so fällt auch ein ausgefallener Server auf. WebSocket oder SSE: **online**. Nur Long-Polling (Proxy, der SSE puffert): **eingeschränkt**, Änderungen kommen leicht verzögert. Fehlt jede Verbindung länger als 4 s (Start, kurze Reconnects überbrückt), fragt die App `/api/v1/health` (danach alle 15 s): Antwortet der Server, gilt ebenfalls **eingeschränkt**, sonst **offline**. Ohne Gerätenetz sofort offline. Anzeige als Punkt grün/gelb/rot: Desktop mit Text unter dem Namen, Handy am Profilbild. ## App-Icons, PWA & Social-Preview (Open Graph) diff --git a/Doku/modules/collaboration.md b/Doku/modules/collaboration.md index 3ba379d..8f6e79d 100644 --- a/Doku/modules/collaboration.md +++ b/Doku/modules/collaboration.md @@ -112,7 +112,7 @@ Daten fließen nur über `/sync` und `/snapshot`. Push-Kanäle melden nur „Not |---|---|---| | 1. WebSocket | Verbindungsfehler oder 35 s ohne `PONG` auf das `PING` alle 15 s (WebSocket angenommen, aber nichts kommt durch) | SSE | | 2. SSE | Erstes Ereignis kommt nicht innerhalb von 5 s (Proxy puffert); gilt dann für den ganzen Tab | Long-Polling | -| 3. Long-Polling (`/changes`, 25 s) | 3 Fehlschläge hintereinander | Intervall | +| 3. Long-Polling: app-weit (`/events/poll`, `DOC_SEQ` wie bei SSE), sonst eigenes (`/changes`, 25 s) | 3 Fehlschläge hintereinander | Intervall | | 4. Intervall | – | Abfrage alle 3 s; bei `online` wird Long-Polling erneut versucht | Zusätzlich, solange Stufe 1–3 läuft: Abgleich alle 7 s, wenn die Engine Sperren hält (Verlängerung), sonst alle 30 s zur Sicherheit. Der Editor zeigt den Weg an („Live“, „Live (Long-Polling)“, „Abgleich alle 3 s“). Gemessen im Browser-Test: 0,2–0,3 s über WebSocket, SSE und Long-Polling, rund 3 s im Intervall. diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 9fffed8..6e8c3c8 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -89,7 +89,7 @@ const CONNECTION_STATUS = { limited: { label: 'eingeschränkt', dot: 'bg-status-warn', - title: 'Eingeschränkt – der Server ist erreichbar, aber keine Live-Verbindung (z. B. Firmen-Proxy); Notizen gleichen sich langsamer ab, andere Änderungen erscheinen beim Zurückkehren in den Tab oder nach dem Neuladen', + title: 'Eingeschränkt – keine direkte Live-Verbindung (z. B. Firmen-Proxy); Änderungen anderer kommen über Long Polling, etwas verzögert', }, offline: { label: 'offline', dot: 'bg-status-err', title: 'Offline – keine Verbindung zum Server; Änderungen werden später synchronisiert' }, } as const; diff --git a/frontend/src/lib/__tests__/realtime.test.ts b/frontend/src/lib/__tests__/realtime.test.ts index 9f0f9bd..d8f52ce 100644 --- a/frontend/src/lib/__tests__/realtime.test.ts +++ b/frontend/src/lib/__tests__/realtime.test.ts @@ -137,4 +137,36 @@ describe('realtime fallback', () => { expect(FakeEventSource.instances).toHaveLength(1); // no reconnect loop: note sync uses long polling api.disconnectWebSocket(); }); + + it('behind such a proxy, events come by long polling', async () => { + sessionStorage.setItem('mindsift_ws_blocked', 'true'); + sessionStorage.setItem('mindsift_sse_buffered', 'true'); + let polls = 0; + vi.stubGlobal( + 'fetch', + vi.fn(async (url: string, init?: RequestInit) => { + if (url === '/api/v1/events/poll') return new Response('{"pollId":"poll_1"}', { status: 200 }); + if (url.startsWith('/api/v1/events/poll/poll_1')) { + polls++; + if (polls === 1) return new Response('{"events":[{"event":"TASK_UPDATED","payload":{"id":"t1"}}]}', { status: 200 }); + // Later polls wait like the server does, until the client gives up on them + return new Promise((_, reject) => init?.signal?.addEventListener('abort', () => reject(new Error('aborted')))); + } + return new Response('{}', { status: 200 }); + }) + ); + const api = await loadApi(); + const seen: string[] = []; + api.onWebSocketMessage((m) => seen.push(m.event)); + await api.connectWebSocket(); + await vi.advanceTimersByTimeAsync(100); + + expect(FakeEventSource.instances).toHaveLength(0); + expect(api.isRealtimeConnected()).toBe(true); + expect(api.getRealtimeTransport()).toBe('poll'); + expect(seen).toEqual(['TASK_UPDATED']); + expect(polls).toBe(2); // the next poll is waiting + api.disconnectWebSocket(); + expect(api.isRealtimeConnected()).toBe(false); + }); }); diff --git a/frontend/src/lib/__tests__/useConnectionState.test.tsx b/frontend/src/lib/__tests__/useConnectionState.test.tsx index 346f1bf..0999aad 100644 --- a/frontend/src/lib/__tests__/useConnectionState.test.tsx +++ b/frontend/src/lib/__tests__/useConnectionState.test.tsx @@ -1,10 +1,10 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import { act, renderHook } from '@testing-library/react'; -let statusListener: ((connected: boolean) => void) | null = null; +let statusListener: ((connected: boolean, transport?: string) => void) | null = null; let currentConnected = false; vi.mock('../api', () => ({ - onRealtimeStatus: (cb: (connected: boolean) => void) => { + onRealtimeStatus: (cb: (connected: boolean, transport?: string) => void) => { statusListener = cb; cb(currentConnected); return () => (statusListener = null); @@ -86,6 +86,15 @@ describe('useConnectionState', () => { expect(fetchMock.mock.calls.length).toBe(calls); }); + it('is limited while changes only come by long polling', async () => { + const { result } = renderHook(() => useConnectionState()); + act(() => statusListener?.(true, 'poll')); + expect(result.current).toBe('limited'); + expect(fetchMock).not.toHaveBeenCalled(); + act(() => statusListener?.(true, 'sse')); + expect(result.current).toBe('online'); + }); + it('repeated failed reconnects do not postpone the check', async () => { currentConnected = true; const { result } = renderHook(() => useConnectionState()); diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 2188710..6f3b21e 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1464,7 +1464,8 @@ export const api = { }; -export type RealtimeTransport = 'ws' | 'sse' | 'none'; +// poll: long polling, when a proxy lets neither WebSocket nor event streams through +export type RealtimeTransport = 'ws' | 'sse' | 'poll' | 'none'; let activeTransport: RealtimeTransport = 'none'; let activeGlobalWS: WebSocket | null = null; @@ -1599,7 +1600,7 @@ export async function connectSSE(rooms?: string[]): Promise { clearTimeout(sseFirstEventWatchdog); if (isSseBufferedByProxy) { - notifyRealtimeStatusChange(false, 'none'); + void connectPoll(); return; } @@ -1640,6 +1641,7 @@ export async function connectSSE(rooms?: string[]): Promise { } catch {} if (activeGlobalSSE === es) activeGlobalSSE = null; notifyRealtimeStatusChange(false, 'none'); + void connectPoll(); }, SSE_FIRST_EVENT_TIMEOUT_MS); }; @@ -1694,6 +1696,78 @@ export async function connectSSE(rooms?: string[]): Promise { } } +// Long polling: the server buffers the events of a poll connection; each request waits until there are +// some (at most POLL_WAIT_S). Room joins use the poll id like an SSE id (activeSSEId). +const POLL_WAIT_S = 25; // below typical proxy timeouts, like the note sync's long polling +let pollRunning = false; +let pollAbort: AbortController | null = null; + +function dispatchRealtimeMessage(msg: WSMessage) { + wsListeners.forEach((fn) => { + try { + fn(msg); + } catch (e) { + console.error('Realtime listener error', e); + } + }); + window.dispatchEvent(new CustomEvent('mindsift:ws', { detail: msg })); +} + +async function connectPoll(): Promise { + if (pollRunning || manualDisconnect || !getToken()) return; + pollRunning = true; + let pollId: string | null = null; + let failures = 0; + try { + while (!manualDisconnect && getToken()) { + const ctrl = new AbortController(); + pollAbort = ctrl; + try { + const fresh = !pollId; + if (!pollId) { + const opened = await request<{ pollId?: string }>('/api/v1/events/poll', { + method: 'POST', + body: JSON.stringify({ rooms: Array.from(globalSubscribedRooms) }), + }); + if (!opened?.pollId) throw new Error('no poll id'); + pollId = opened.pollId; + activeSSEId = pollId; + notifyRealtimeStatusChange(true, 'poll'); + } + const started = Date.now(); + const res = await fetch(`/api/v1/events/poll/${encodeURIComponent(pollId)}?wait=${POLL_WAIT_S}`, { + headers: { Authorization: `Bearer ${getToken()}` }, + cache: 'no-store', + signal: ctrl.signal, + }); + if (res.status === 404 && !fresh) { + // Expired (e.g. device asleep, server restart): open a new one, rooms go along + pollId = null; + continue; + } + if (!res.ok) throw new Error(`HTTP ${res.status}`); + const events = ((await res.json()) as { events?: WSMessage[] }).events || []; + failures = 0; + events.forEach(dispatchRealtimeMessage); + // An empty answer long before the wait ran out (proxy answering on its own): no tight loop + if (events.length === 0 && Date.now() - started < 1000) await new Promise((r) => setTimeout(r, 1000)); + } catch { + if (ctrl.signal.aborted || manualDisconnect) break; + pollId = null; + if (activeTransport === 'poll') { + activeSSEId = null; + notifyRealtimeStatusChange(false, 'none'); + } + failures++; + await new Promise((r) => setTimeout(r, Math.min(30000, 2000 * failures))); + } + } + } finally { + pollRunning = false; + pollAbort = null; + } +} + let wsTicketPending = false; export async function connectWebSocket(): Promise { @@ -1845,6 +1919,7 @@ if (typeof window !== 'undefined') { } export function disconnectWebSocket(): void { + pollAbort?.abort(); clearInterval(wsHeartbeatTimer); clearTimeout(sseFirstEventWatchdog); manualDisconnect = true; diff --git a/frontend/src/lib/docSync/engine.ts b/frontend/src/lib/docSync/engine.ts index ae85552..cc9c53a 100644 --- a/frontend/src/lib/docSync/engine.ts +++ b/frontend/src/lib/docSync/engine.ts @@ -641,6 +641,8 @@ export class DocSyncEngine { } private channel(): SyncChannel { + // The app-wide long polling brings the notifications: same channel as the note's own long polling + if (this.pushKind === 'poll') return 'longpoll'; if (this.pushKind !== 'none') return this.pushKind; if (this.longPollRunning && !this.longPollBroken) return 'longpoll'; return 'interval'; diff --git a/frontend/src/lib/docSync/push.ts b/frontend/src/lib/docSync/push.ts index c3dad2d..1497b4f 100644 --- a/frontend/src/lib/docSync/push.ts +++ b/frontend/src/lib/docSync/push.ts @@ -1,8 +1,9 @@ // 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). +// leases or its presence change. Delivery uses the app-wide realtime connection (WebSocket, else SSE, else +// long polling). import { createRealtimeConnection, getRealtimeTransport, isRealtimeConnected } from '../api'; -export type PushKind = 'ws' | 'sse' | 'none'; +export type PushKind = 'ws' | 'sse' | 'poll' | 'none'; export interface PushChannel { /** Calls onPoke for changes of docId and onStatus whenever the channel goes up or down. */ diff --git a/frontend/src/lib/useConnectionState.ts b/frontend/src/lib/useConnectionState.ts index 94268ef..123877a 100644 --- a/frontend/src/lib/useConnectionState.ts +++ b/frontend/src/lib/useConnectionState.ts @@ -6,9 +6,9 @@ export const SERVER_OFFLINE_GRACE_MS = 4000; // How often the server is asked while the realtime connection is missing export const SERVER_PROBE_INTERVAL_MS = 15000; -// online: realtime connection (WebSocket/SSE) is up; limited: the server answers plain requests, but no -// realtime connection gets through (e.g. a proxy that buffers event streams; notes fall back to long -// polling); offline: no server at all +// online: realtime connection (WebSocket/SSE) is up; limited: changes only come by long polling (a proxy +// that buffers event streams), or not live at all while the server still answers plain requests; +// offline: no server at all export type ConnectionState = 'online' | 'limited' | 'offline'; async function serverReachable(): Promise { @@ -29,6 +29,7 @@ export function useConnectionState(): ConnectionState { useEffect(() => { let connected = false; + let viaPoll = false; let disposed = false; let timer: number | undefined; const stopProbing = () => { @@ -45,7 +46,7 @@ export function useConnectionState(): ConnectionState { const update = () => { if (connected) { stopProbing(); - setState('online'); + setState(viaPoll ? 'limited' : 'online'); } else if (navigator.onLine === false) { stopProbing(); setState('offline'); @@ -53,8 +54,9 @@ export function useConnectionState(): ConnectionState { timer = window.setTimeout(probe, SERVER_OFFLINE_GRACE_MS); } }; - const unsubscribe = onRealtimeStatus((isConnected) => { + const unsubscribe = onRealtimeStatus((isConnected, transport) => { connected = isConnected; + viaPoll = transport === 'poll'; update(); }); window.addEventListener('online', update); diff --git a/internal/api/realtime.go b/internal/api/realtime.go index fe44e79..5140fd8 100644 --- a/internal/api/realtime.go +++ b/internal/api/realtime.go @@ -3,7 +3,9 @@ package api import ( "encoding/json" "net/http" + "strconv" "strings" + "time" "github.com/go-chi/chi/v5" "github.com/sandrodach/mindsift/internal/auth" @@ -125,3 +127,29 @@ func (s *Server) handleSSERooms(w http.ResponseWriter, r *http.Request) { } writeJSON(w, http.StatusOK, map[string]bool{"success": true}) } + +// handleOpenPoll starts a long-poll connection (third transport after WebSocket and SSE, for proxies that +// let neither through). Rooms can be given right away or joined later via /events/rooms with the id. +func (s *Server) handleOpenPoll(w http.ResponseWriter, r *http.Request) { + var req struct { + Rooms []string `json:"rooms"` + } + _ = json.NewDecoder(http.MaxBytesReader(w, r.Body, 16384)).Decode(&req) + writeJSON(w, http.StatusOK, map[string]string{"pollId": s.wsHub.OpenPoll(auth.GetUserID(r.Context()), req.Rooms)}) +} + +// handlePoll returns the events of a poll connection as soon as there are some, at the latest after +// `wait` seconds (empty list). 404: the connection expired, the client opens a new one. +func (s *Server) handlePoll(w http.ResponseWriter, r *http.Request) { + wait := longPollMax + if sec, err := strconv.Atoi(r.URL.Query().Get("wait")); err == nil && sec >= 0 && time.Duration(sec)*time.Second < wait { + wait = time.Duration(sec) * time.Second + } + _ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(wait + 10*time.Second)) + events, ok := s.wsHub.Poll(r.Context(), chi.URLParam(r, "id"), auth.GetUserID(r.Context()), wait) + if !ok { + writeError(w, http.StatusNotFound, "Verbindung abgelaufen") + return + } + writeJSON(w, http.StatusOK, map[string]interface{}{"events": events}) +} diff --git a/internal/api/realtime_test.go b/internal/api/realtime_test.go index 5c22817..65fc55e 100644 --- a/internal/api/realtime_test.go +++ b/internal/api/realtime_test.go @@ -7,6 +7,7 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "strconv" "strings" "testing" "time" @@ -354,3 +355,62 @@ func TestLogoutEndsLiveStreams(t *testing.T) { } } } + +// Long polling as the third transport: events of the user and of joined rooms arrive with the next +// poll, a poll waits until something happens, and nobody else can read the connection. +func TestLongPollDeliversEvents(t *testing.T) { + srv, store, ts, owner, token := startRealtimeServer(t) + _, strangerToken := newTestUser(t, store, "stranger@example.com") + + resp := doJSON(t, ts, http.MethodPost, "/api/v1/events/poll", token, map[string]interface{}{}) + var opened struct { + PollID string `json:"pollId"` + } + _ = json.NewDecoder(resp.Body).Decode(&opened) + if resp.StatusCode != http.StatusOK || opened.PollID == "" { + t.Fatalf("open poll: %d %+v", resp.StatusCode, opened) + } + poll := func(tok string, wait int) (int, []sseMsg) { + resp := doJSON(t, ts, http.MethodGet, "/api/v1/events/poll/"+opened.PollID+"?wait="+strconv.Itoa(wait), tok, nil) + var out struct { + Events []sseMsg `json:"events"` + } + _ = json.NewDecoder(resp.Body).Decode(&out) + return resp.StatusCode, out.Events + } + + // Nothing happens: the poll answers empty after the wait + if code, events := poll(token, 0); code != http.StatusOK || len(events) != 0 { + t.Fatalf("empty poll: %d %v", code, events) + } + + // A poll that is waiting returns as soon as an event comes + go func() { + time.Sleep(200 * time.Millisecond) + srv.wsHub.BroadcastToUser(owner.ID, "TASK_UPDATED", map[string]string{"id": "t1"}) + }() + start := time.Now() + code, events := poll(token, 5) + if code != http.StatusOK || len(events) != 1 || events[0].Event != "TASK_UPDATED" || time.Since(start) > 3*time.Second { + t.Fatalf("waiting poll: %d %v after %s", code, events, time.Since(start)) + } + + // List rooms are joined like with SSE; events between polls wait in the buffer + list, _ := store.CreateList(owner.ID, "Einkauf", "🛒", "shopping") + if r := doJSON(t, ts, http.MethodPost, "/api/v1/events/rooms", token, map[string]string{"sseId": opened.PollID, "room": list.ID}); r.StatusCode != http.StatusOK { + t.Fatalf("join list room: %d", r.StatusCode) + } + srv.wsHub.BroadcastListEvent(list.ID, "ITEM_ADDED", map[string]string{"id": "i1"}) + srv.wsHub.BroadcastListEvent(list.ID, "ITEM_UPDATED", map[string]string{"id": "i1"}) + if code, events = poll(token, 5); code != http.StatusOK || len(events) != 2 || events[1].Event != "ITEM_UPDATED" { + t.Fatalf("list events: %d %v", code, events) + } + + if code, _ = poll(strangerToken, 0); code != http.StatusNotFound { + t.Errorf("someone else's poll connection: want 404, got %d", code) + } + resp = doJSON(t, ts, http.MethodGet, "/api/v1/events/poll/poll_unknown?wait=0", token, nil) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("unknown poll connection: want 404, got %d", resp.StatusCode) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 6b1c087..91fd275 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -164,6 +164,8 @@ func (s *Server) Router() http.Handler { r.Get("/api/v1/auth/me", s.handleMe) r.Post("/api/v1/events/rooms", s.handleSSERooms) + r.Post("/api/v1/events/poll", s.handleOpenPoll) + r.Get("/api/v1/events/poll/{id}", s.handlePoll) r.Post("/api/v1/auth/stream-ticket", s.handleStreamTicket) r.Post("/api/v1/voice/transcribe", s.handleVoiceTranscribe) r.Post("/api/v1/documents/{id}/media", s.handleUploadMedia) diff --git a/internal/ws/hub.go b/internal/ws/hub.go index c40bbec..de19dfc 100644 --- a/internal/ws/hub.go +++ b/internal/ws/hub.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "log/slog" @@ -104,6 +105,7 @@ type Hub struct { sseClients map[*SSEClient]bool sseUserRooms map[string]map[*SSEClient]bool sseRooms map[string]map[*SSEClient]bool + polls map[string]*pollConn closed bool // authorizeRoom decides whether a user may receive events of a room; without it every join is denied @@ -136,6 +138,112 @@ func NewHub() *Hub { sseClients: make(map[*SSEClient]bool), sseUserRooms: make(map[string]map[*SSEClient]bool), sseRooms: make(map[string]map[*SSEClient]bool), + polls: make(map[string]*pollConn), + } +} + +// Long polling, for networks that let neither WebSocket nor event streams through (proxies that buffer +// SSE): a poll connection is an SSEClient without a stream. Its events wait in the send buffer until the +// next poll picks them up; room joins work like for SSE (JoinSSERoomByID with its id). + +// PollIdleTimeout: a poll connection nobody asked for this long is closed (tab closed, device asleep). +const PollIdleTimeout = 2 * time.Minute + +type pollConn struct { + client *SSEClient + lastPoll time.Time + waiting int // polls running right now +} + +// OpenPoll registers a poll connection for the user and returns its id. +func (h *Hub) OpenPoll(userID string, rooms []string) string { + h.sweepPolls() + c := &SSEClient{ + id: domain.NewID("poll"), + userID: userID, + send: make(chan []byte, 256), + subscribedRooms: make(map[string]bool), + } + for _, room := range rooms { + if room = strings.TrimSpace(room); room != "" { + c.subscribedRooms[room] = true + } + } + h.RegisterSSE(c) + h.mu.Lock() + h.polls[c.id] = &pollConn{client: c, lastPoll: time.Now()} + h.mu.Unlock() + return c.id +} + +// Poll waits up to wait for events of the user's poll connection id and returns everything buffered. +// ok is false when the connection is unknown or closed (expired, server restart): open a new one. +func (h *Hub) Poll(ctx context.Context, id, userID string, wait time.Duration) (events []json.RawMessage, ok bool) { + h.sweepPolls() + h.mu.Lock() + p := h.polls[id] + if p == nil || p.client.userID != userID { + h.mu.Unlock() + return nil, false + } + p.waiting++ + h.mu.Unlock() + defer func() { + h.mu.Lock() + p.waiting-- + p.lastPoll = time.Now() + h.mu.Unlock() + }() + + // What is already waiting goes out right away; otherwise wait for the first event + events, open := drainBuffered(p.client.send) + if len(events) > 0 || !open { + return events, open || len(events) > 0 + } + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case msg, open := <-p.client.send: + if !open { + return nil, false + } + rest, _ := drainBuffered(p.client.send) // whatever came along goes in the same answer + return append([]json.RawMessage{msg}, rest...), true + case <-timer.C: + case <-ctx.Done(): + } + return events, true +} + +// drainBuffered takes everything waiting in ch without blocking; open is false once ch is closed. +func drainBuffered(ch chan []byte) (events []json.RawMessage, open bool) { + events = []json.RawMessage{} + for { + select { + case msg, ok := <-ch: + if !ok { + return events, false + } + events = append(events, msg) + default: + return events, true + } + } +} + +// sweepPolls closes poll connections that were not asked for within PollIdleTimeout. +func (h *Hub) sweepPolls() { + h.mu.Lock() + var stale []*SSEClient + for id, p := range h.polls { + if p.waiting == 0 && time.Since(p.lastPoll) > PollIdleTimeout { + stale = append(stale, p.client) + delete(h.polls, id) + } + } + h.mu.Unlock() + for _, c := range stale { + h.UnregisterSSE(c) } } diff --git a/internal/ws/hub_test.go b/internal/ws/hub_test.go index e107485..9a6eb03 100644 --- a/internal/ws/hub_test.go +++ b/internal/ws/hub_test.go @@ -1,6 +1,7 @@ package ws import ( + "context" "sync" "testing" "time" @@ -90,3 +91,29 @@ func TestRoomJoinsRequireAuthorization(t *testing.T) { t.Fatal("joining with another user's sseId must fail") } } + +// Poll connections nobody asks for any more (tab closed) are closed and stop receiving events. +func TestIdlePollConnectionsAreClosed(t *testing.T) { + hub := NewHub() + id := hub.OpenPoll("user-1", nil) + hub.BroadcastToUser("user-1", "TEST_EVENT", nil) + if events, ok := hub.Poll(context.Background(), id, "user-1", 0); !ok || len(events) != 1 { + t.Fatalf("fresh poll connection: ok=%v events=%d", ok, len(events)) + } + if _, ok := hub.Poll(context.Background(), id, "user-2", 0); ok { + t.Fatal("another user must not read the connection") + } + + hub.mu.Lock() + hub.polls[id].lastPoll = time.Now().Add(-PollIdleTimeout - time.Second) + hub.mu.Unlock() + hub.OpenPoll("user-1", nil) // any poll activity sweeps + if _, ok := hub.Poll(context.Background(), id, "user-1", 0); ok { + t.Fatal("idle poll connection should be gone") + } + hub.mu.RLock() + defer hub.mu.RUnlock() + if len(hub.sseUserRooms["user-1"]) != 1 { + t.Fatalf("only the new connection should stay registered, got %d", len(hub.sseUserRooms["user-1"])) + } +}