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"])) + } +}