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 <noreply@anthropic.com>
This commit is contained in:
sandrodachandClaude Opus 5.5 committed 2026-11-02 11:50:27 +01:00
1 parent c19d8fd03e
commit 2b7356d043
16 files changed
+369 -20

No files matched your search

+2
View File
@@ -28,6 +28,8 @@ Basis: `/api/v1`. Antworten sind JSON; Fehler haben die Form `{"error": "Text"}`
|---|---|---| |---|---|---|
| GET | `/auth/me` | Benutzer und Kennzahlen | | 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/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 | `/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 | `/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 | | POST | `/auth/logout` | Sitzung dieses Geräts beenden |
+3 -2
View File
@@ -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). 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 | | 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` | | 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: 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. Ü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. **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.
+5 -5
View File
@@ -5,10 +5,10 @@
``` ```
Browser (React SPA, PWA-Manifest) Browser (React SPA, PWA-Manifest)
│ REST /api/v1/* Authorization: Bearer <JWT | mindsift_sec_…> │ REST /api/v1/* Authorization: Bearer <JWT | mindsift_sec_…>
│ 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 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/auth JWT, bcrypt, API-Key-Hash
▼ internal/config ▼ internal/config
SQLite (modernc.org/sqlite, WAL) ── internal/storage 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 | | Pfad | Inhalt |
|---|---| |---|---|
| `cmd/server/` | `main.go`, eingebettetes `dist/` | | `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/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/domain/` | Strukturen (`models.go`), ID-Erzeugung, Tag-Extraktion |
| `internal/ws/` | `hub.go` | | `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/auth/` | Tokens und Passwörter |
| `internal/config/` | Umgebungsvariablen | | `internal/config/` | Umgebungsvariablen |
| `frontend/src/features/` | `triage`, `workspace`, `lists`, `teams`, `settings`, `capture`, `auth`, `help` | | `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`) | | `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`) | | `e2e/` | Browser-Tests mit Playwright gegen einen echten Server (`e2e/run.sh`) |
| `frontend/src/themes/` | CSS pro Theme | | `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)). - **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. - **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…“. - **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) ## App-Icons, PWA & Social-Preview (Open Graph)
+1 -1
View File
@@ -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 | | 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 | | 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 | | 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. 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.
+1 -1
View File
@@ -89,7 +89,7 @@ const CONNECTION_STATUS = {
limited: { limited: {
label: 'eingeschränkt', label: 'eingeschränkt',
dot: 'bg-status-warn', 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' }, offline: { label: 'offline', dot: 'bg-status-err', title: 'Offline – keine Verbindung zum Server; Änderungen werden später synchronisiert' },
} as const; } as const;
@@ -137,4 +137,36 @@ describe('realtime fallback', () => {
expect(FakeEventSource.instances).toHaveLength(1); // no reconnect loop: note sync uses long polling expect(FakeEventSource.instances).toHaveLength(1); // no reconnect loop: note sync uses long polling
api.disconnectWebSocket(); 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<Response>((_, 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);
});
}); });
@@ -1,10 +1,10 @@
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import { act, renderHook } from '@testing-library/react'; 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; let currentConnected = false;
vi.mock('../api', () => ({ vi.mock('../api', () => ({
onRealtimeStatus: (cb: (connected: boolean) => void) => { onRealtimeStatus: (cb: (connected: boolean, transport?: string) => void) => {
statusListener = cb; statusListener = cb;
cb(currentConnected); cb(currentConnected);
return () => (statusListener = null); return () => (statusListener = null);
@@ -86,6 +86,15 @@ describe('useConnectionState', () => {
expect(fetchMock.mock.calls.length).toBe(calls); 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 () => { it('repeated failed reconnects do not postpone the check', async () => {
currentConnected = true; currentConnected = true;
const { result } = renderHook(() => useConnectionState()); const { result } = renderHook(() => useConnectionState());
+77 -2
View File
@@ -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 activeTransport: RealtimeTransport = 'none';
let activeGlobalWS: WebSocket | null = null; let activeGlobalWS: WebSocket | null = null;
@@ -1599,7 +1600,7 @@ export async function connectSSE(rooms?: string[]): Promise<void> {
clearTimeout(sseFirstEventWatchdog); clearTimeout(sseFirstEventWatchdog);
if (isSseBufferedByProxy) { if (isSseBufferedByProxy) {
notifyRealtimeStatusChange(false, 'none'); void connectPoll();
return; return;
} }
@@ -1640,6 +1641,7 @@ export async function connectSSE(rooms?: string[]): Promise<void> {
} catch {} } catch {}
if (activeGlobalSSE === es) activeGlobalSSE = null; if (activeGlobalSSE === es) activeGlobalSSE = null;
notifyRealtimeStatusChange(false, 'none'); notifyRealtimeStatusChange(false, 'none');
void connectPoll();
}, SSE_FIRST_EVENT_TIMEOUT_MS); }, SSE_FIRST_EVENT_TIMEOUT_MS);
}; };
@@ -1694,6 +1696,78 @@ export async function connectSSE(rooms?: string[]): Promise<void> {
} }
} }
// 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<void> {
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; let wsTicketPending = false;
export async function connectWebSocket(): Promise<void> { export async function connectWebSocket(): Promise<void> {
@@ -1845,6 +1919,7 @@ if (typeof window !== 'undefined') {
} }
export function disconnectWebSocket(): void { export function disconnectWebSocket(): void {
pollAbort?.abort();
clearInterval(wsHeartbeatTimer); clearInterval(wsHeartbeatTimer);
clearTimeout(sseFirstEventWatchdog); clearTimeout(sseFirstEventWatchdog);
manualDisconnect = true; manualDisconnect = true;
+2
View File
@@ -641,6 +641,8 @@ export class DocSyncEngine {
} }
private channel(): SyncChannel { 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.pushKind !== 'none') return this.pushKind;
if (this.longPollRunning && !this.longPollBroken) return 'longpoll'; if (this.longPollRunning && !this.longPollBroken) return 'longpoll';
return 'interval'; return 'interval';
+3 -2
View File
@@ -1,8 +1,9 @@
// Push notifications for note sync: the server sends DOC_SEQ into the note's room whenever the note, its // 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'; import { createRealtimeConnection, getRealtimeTransport, isRealtimeConnected } from '../api';
export type PushKind = 'ws' | 'sse' | 'none'; export type PushKind = 'ws' | 'sse' | 'poll' | 'none';
export interface PushChannel { export interface PushChannel {
/** Calls onPoke for changes of docId and onStatus whenever the channel goes up or down. */ /** Calls onPoke for changes of docId and onStatus whenever the channel goes up or down. */
+7 -5
View File
@@ -6,9 +6,9 @@ export const SERVER_OFFLINE_GRACE_MS = 4000;
// How often the server is asked while the realtime connection is missing // How often the server is asked while the realtime connection is missing
export const SERVER_PROBE_INTERVAL_MS = 15000; export const SERVER_PROBE_INTERVAL_MS = 15000;
// online: realtime connection (WebSocket/SSE) is up; limited: the server answers plain requests, but no // online: realtime connection (WebSocket/SSE) is up; limited: changes only come by long polling (a proxy
// realtime connection gets through (e.g. a proxy that buffers event streams; notes fall back to long // that buffers event streams), or not live at all while the server still answers plain requests;
// polling); offline: no server at all // offline: no server at all
export type ConnectionState = 'online' | 'limited' | 'offline'; export type ConnectionState = 'online' | 'limited' | 'offline';
async function serverReachable(): Promise<boolean> { async function serverReachable(): Promise<boolean> {
@@ -29,6 +29,7 @@ export function useConnectionState(): ConnectionState {
useEffect(() => { useEffect(() => {
let connected = false; let connected = false;
let viaPoll = false;
let disposed = false; let disposed = false;
let timer: number | undefined; let timer: number | undefined;
const stopProbing = () => { const stopProbing = () => {
@@ -45,7 +46,7 @@ export function useConnectionState(): ConnectionState {
const update = () => { const update = () => {
if (connected) { if (connected) {
stopProbing(); stopProbing();
setState('online'); setState(viaPoll ? 'limited' : 'online');
} else if (navigator.onLine === false) { } else if (navigator.onLine === false) {
stopProbing(); stopProbing();
setState('offline'); setState('offline');
@@ -53,8 +54,9 @@ export function useConnectionState(): ConnectionState {
timer = window.setTimeout(probe, SERVER_OFFLINE_GRACE_MS); timer = window.setTimeout(probe, SERVER_OFFLINE_GRACE_MS);
} }
}; };
const unsubscribe = onRealtimeStatus((isConnected) => { const unsubscribe = onRealtimeStatus((isConnected, transport) => {
connected = isConnected; connected = isConnected;
viaPoll = transport === 'poll';
update(); update();
}); });
window.addEventListener('online', update); window.addEventListener('online', update);
+28
View File
@@ -3,7 +3,9 @@ package api
import ( import (
"encoding/json" "encoding/json"
"net/http" "net/http"
"strconv"
"strings" "strings"
"time"
"github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5"
"github.com/sandrodach/mindsift/internal/auth" "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}) 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})
}
+60
View File
@@ -7,6 +7,7 @@ import (
"encoding/json" "encoding/json"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strconv"
"strings" "strings"
"testing" "testing"
"time" "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)
}
}
+2
View File
@@ -164,6 +164,8 @@ func (s *Server) Router() http.Handler {
r.Get("/api/v1/auth/me", s.handleMe) r.Get("/api/v1/auth/me", s.handleMe)
r.Post("/api/v1/events/rooms", s.handleSSERooms) 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/auth/stream-ticket", s.handleStreamTicket)
r.Post("/api/v1/voice/transcribe", s.handleVoiceTranscribe) r.Post("/api/v1/voice/transcribe", s.handleVoiceTranscribe)
r.Post("/api/v1/documents/{id}/media", s.handleUploadMedia) r.Post("/api/v1/documents/{id}/media", s.handleUploadMedia)
+108
View File
@@ -1,6 +1,7 @@
package ws package ws
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log/slog" "log/slog"
@@ -104,6 +105,7 @@ type Hub struct {
sseClients map[*SSEClient]bool sseClients map[*SSEClient]bool
sseUserRooms map[string]map[*SSEClient]bool sseUserRooms map[string]map[*SSEClient]bool
sseRooms map[string]map[*SSEClient]bool sseRooms map[string]map[*SSEClient]bool
polls map[string]*pollConn
closed bool closed bool
// authorizeRoom decides whether a user may receive events of a room; without it every join is denied // 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), sseClients: make(map[*SSEClient]bool),
sseUserRooms: make(map[string]map[*SSEClient]bool), sseUserRooms: make(map[string]map[*SSEClient]bool),
sseRooms: 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)
} }
} }
+27
View File
@@ -1,6 +1,7 @@
package ws package ws
import ( import (
"context"
"sync" "sync"
"testing" "testing"
"time" "time"
@@ -90,3 +91,29 @@ func TestRoomJoinsRequireAuthorization(t *testing.T) {
t.Fatal("joining with another user's sseId must fail") 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"]))
}
}