From 6343d62bab5ddbe3c8d7dbc45c764456dd4b0b69 Mon Sep 17 00:00:00 2001 From: Sandro Date: Mon, 26 Oct 2026 16:55:11 +0100 Subject: [PATCH] feat(realtime): implement SSE fallback, dual-transport hub & proxy resilience (resolves FLAW-03) --- Doku/README.md | 11 +- Doku/api/websocket.md | 57 ++- Doku/flaws.md | 29 +- Doku/reverse-proxy.md | 229 ++++++++++ .../src/features/lists/SharedListsView.tsx | 172 +++---- .../src/features/workspace/WorkspaceView.tsx | 189 ++++---- frontend/src/lib/api.ts | 422 +++++++++++++++++- internal/api/server.go | 116 ++++- internal/api/server_test.go | 107 +++++ internal/ws/hub.go | 310 ++++++++++++- 10 files changed, 1402 insertions(+), 240 deletions(-) create mode 100644 Doku/reverse-proxy.md diff --git a/Doku/README.md b/Doku/README.md index bea5359..ef02845 100644 --- a/Doku/README.md +++ b/Doku/README.md @@ -20,6 +20,7 @@ Doku/ │ ├── workspace.md # Modul 3: Workspace Outliner-Engine, Wiki-Links & Pure-DB Storage │ └── lists.md # Modul 4: Geteilte Listen (Live-Kollaboration & D&D) ├── sharing-and-permissions.md # Synchrones Teilen, Rollen, Isolation & Mount-Entkopplung +├── reverse-proxy.md # Nginx, Traefik, Caddy, Cloudflare Konfiguration & SSE ├── theming.md # Single-File CSS Theme-Engine & Design-System └── flaws.md # Technische Schwachstellen, Risiken & zukünftige Optimierungen ``` @@ -32,7 +33,7 @@ Doku/ 3. **[REST-API Referenz](api/rest.md)** Spezifikation aller Endpunkte für Authentifizierung, Gedanken, Aufgaben, Dokumente, Listen, Teams und Webhooks. 4. **[WebSocket & Echtzeit-Synchronisation](api/websocket.md)** - Gorilla WebSocket Hub, Rooms, Presence-Tracking, Document Locking Mechanismen und Polling-Fallback. + Gorilla WebSocket Hub, Rooms, Presence-Tracking, Yjs CRDT Kollaboration und SSE Fallback. 5. **[Kernmodul: Triage & Quick-Dump](modules/triage.md)** Instant-Capture-Pipeline, Finite State Machine, Gestensteuerung, Spracheingabe und Webhook-Ingestion. 6. **[Kernmodul: Aufgaben & Kanban-System](modules/tasks.md)** @@ -43,10 +44,12 @@ Doku/ Listen-Typen, Mengen, Kategorien, Live-Checkoff und Drag & Drop Sortierung. 9. **[Synchrones Teilen & Berechtigungskonzept](sharing-and-permissions.md)** Rollenmodell (Admin/Member), System-Isolations-Modi (`strict_private`, `teams_only`, `open`), entkoppelte Ordner-Mounts und Notizen-Freigaben. -10. **[Single-File Theming & Design-System](theming.md)** +10. **[Reverse-Proxy- & Ingress-Handbuch](reverse-proxy.md)** + Produktionsfertige Konfigurationen für Nginx, Traefik, Caddy, Cloudflare und Apache mit WebSocket-Upgrades und SSE-Streaming. +11. **[Single-File Theming & Design-System](theming.md)** 1-Datei = 1-Theme Philosophie, die 6 CSS-Variablen-Blöcke, Semantische CSS-Klassen und Geometrie-Steuerung. -11. **[Technische Schwachstellen & Optimierungspotenziale (Flaws)](flaws.md)** - Kritische Analyse von Concurrency-Grenzen, WebSocket-Proxies, Undo/Redo, Volltextsuche und Skalierungs-Engpässen. +12. **[Technische Schwachstellen & Optimierungspotenziale (Flaws)](flaws.md)** + Kritische Analyse von Concurrency-Grenzen, Volltextsuche, Viewport-Anpassung und Skalierungs-Engpässen. --- diff --git a/Doku/api/websocket.md b/Doku/api/websocket.md index e90cfa3..ba172ba 100644 --- a/Doku/api/websocket.md +++ b/Doku/api/websocket.md @@ -114,10 +114,57 @@ Vor Version 2.1 nutzte MindSift ein serverseitiges Lease-Locking mit 30-Sekunden --- -## 🔄 5. Fallback-Strategie & Reconnect +## 🔄 5. Fallback-Strategie & Proxy-Resilienz (Dual-Transport Architektur) -In strikten Firmennetzwerken oder bei Proxy-Problemen (z.B. fehlende `Upgrade`-Header bei HTTP/1.1) kann eine WebSocket-Verbindung fehlschlagen: +In restriktiven Firmennetzwerken, hinter Unternehmens-Proxies (z.B. Zscaler, BlueCoat) oder fehlerhaft konfigurierten Reverse-Proxies wird der `Upgrade: websocket`-Header häufig verworfen (`NS_ERROR_CONNECTION_REFUSED`, Code 1006 / 1002 / 1008). + +MindSift löst dieses Problem durch eine **vollwertige Dual-Transport-Architektur** ([FLAW-03](../flaws.md)): + +``` + ┌───────────────────────────────┐ + │ MindSift Frontend │ + └───────────────┬───────────────┘ + │ + ┌─────────────────────┴─────────────────────┐ + ▼ ▼ + Standard: WebSockets Fallback: SSE + REST + (wss://.../api/v1/ws) (https://.../api/v1/events) + │ │ + │ [Proxy Upgrade Block (Code 1006/Timeout)]│ + │ ────────────────────────────────────────> │ + ▼ ▼ + ┌────────────────────────┐ ┌────────────────────────┐ + │ Gorilla WS Conn │ │ Server-Sent Events │ + │ Bidirektionaler │ │ (text/event-stream) │ + │ TCP-WebSocket │ │ Standard HTTPS Port │ + └────────────┬───────────┘ └────────────┬───────────┘ + │ │ + └─────────────────────┬─────────────────────┘ + ▼ + ┌─────────────────────────────┐ + │ Realtime Hub │ + │ (Multiplexing WS & SSE) │ + └─────────────────────────────┘ +``` + +### 1. Automatischer SSE-Fallback (`GET /api/v1/events`) +- Scheitert der WebSocket-Handshake oder blockiert ein Proxy das Upgrade, wechselt MindSift innerhalb von 3,5 Sekunden automatisch auf einen **Server-Sent Events (SSE) Stream**. +- Da SSE ein standardmäßiger HTTP/1.1- bzw. HTTP/2-GET-Request über Standard-Port 443/80 ist (`Content-Type: text/event-stream`), wird er von jedem Proxy und jeder Firewall ungehindert durchgeleitet. +- Header `X-Accel-Buffering: no` und `Cache-Control: no-cache` verhindern Pufferung im Proxy; periodische Ping-Kommentare (`: ping\n\n`) alle 15 Sekunden halten Timeouts fern. +- Der Client bleibt dadurch **vollständig echtzeitfähig** (`Live (SSE)`). + +### 2. REST-Endpunkte für Upstream-Aktionen im SSE-Modus +Da SSE unidirektional (Server ➔ Client) ist, stellt MindSift für Schreibvorgänge im SSE-Modus REST-Äquivalente bereit: +- `PUT /api/v1/documents/:id/crdt`: Unterstützt sowohl volles `state` (Compaction) als auch inkrementelles `update` (Delta). Der Hub persistiert das Update und broadcastet es sofort an alle anderen WS- und SSE-Clients im Dokumentenraum. +- `POST /api/v1/documents/:id/awareness`: Überträgt Cursor- und Blockfokus-Präsenzen. +- `DELETE /api/v1/documents/:id/awareness`: Informiert Peers beim Verlassen des Dokuments. +- `POST /api/v1/events/rooms`: Erlaubt SSE-Clients, Dokument- oder Listenräumen dynamisch beizutreten (`join`) oder sie zu verlassen (`leave`). + +### 3. Verbindungs-Indikator +- **Grün pulsierend:** Live verbunden über WebSockets (`Live (WebSocket)`). +- **Türkis pulsierend:** Live verbunden über Server-Sent Events (`Live (Server-Sent Events / Proxy-Stream)`). +- **Gelb/Amber:** Temporäre Unterbrechung mit automatischem HTTP-Polling-Schutz. + +Ausführliche Konfigurationsbeispiele für Nginx, Traefik, Caddy, Cloudflare und Apache finden sich im separaten Handbuch: +👉 **[Reverse-Proxy-Konfigurationshandbuch](../reverse-proxy.md)** -1. **Exponentielles Backoff:** Das Frontend ([`frontend/src/lib/api.ts`](../../frontend/src/lib/api.ts)) versucht automatisch die Wiederverbindung nach 1s, 2s, 4s, 8s, bis maximal 30s. -2. **HTTP-Polling Fallback:** Geteilte Listen und Notizen aktivieren im Hintergrund ein periodisches Polling (5s Intervall), falls der WebSocket-Status nicht `CONNECTED` ist. -3. **Status-Indikator:** Ein dezenter grüner bzw. gelber Indikator oben rechts im Header zeigt dem Benutzer jederzeit den aktuellen Live-Verbindungsstatus an. diff --git a/Doku/flaws.md b/Doku/flaws.md index c59a7b3..13b33a8 100644 --- a/Doku/flaws.md +++ b/Doku/flaws.md @@ -47,15 +47,26 @@ Diese Datei dokumentiert eine ehrliche und schonungslose technische Analyse alle --- -### FLAW-03: Proxy-Inkompatibilität bei WebSocket-Upgrades -- **Betroffene Komponenten:** [`api/websocket.md`](./api/websocket.md), `frontend/src/lib/stores/websocket.js` -- **Ursache:** In restriktiven Firmennetzwerken oder über bestimmte Reverse-Proxies (Corporate SSL-Termination, Firewalls) schlägt der HTTP-Upgrade-Header `Upgrade: websocket` fehl (`NS_ERROR_CONNECTION_REFUSED` oder `400 Bad Request`). -- **Folgen:** Der Client schaltet in den HTTP-Polling-Fallback um (alle 2 Sekunden). Dadurch: - - Werden Sperren (*Document Locks*) und Online-Statusanzeigen verzögert übertragen. - - Entsteht unnötige Serverlast durch kontinuierliche HTTP-Requests. -- **Empfohlene Lösung:** - 1. Sicherstellen von standardmäßigen Nginx/Traefik-Snippets in der Dokumentation (`proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade";`). - 2. Optionale Implementierung von Server-Sent Events (SSE) als unidirektionale Echtzeit-Alternative für Lesevorgänge bei blockiertem WebSocket. +### FLAW-03: [GELÖST / IMPLEMENTIERT] Proxy-Inkompatibilität bei WebSocket-Upgrades & Polling-Fallback +- **Status:** **Gelöst & Vollständig Implementiert (Dual-Transport SSE Fallback & Proxy-Resilienz)** +- **Betroffene Komponenten:** [`api/websocket.md`](./api/websocket.md), [`reverse-proxy.md`](./reverse-proxy.md), [`internal/ws/hub.go`](../internal/ws/hub.go), [`internal/api/server.go`](../internal/api/server.go), [`frontend/src/lib/api.ts`](../frontend/src/lib/api.ts) +- **Problemursache:** Corporate Proxies (z.B. Zscaler, BlueCoat, strikte Firewalls) oder unvollständig konfigurierte Reverse-Proxies verwerfen den HTTP-Upgrade-Header `Upgrade: websocket` (`NS_ERROR_CONNECTION_REFUSED`, Status 1006 / 1002). Früher fiel der Client dabei auf 2-Sekunden-Polling zurück, was zu Verzögerungen und Last führte. +- **Implementierte Lösung:** + 1. **Unidirektionaler Server-Sent Events (SSE) Fallback:** + - Neuer Streaming-Endpunkt `GET /api/v1/events` im Go-Backend mit `Content-Type: text/event-stream`, `X-Accel-Buffering: no` und 15s Heartbeat-Pings (`: ping\n\n`). + - Standard HTTPS Ports (443/80), wodurch kein Proxy den Datenstrom blockieren kann. + 2. **Hub-Multiplexing:** + - `ws.Hub` verwaltet WebSocket- und SSE-Clients parallel in gemeinsamen Räumen (`rooms`, `userRooms`, `allClients`). + - Sämtliche Events (`BroadcastToAll`, `BroadcastToUser`, `BroadcastRoom`) werden transparent an beide Transportarten verteilt. + 3. **REST-Parität für Outbound CRDT & Awareness:** + - `PUT /api/v1/documents/:id/crdt`: Akzeptiert Delta-Updates (`update`) und broadcastet sie an alle Peers. + - `POST /api/v1/documents/:id/awareness` und `DELETE /api/v1/documents/:id/awareness`: Übertragen Blockfokus und Cursor ohne WebSocket. + - `POST /api/v1/events/rooms`: Erlaubt SSE-Clients dynamisches Room-Joining. + 4. **Frontend Realtime Transport Manager (`frontend/src/lib/api.ts`):** + - Automatischer 3,5s-Watchdog: Schlägt das WebSocket-Upgrade fehl oder wird Code 1006/1002 gemeldet, wechselt MindSift nahtlos auf `EventSource('/api/v1/events')`. + - Grüner/türkiser Statusindikator im Header zeigt `Live (WebSocket)` bzw. `Live (Server-Sent Events / Proxy-Stream)`. + 5. **Reverse-Proxy-Handbuch:** + - Vollständige Produktions-Konfigurationen für Nginx, Traefik, Caddy, Cloudflare und Apache unter [`reverse-proxy.md`](./reverse-proxy.md). --- diff --git a/Doku/reverse-proxy.md b/Doku/reverse-proxy.md new file mode 100644 index 0000000..1cb3e52 --- /dev/null +++ b/Doku/reverse-proxy.md @@ -0,0 +1,229 @@ +# 🛡️ Reverse-Proxy & WebServer-Konfigurationshandbuch + +Dieses Dokument beschreibt die empfohlene Konfiguration von Reverse-Proxies und Ingress-Controllern für MindSift. MindSift nutzt eine **hybride Realtime-Architektur** bestehend aus bidirektionalen **WebSockets** und einem automatischen **Server-Sent Events (SSE) Fallback**, um selbst in restriktivsten Firmennetzwerken, hinter strikten Firewalls oder bei Deep-Packet-Inspection unterbrechungsfreie Live-Synchronisation zu gewährleisten. + +--- + +## ⚡ Wichtige Anforderungen auf einen Blick + +| Protokoll | Standard-Port | Benötigte Header & Einstellungen | Zweck | +| :--- | :--- | :--- | :--- | +| **WebSockets** | 443 (WSS) / 80 (WS) | `Upgrade $http_upgrade`
`Connection "Upgrade"`
`proxy_read_timeout 3600s` | Bidirektionale Yjs-CRDT-Synchronisation & Awareness | +| **Server-Sent Events (SSE)** | 443 (HTTPS) / 80 (HTTP) | `X-Accel-Buffering: no`
`proxy_buffering off`
`proxy_cache off` | Fallback bei blockiertem WebSocket-Upgrade | +| **REST API & Uploads** | 443 (HTTPS) / 80 (HTTP) | `client_max_body_size 64M` | Notizen, Anhänge & CRDT-Zustandsblöcke | + +--- + +## 1. Nginx + +Eine typische `nginx.conf` bzw. Site-Konfiguration unter `/etc/nginx/sites-available/mindsift`: + +```nginx +# Map für sauberes Connection-Handling bei WebSocket-Upgrades +map $http_upgrade $connection_upgrade { + default upgrade; + '' close; +} + +server { + listen 80; + server_name mindsift.example.com; + return 301 https://$host$request_uri; +} + +server { + listen 443 ssl http2; + server_name mindsift.example.com; + + ssl_certificate /etc/letsencrypt/live/mindsift.example.com/fullchain.pem; + ssl_certificate_key /etc/letsencrypt/live/mindsift.example.com/privkey.pem; + + # Maximale Payload-Größe für Backups und große CRDT-Zustände + client_max_body_size 64M; + + # MindSift Frontend & REST API + location / { + proxy_pass http://127.0.0.1:8080; + proxy_http_version 1.1; + + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + } + + # 1. WebSocket Endpunkte (/api/v1/ws) + location /api/v1/ws { + proxy_pass http://127.0.0.1:8080; + proxy_http_version 1.1; + + # Upgrade-Header zwingend erforderlich + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection $connection_upgrade; + + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + # Verhindert vorzeitigen Verbindungsabbruch durch Proxy-Inaktivitäts-Timeout + proxy_read_timeout 3600s; + proxy_send_timeout 3600s; + } + + # 2. Server-Sent Events Fallback (/api/v1/events) + location /api/v1/events { + proxy_pass http://127.0.0.1:8080; + proxy_http_version 1.1; + + # Pufferung zwingend deaktivieren für sofortiges Event-Streaming + proxy_buffering off; + proxy_cache off; + proxy_set_header X-Accel-Buffering no; + + proxy_set_header Connection ''; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + proxy_read_timeout 86400s; + } +} +``` + +--- + +## 2. Traefik (Docker Compose / Ingress) + +Bei Verwendung von Traefik Version 2/3 müssen keine expliziten WebSocket-Upgrades deklariert werden, da Traefik diese transparent durchleitet. Für Server-Sent Events und Timeouts empfehlen sich Middleware-Definitionen: + +```yaml +version: '3.8' + +services: + mindsift: + image: ghcr.io/sandrodach/mindsift:latest + restart: unless-stopped + environment: + - PORT=8080 + - DB_PATH=/app/data/mindsift.db + volumes: + - ./data:/app/data + labels: + - "traefik.enable=true" + - "traefik.http.routers.mindsift.rule=Host(`mindsift.example.com`)" + - "traefik.http.routers.mindsift.entrypoints=websecure" + - "traefik.http.routers.mindsift.tls.certresolver=letsencrypt" + - "traefik.http.services.mindsift.loadbalancer.server.port=8080" + + # Verhindert Pufferung und hält persistente SSE-Streams am Leben + - "traefik.http.middlewares.mindsift-buffering.buffering.maxRequestBodyBytes=67108864" + - "traefik.http.routers.mindsift.middlewares=mindsift-buffering" +``` + +--- + +## 3. Caddy Server (Caddyfile) + +Caddy unterstützt WebSockets und HTTP/2 Server-Sent Events von Haus aus vollständig ("out-of-the-box"): + +```caddyfile +mindsift.example.com { + encode gzip zstd + + # Notizen, CRDT & Event-Stream ohne Pufferung weiterleiten + reverse_proxy 127.0.0.1:8080 { + # Streaming für SSE optimieren + flush_interval -1 + } +} +``` + +--- + +## 4. Cloudflare Proxy (Orange Cloud) + +Wird MindSift hinter dem Cloudflare CDN betrieben: +1. **WebSockets aktivieren:** + - Navigiere in das Cloudflare Dashboard: **Network** > **WebSockets** > Schalter auf **ON** stellen. +2. **Cloudflare Rocket Loader / Auto Minify:** + - Deaktiviere "Rocket Loader" für MindSift, da asynchrone Skripteingriffe die Yjs-CRDT-Initialisierung stören können. +3. **Cloudflare WAF / Bot Fight Mode:** + - Stelle sicher, dass `GET /api/v1/events` und `GET /api/v1/ws*` nicht von "Challenge"-Seiten blockiert werden. +4. **SSE-Unterstützung:** + - Cloudflare unterstützt Server-Sent Events nativ über HTTP/2 und HTTP/3. Dank der vom MindSift-Backend gesetzten Header `X-Accel-Buffering: no` und `Cache-Control: no-cache` streamt Cloudflare Echtzeit-Events verzögerungsfrei an die Clients. + +--- + +## 5. Apache 2.4 (`mod_proxy_wstunnel`) + +Unter Debian/Ubuntu: +```bash +sudo a2enmod proxy proxy_http proxy_wstunnel headers +sudo systemctl restart apache2 +``` + +VirtualHost-Konfiguration (`/etc/apache2/sites-available/mindsift.conf`): + +```apache + + ServerName mindsift.example.com + SSLEngine on + SSLCertificateFile /etc/letsencrypt/live/mindsift.example.com/fullchain.pem + SSLCertificateKeyFile /etc/letsencrypt/live/mindsift.example.com/privkey.pem + + # 1. WebSocket Tunneling + RewriteEngine On + RewriteCond %{HTTP:Upgrade} =websocket [NC] + RewriteRule /(.*) ws://127.0.0.1:8080/$1 [P,L] + RewriteCond %{HTTP:Upgrade} !=websocket [NC] + RewriteRule /(.*) http://127.0.0.1:8080/$1 [P,L] + + # 2. SSE Response Flushing + + ProxyPass "http://127.0.0.1:8080/api/v1/events" flushpackets=on + ProxyPassReverse "http://127.0.0.1:8080/api/v1/events" + + + ProxyPass / http://127.0.0.1:8080/ + ProxyPassReverse / http://127.0.0.1:8080/ + +``` + +--- + +## 🔍 Diagnose & Verifikation + +Mit den folgenden Befehlen lässt sich die Erreichbarkeit und Streaming-Funktion prüfen: + +### 1. Server-Sent Events Test (Standard-HTTPS) +```bash +curl -N -H "Accept: text/event-stream" -H "Authorization: Bearer " https://mindsift.example.com/api/v1/events +``` +*Erwartete Ausgabe:* +```text +: connected (transport: sse, id: sse_...) +data: {"event":"SSE_CONNECTED","payload":{"sseId":"sse_...","transport":"sse"},"timestamp":"..."} +: ping +``` + +### 2. WebSocket Handshake Test +```bash +curl -i -N \ + -H "Connection: Upgrade" \ + -H "Upgrade: websocket" \ + -H "Host: mindsift.example.com" \ + -H "Origin: https://mindsift.example.com" \ + -H "Sec-WebSocket-Key: SGVsbG8sIHdvcmxkIQ==" \ + -H "Sec-WebSocket-Version: 13" \ + https://mindsift.example.com/api/v1/ws?token= +``` +*Erwartete Antwort:* `101 Switching Protocols` + +--- + +## 🧭 Verwandte Dokumente +- 🔌 [WebSocket & Live-Protokoll](api/websocket.md) +- 🏗️ [Gesamtarchitektur](architecture.md) +- 🚨 [Bekannte Schwachstellen & Flaws](flaws.md) diff --git a/frontend/src/features/lists/SharedListsView.tsx b/frontend/src/features/lists/SharedListsView.tsx index 2fd00e0..8fc62f0 100644 --- a/frontend/src/features/lists/SharedListsView.tsx +++ b/frontend/src/features/lists/SharedListsView.tsx @@ -10,7 +10,7 @@ import { Hash, } from 'lucide-react'; import { ListItem, SharedList, ListShare, Team, User, WSMessage } from '../../lib/types'; -import { api, buildWebSocketURL, triggerHaptic } from '../../lib/api'; +import { api, buildWebSocketURL, triggerHaptic, createRealtimeConnection, RealtimeTransport } from '../../lib/api'; import { UnifiedShareModal } from '../../components/UnifiedShareModal'; import { useMobileActions } from '../../context/MobileActionContext'; @@ -86,6 +86,7 @@ export const SharedListsView: React.FC = ({ onRefreshGloba const [isSubmittingItem, setIsSubmittingItem] = useState(false); const [wsConnected, setWsConnected] = useState(false); + const [realtimeTransport, setRealtimeTransport] = useState('none'); const [showNewListModal, setShowNewListModal] = useState(false); const [newListTitle, setNewListTitle] = useState(''); const [newListIcon, setNewListIcon] = useState('🛒'); @@ -105,6 +106,7 @@ export const SharedListsView: React.FC = ({ onRefreshGloba const [dragOverCategoryTarget, setDragOverCategoryTarget] = useState(null); const wsRef = useRef(null); + const realtimeConnRef = useRef(null); const loadLists = async () => { try { @@ -147,83 +149,79 @@ export const SharedListsView: React.FC = ({ onRefreshGloba useEffect(() => { if (!activeList?.id) return; - if (wsRef.current) { - wsRef.current.close(); + if (realtimeConnRef.current) { + try { + realtimeConnRef.current.close(); + } catch {} + realtimeConnRef.current = null; } - let ws: WebSocket | null = null; - try { - ws = new WebSocket(buildWebSocketURL(activeList.id)); - wsRef.current = ws; - - ws.onopen = () => { - setWsConnected(true); - }; - - ws.onclose = () => { - setWsConnected(false); - }; - - ws.onmessage = (evt) => { - try { - const msg: WSMessage = JSON.parse(evt.data); - const msgListId = msg.listId || (msg as any).list_id; - if (msg.event === 'ITEM_CREATED' && msgListId) { - const newItem: ListItem = msg.payload; - if (!newItem || !newItem.id) return; - setLists((prev) => - prev.map((l) => { - if (l.id !== msgListId) return l; - if (l.items.some((i) => i.id === newItem.id)) return l; - return { ...l, items: deduplicateItems([newItem, ...l.items]) }; - }) - ); - } else if (msg.event === 'ITEM_TOGGLED' && msgListId) { - const updatedItem: ListItem = msg.payload; - setLists((prev) => - prev.map((l) => { - if (l.id !== msgListId) return l; - return { - ...l, - items: l.items.map((i) => (i.id === updatedItem.id ? updatedItem : i)), - }; - }) - ); - } else if (msg.event === 'ITEM_DELETED' && msgListId) { - const deletedId = msg.payload?.itemId; - setLists((prev) => - prev.map((l) => { - if (l.id !== msgListId) return l; - return { ...l, items: l.items.filter((i) => i.id !== deletedId) }; - }) - ); - } else if (msg.event === 'LIST_CLEARED' && msgListId) { - setLists((prev) => - prev.map((l) => { - if (l.id !== msgListId) return l; - return { ...l, items: l.items.filter((i) => !i.isCompleted) }; - }) - ); - } else if ( - msg.event === 'LIST_CREATED' || - msg.event === 'LIST_DELETED' || - msg.event === 'LIST_SHARED' || - msg.event === 'LIST_UNSHARED' || - msg.event === 'ITEM_REORDERED' || - msg.event === 'LIST_UPDATED' - ) { - loadLists(); - } - } catch { - // ignore malformed frames + const conn = createRealtimeConnection({ + listId: activeList.id, + onStatusChange: (status) => { + setWsConnected(status.connected); + setRealtimeTransport(status.transport); + }, + onMessage: (msg: WSMessage) => { + const msgListId = msg.listId || (msg as any).list_id; + if (msg.event === 'ITEM_CREATED' && msgListId) { + const newItem: ListItem = msg.payload; + if (!newItem || !newItem.id) return; + setLists((prev) => + prev.map((l) => { + if (l.id !== msgListId) return l; + if (l.items.some((i) => i.id === newItem.id)) return l; + return { ...l, items: deduplicateItems([newItem, ...l.items]) }; + }) + ); + } else if (msg.event === 'ITEM_TOGGLED' && msgListId) { + const updatedItem: ListItem = msg.payload; + setLists((prev) => + prev.map((l) => { + if (l.id !== msgListId) return l; + return { + ...l, + items: l.items.map((i) => (i.id === updatedItem.id ? updatedItem : i)), + }; + }) + ); + } else if (msg.event === 'ITEM_DELETED' && msgListId) { + const deletedId = msg.payload?.itemId; + setLists((prev) => + prev.map((l) => { + if (l.id !== msgListId) return l; + return { ...l, items: l.items.filter((i) => i.id !== deletedId) }; + }) + ); + } else if (msg.event === 'LIST_CLEARED' && msgListId) { + setLists((prev) => + prev.map((l) => { + if (l.id !== msgListId) return l; + return { ...l, items: l.items.filter((i) => !i.isCompleted) }; + }) + ); + } else if ( + msg.event === 'LIST_CREATED' || + msg.event === 'LIST_DELETED' || + msg.event === 'LIST_SHARED' || + msg.event === 'LIST_UNSHARED' || + msg.event === 'ITEM_REORDERED' || + msg.event === 'LIST_UPDATED' + ) { + loadLists(); } - }; - } catch { - setWsConnected(false); - } + }, + }); + + realtimeConnRef.current = conn; return () => { - if (ws) ws.close(); + if (realtimeConnRef.current) { + try { + realtimeConnRef.current.close(); + } catch {} + realtimeConnRef.current = null; + } }; }, [activeList?.id]); @@ -821,12 +819,20 @@ export const SharedListsView: React.FC = ({ onRefreshGloba )}
@@ -927,12 +933,20 @@ export const SharedListsView: React.FC = ({ onRefreshGloba {/* Live Sync Status */}
diff --git a/frontend/src/features/workspace/WorkspaceView.tsx b/frontend/src/features/workspace/WorkspaceView.tsx index 2b9161e..4b46cf6 100644 --- a/frontend/src/features/workspace/WorkspaceView.tsx +++ b/frontend/src/features/workspace/WorkspaceView.tsx @@ -50,7 +50,7 @@ import { Radio, } from 'lucide-react'; import { BlockType, Document, DocumentBlock, DocBlocksPatchEvent, DocLockInfo, DocShare, FolderShare, SharedFolder, TagDetails, TagSummary, Team, User } from '../../lib/types'; -import { api, getToken, triggerHaptic, onWebSocketStatus, isWebSocketConnected, buildWebSocketURL } from '../../lib/api'; +import { api, getToken, triggerHaptic, onWebSocketStatus, isWebSocketConnected, buildWebSocketURL, createRealtimeConnection, RealtimeTransport } from '../../lib/api'; import { CodeBlockItem } from './CodeBlockItem'; import { UserAutocomplete } from '../UserAutocomplete'; import { UnifiedShareModal } from '../../components/UnifiedShareModal'; @@ -130,11 +130,13 @@ export const WorkspaceView: React.FC = ({ const [showIconPicker, setShowIconPicker] = useState(false); const [showWikiLinkMenu, setShowWikiLinkMenu] = useState(false); - // Dedicated WebSocket connection & status for live indicator + // Dedicated WebSocket/SSE connection & status for live indicator const wsRef = useRef(null); + const realtimeConnRef = useRef(null); const wsReconnectTimerRef = useRef(null); const wsReconnectDelayRef = useRef(2000); const [wsConnected, setWsConnected] = useState(false); + const [realtimeTransport, setRealtimeTransport] = useState('none'); const [wsDisconnectReason, setWsDisconnectReason] = useState(null); // Folder state (synchronized with server) @@ -431,22 +433,33 @@ export const WorkspaceView: React.FC = ({ }; const broadcastAwareness = (blockId: string | null) => { - if (!selectedDoc || !wsRef.current || wsRef.current.readyState !== WebSocket.OPEN) return; - try { - wsRef.current.send( - JSON.stringify({ - event: 'CRDT_AWARENESS', - docId: selectedDoc.id, - payload: { - userId: currentUser?.id, - userName: currentUser?.displayName || currentUser?.username || 'Kollege', - activeBlockId: blockId, - color: getCollaboratorColor(currentUser?.id || ''), - lastActive: Date.now(), - }, - }) - ); - } catch {} + if (!selectedDoc) return; + const payload = { + userId: currentUser?.id, + userName: currentUser?.displayName || currentUser?.username || 'Kollege', + activeBlockId: blockId, + color: getCollaboratorColor(currentUser?.id || ''), + lastActive: Date.now(), + }; + if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { + realtimeConnRef.current.send({ + event: 'CRDT_AWARENESS', + docId: selectedDoc.id, + payload, + }); + } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { + try { + wsRef.current.send( + JSON.stringify({ + event: 'CRDT_AWARENESS', + docId: selectedDoc.id, + payload, + }) + ); + } catch {} + } else { + api.sendDocumentAwareness(selectedDoc.id, payload).catch(() => {}); + } }; const handleBlockFocus = (blockIdOrEvent?: string | React.FocusEvent) => { @@ -779,10 +792,14 @@ export const WorkspaceView: React.FC = ({ if (prevId && prevId !== selectedDoc?.id) { flushPendingSave(); releaseMyLock(prevId); - if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { + if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { + realtimeConnRef.current.send({ event: 'CRDT_LEAVE', docId: prevId }); + } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { try { wsRef.current.send(JSON.stringify({ event: 'CRDT_LEAVE', docId: prevId })); } catch {} + } else { + api.clearDocumentAwareness(prevId).catch(() => {}); } if (crdtSessionRef.current) { crdtSessionRef.current.destroy(); @@ -797,7 +814,13 @@ export const WorkspaceView: React.FC = ({ const session = new DocumentCRDTSession(targetDoc.id); session.initFromBlocks(targetDoc.blocks || []); session.onLocalUpdate = (b64Update) => { - if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { + if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { + realtimeConnRef.current.send({ + event: 'CRDT_UPDATE', + docId: targetDoc.id, + payload: { update: b64Update }, + }); + } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { try { wsRef.current.send( JSON.stringify({ @@ -807,6 +830,8 @@ export const WorkspaceView: React.FC = ({ }) ); } catch {} + } else { + api.saveDocumentCRDT(targetDoc.id, { update: b64Update }).catch(() => {}); } }; session.onRemoteChange = (crdtBlocks) => { @@ -815,7 +840,9 @@ export const WorkspaceView: React.FC = ({ }; crdtSessionRef.current = session; - if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { + if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { + realtimeConnRef.current.send({ event: 'CRDT_JOIN', docId: targetDoc.id }); + } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { try { wsRef.current.send(JSON.stringify({ event: 'CRDT_JOIN', docId: targetDoc.id })); } catch {} @@ -857,10 +884,14 @@ export const WorkspaceView: React.FC = ({ return () => { flushPendingSave(); if (currentDocIdRef.current) { - if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { + if (realtimeConnRef.current && realtimeConnRef.current.isConnected()) { + realtimeConnRef.current.send({ event: 'CRDT_LEAVE', docId: currentDocIdRef.current }); + } else if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { try { wsRef.current.send(JSON.stringify({ event: 'CRDT_LEAVE', docId: currentDocIdRef.current })); } catch {} + } else { + api.clearDocumentAwareness(currentDocIdRef.current).catch(() => {}); } if (crdtSessionRef.current) { crdtSessionRef.current.destroy(); @@ -1097,87 +1128,26 @@ export const WorkspaceView: React.FC = ({ } }; - const connect = () => { - if (!isMounted) return; - const token = getToken(); - if (!token) { - setWsConnected(false); - return; - } - - if (wsRef.current) { - try { - wsRef.current.close(); - } catch {} - wsRef.current = null; - } - - try { - const url = buildWebSocketURL(); - const ws = new WebSocket(url); - wsRef.current = ws; - - ws.onopen = () => { - if (!isMounted) return; - wsReconnectDelayRef.current = 2000; + const conn = createRealtimeConnection({ + docId: selectedDocIdRef.current || undefined, + onMessage: handleWsMsg, + onStatusChange: (status) => { + if (!isMounted) return; + setWsConnected(status.connected); + setRealtimeTransport(status.transport); + if (status.connected) { setWsDisconnectReason(null); - setWsConnected(true); if (selectedDocIdRef.current) { - try { - ws.send(JSON.stringify({ event: 'CRDT_JOIN', docId: selectedDocIdRef.current })); - } catch {} + conn.send({ event: 'CRDT_JOIN', docId: selectedDocIdRef.current }); } api.getFolders().then(setCustomFolders).catch(() => {}); api.getSharedFolders().then(setSharedFolders).catch(() => {}); - }; - - ws.onclose = (evt: CloseEvent) => { - if (!isMounted) return; - setWsConnected(false); - const reasonText = - evt.reason || - (evt.code === 1006 - ? 'Verbindung durch Firmen-Proxy oder Netzwerk getrennt (Code 1006)' - : evt.code === 1008 - ? 'Richtlinien-Verletzung durch Proxy/Firewall (Code 1008)' - : evt.code === 1002 - ? 'Protokollfehler beim Handshake (Code 1002)' - : evt.code === 1000 - ? 'Normal geschlossen (Code 1000)' - : `Code ${evt.code}`); - console.warn('[MindSift Workspace WS] getrennt:', evt.code, evt.reason || reasonText); - setWsDisconnectReason(reasonText); - clearTimeout(wsReconnectTimerRef.current); - const delay = wsReconnectDelayRef.current; - wsReconnectDelayRef.current = Math.min(30000, Math.round(wsReconnectDelayRef.current * 1.5)); - wsReconnectTimerRef.current = setTimeout(connect, delay); - }; - - ws.onerror = (e) => { - if (!isMounted) return; - console.warn('[MindSift Workspace WS] Fehler aufgetreten:', e); - setWsConnected(false); - }; - - ws.onmessage = (evt) => { - try { - const msg = JSON.parse(evt.data); - handleWsMsg(msg); - } catch {} - }; - } catch (err: any) { - if (!isMounted) return; - setWsConnected(false); - const errText = err?.message || 'Verbindungsfehler'; - setWsDisconnectReason(errText); - clearTimeout(wsReconnectTimerRef.current); - const delay = wsReconnectDelayRef.current; - wsReconnectDelayRef.current = Math.min(30000, Math.round(wsReconnectDelayRef.current * 1.5)); - wsReconnectTimerRef.current = setTimeout(connect, delay); - } - }; - - connect(); + } else { + setWsDisconnectReason(status.reason || 'Verbindung getrennt'); + } + }, + }); + realtimeConnRef.current = conn; // Also synchronize with global WebSocket status const unsubGlobalStatus = onWebSocketStatus((connected) => { @@ -1199,12 +1169,11 @@ export const WorkspaceView: React.FC = ({ return () => { isMounted = false; unsubGlobalStatus(); - clearTimeout(wsReconnectTimerRef.current); - if (wsRef.current) { + if (realtimeConnRef.current) { try { - wsRef.current.close(); + realtimeConnRef.current.close(); } catch {} - wsRef.current = null; + realtimeConnRef.current = null; } window.removeEventListener('mindsift:ws', handleGlobalCustomEvent); }; @@ -4058,12 +4027,16 @@ export const WorkspaceView: React.FC = ({
@@ -4098,12 +4071,16 @@ export const WorkspaceView: React.FC = ({
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index d4f71ed..be47083 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -606,10 +606,31 @@ export const api = { return request<{ docId: string; state?: string; updates?: string[] }>(`/api/v1/documents/${id}/crdt`); }, - async saveDocumentCRDT(id: string, state: string): Promise<{ success: boolean }> { + async saveDocumentCRDT(id: string, payload: string | { state?: string; update?: string }): Promise<{ success: boolean }> { + const body = typeof payload === 'string' ? { state: payload } : payload; return request<{ success: boolean }>(`/api/v1/documents/${id}/crdt`, { method: 'PUT', - body: JSON.stringify({ state }), + body: JSON.stringify(body), + }); + }, + + async sendDocumentAwareness(id: string, payload: any): Promise<{ success: boolean }> { + return request<{ success: boolean }>(`/api/v1/documents/${id}/awareness`, { + method: 'POST', + body: JSON.stringify(payload), + }); + }, + + async clearDocumentAwareness(id: string): Promise<{ success: boolean }> { + return request<{ success: boolean }>(`/api/v1/documents/${id}/awareness`, { + method: 'DELETE', + }); + }, + + async joinSSERoom(sseId: string, room: string, action: 'join' | 'leave' = 'join'): Promise<{ success: boolean }> { + return request<{ success: boolean }>('/api/v1/events/rooms', { + method: 'POST', + body: JSON.stringify({ sseId, room, action }), }); }, @@ -1053,11 +1074,17 @@ export function getOfflineQueue(): Thought[] { } } +export type RealtimeTransport = 'ws' | 'sse' | 'none'; + +let activeTransport: RealtimeTransport = 'none'; let activeGlobalWS: WebSocket | null = null; +let activeGlobalSSE: EventSource | null = null; +let activeSSEId: string | null = null; let wsReconnectTimer: any = null; +let wsConnectionWatchdog: any = null; let wsListeners: ((msg: WSMessage) => void)[] = []; -let wsStatusListeners: ((connected: boolean) => void)[] = []; -let isWSConnected = false; +let realtimeStatusListeners: ((connected: boolean, transport: RealtimeTransport) => void)[] = []; +let isRealtimeConnectedState = false; let manualDisconnect = false; let wsReconnectDelay = 2000; @@ -1070,28 +1097,116 @@ export function buildWebSocketURL(listId?: string): string { return `${protocol}//${window.location.host}/api/v1/ws?token=${encodeURIComponent(token)}`; } +export function buildSSEURL(listId?: string, docId?: string, rooms?: string[]): string { + const token = getToken() || ''; + const params = new URLSearchParams(); + if (token) params.set('token', token); + if (listId) params.set('listId', listId); + if (docId) params.set('docId', docId); + if (rooms && rooms.length > 0) params.set('rooms', rooms.join(',')); + return `/api/v1/events?${params.toString()}`; +} + export function isWebSocketConnected(): boolean { - return isWSConnected && activeGlobalWS !== null && activeGlobalWS.readyState === WebSocket.OPEN; + return activeTransport === 'ws' && activeGlobalWS !== null && activeGlobalWS.readyState === WebSocket.OPEN; +} + +export function isRealtimeConnected(): boolean { + return isRealtimeConnectedState; +} + +export function getRealtimeTransport(): RealtimeTransport { + return activeTransport; +} + +export function getSSEId(): string | null { + return activeSSEId; } export function onWebSocketStatus(callback: (connected: boolean) => void): () => void { - wsStatusListeners.push(callback); - callback(isWebSocketConnected()); + const wrapper = (connected: boolean) => callback(connected); + realtimeStatusListeners.push(wrapper); + callback(isRealtimeConnected()); return () => { - wsStatusListeners = wsStatusListeners.filter((fn) => fn !== callback); + realtimeStatusListeners = realtimeStatusListeners.filter((fn) => fn !== wrapper); }; } -function notifyStatusChange(connected: boolean) { - isWSConnected = connected; - wsStatusListeners.forEach((fn) => { +export function onRealtimeStatus(callback: (connected: boolean, transport: RealtimeTransport) => void): () => void { + realtimeStatusListeners.push(callback); + callback(isRealtimeConnected(), activeTransport); + return () => { + realtimeStatusListeners = realtimeStatusListeners.filter((fn) => fn !== callback); + }; +} + +function notifyRealtimeStatusChange(connected: boolean, transport: RealtimeTransport) { + isRealtimeConnectedState = connected; + activeTransport = transport; + realtimeStatusListeners.forEach((fn) => { try { - fn(connected); + fn(connected, transport); } catch (e) { - console.error('WS status listener error', e); + console.error('Realtime status listener error', e); } }); - window.dispatchEvent(new CustomEvent('mindsift:ws:status', { detail: { connected } })); + window.dispatchEvent(new CustomEvent('mindsift:ws:status', { detail: { connected, transport } })); +} + +export function connectSSE(rooms?: string[]): void { + if (activeGlobalSSE) { + try { activeGlobalSSE.close(); } catch {} + activeGlobalSSE = null; + } + + try { + const url = buildSSEURL(undefined, undefined, rooms); + const es = new EventSource(url); + activeGlobalSSE = es; + + es.onopen = () => { + console.info('[MindSift Realtime] Server-Sent Events (SSE) verbunden (Proxy-Fallback aktiv).'); + wsReconnectDelay = 2000; + notifyRealtimeStatusChange(true, 'sse'); + }; + + es.onmessage = (event) => { + try { + const msg = JSON.parse(event.data) as WSMessage; + if (msg.event === 'SSE_CONNECTED') { + activeSSEId = msg.payload?.sseId || null; + } + wsListeners.forEach((fn) => { + try { + fn(msg); + } catch (e) { + console.error('SSE listener error', e); + } + }); + window.dispatchEvent(new CustomEvent('mindsift:ws', { detail: msg })); + } catch (err) { + console.error('Failed to parse SSE message', err); + } + }; + + es.onerror = (e) => { + console.warn('[MindSift Realtime] SSE getrennt oder Fehler aufgetreten:', e); + if (activeTransport === 'sse') { + notifyRealtimeStatusChange(false, 'none'); + } + try { es.close(); } catch {} + activeGlobalSSE = null; + if (!manualDisconnect && getToken()) { + clearTimeout(wsReconnectTimer); + wsReconnectTimer = setTimeout(() => { + connectWebSocket(); + }, 5000); + } + }; + } catch (err) { + console.error('Failed to initialize EventSource SSE', err); + notifyRealtimeStatusChange(false, 'none'); + } } export function connectWebSocket(): void { @@ -1103,14 +1218,31 @@ export function connectWebSocket(): void { return; } + clearTimeout(wsConnectionWatchdog); + try { const url = buildWebSocketURL(); const ws = new WebSocket(url); activeGlobalWS = ws; + // 3.5s watchdog for corporate proxies silently stripping Upgrade header + wsConnectionWatchdog = setTimeout(() => { + if (ws && ws.readyState !== WebSocket.OPEN) { + console.warn('[MindSift Realtime] WebSocket connection timeout (Proxy Upgrade Block). Wechsle automatisch zu SSE...'); + try { ws.close(); } catch {} + activeGlobalWS = null; + connectSSE(); + } + }, 3500); + ws.onopen = () => { + clearTimeout(wsConnectionWatchdog); + if (activeGlobalSSE) { + try { activeGlobalSSE.close(); } catch {} + activeGlobalSSE = null; + } wsReconnectDelay = 2000; - notifyStatusChange(true); + notifyRealtimeStatusChange(true, 'ws'); }; ws.onmessage = (event) => { @@ -1130,10 +1262,22 @@ export function connectWebSocket(): void { }; ws.onclose = (evt: CloseEvent) => { + clearTimeout(wsConnectionWatchdog); activeGlobalWS = null; console.warn('[MindSift Global WS] Closed with code:', evt?.code, 'reason:', evt?.reason); - notifyStatusChange(false); - if (!manualDisconnect && getToken()) { + + const isProxyIssue = evt?.code === 1006 || evt?.code === 1002 || evt?.code === 1008; + if (isProxyIssue && !manualDisconnect && getToken()) { + console.warn('[MindSift Realtime] Proxy/Firewall Inkompatibilität erkannt (Code ' + evt.code + '). Wechsle zu SSE...'); + connectSSE(); + return; + } + + if (activeTransport === 'ws') { + notifyRealtimeStatusChange(false, 'none'); + } + + if (!manualDisconnect && getToken() && activeTransport !== 'sse') { clearTimeout(wsReconnectTimer); const delay = wsReconnectDelay; wsReconnectDelay = Math.min(30000, Math.round(wsReconnectDelay * 1.5)); @@ -1144,27 +1288,28 @@ export function connectWebSocket(): void { }; ws.onerror = (e) => { + clearTimeout(wsConnectionWatchdog); console.warn('[MindSift Global WS] Error:', e); - notifyStatusChange(false); try { ws.close(); } catch {} }; } catch (e) { - console.error('Failed to connect WebSocket', e); - notifyStatusChange(false); + clearTimeout(wsConnectionWatchdog); + console.warn('Failed to connect WebSocket, switching to SSE:', e); + connectSSE(); } } // Ensure reconnect when browser recovers from sleep or tab visibility if (typeof window !== 'undefined') { window.addEventListener('online', () => { - if (getToken() && !isWebSocketConnected()) { + if (getToken() && !isRealtimeConnected()) { connectWebSocket(); } }); document.addEventListener('visibilitychange', () => { - if (document.visibilityState === 'visible' && getToken() && !isWebSocketConnected()) { + if (document.visibilityState === 'visible' && getToken() && !isRealtimeConnected()) { connectWebSocket(); } }); @@ -1173,18 +1318,47 @@ if (typeof window !== 'undefined') { export function disconnectWebSocket(): void { manualDisconnect = true; clearTimeout(wsReconnectTimer); + clearTimeout(wsConnectionWatchdog); if (activeGlobalWS) { try { activeGlobalWS.close(); } catch {} activeGlobalWS = null; } - notifyStatusChange(false); + if (activeGlobalSSE) { + try { + activeGlobalSSE.close(); + } catch {} + activeGlobalSSE = null; + } + notifyRealtimeStatusChange(false, 'none'); } export function sendWebSocketMessage(msg: any): void { if (activeGlobalWS && activeGlobalWS.readyState === WebSocket.OPEN) { activeGlobalWS.send(typeof msg === 'string' ? msg : JSON.stringify(msg)); + return; + } + + // SSE Fallback: dispatch via REST endpoints + try { + const parsed = typeof msg === 'string' ? JSON.parse(msg) : msg; + if (parsed.event === 'CRDT_UPDATE' && parsed.docId) { + api.saveDocumentCRDT(parsed.docId, { update: parsed.payload?.update }); + } else if (parsed.event === 'CRDT_AWARENESS' && parsed.docId) { + api.sendDocumentAwareness(parsed.docId, parsed.payload); + } else if (parsed.event === 'CRDT_LEAVE' && parsed.docId) { + api.clearDocumentAwareness(parsed.docId); + if (activeSSEId) api.joinSSERoom(activeSSEId, `doc:${parsed.docId}`, 'leave'); + } else if (parsed.event === 'CRDT_JOIN' && parsed.docId) { + if (activeSSEId) api.joinSSERoom(activeSSEId, `doc:${parsed.docId}`, 'join'); + } else if (parsed.event === 'LOCK_DOC' && parsed.docId) { + api.lockDocument(parsed.docId); + } else if (parsed.event === 'UNLOCK_DOC' && parsed.docId) { + api.unlockDocument(parsed.docId); + } + } catch (err) { + console.error('Error dispatching fallback REST message for SSE:', err); } } @@ -1194,3 +1368,205 @@ export function onWebSocketMessage(callback: (msg: WSMessage) => void): () => vo wsListeners = wsListeners.filter((fn) => fn !== callback); }; } + +export function createRealtimeConnection(options: { + listId?: string; + docId?: string; + rooms?: string[]; + onMessage?: (msg: WSMessage) => void; + onStatusChange?: (status: { connected: boolean; transport: RealtimeTransport; reason?: string }) => void; +}) { + let isClosed = false; + let currentTransport: RealtimeTransport = 'none'; + let ws: WebSocket | null = null; + let es: EventSource | null = null; + let watchdogTimer: any = null; + let reconnectTimer: any = null; + let reconnectDelay = 2000; + let sseId: string | null = null; + + const notify = (connected: boolean, transport: RealtimeTransport, reason?: string) => { + currentTransport = transport; + options.onStatusChange?.({ connected, transport, reason }); + }; + + const connectSSE = () => { + if (isClosed) return; + if (es) { + try { es.close(); } catch {} + es = null; + } + + try { + const allRooms = [...(options.rooms || [])]; + if (options.docId) allRooms.push(`doc:${options.docId}`); + if (options.listId) allRooms.push(options.listId); + const url = buildSSEURL(options.listId, options.docId, allRooms); + es = new EventSource(url); + + es.onopen = () => { + if (isClosed) return; + reconnectDelay = 2000; + notify(true, 'sse'); + }; + + es.onmessage = (evt) => { + if (isClosed) return; + try { + const msg = JSON.parse(evt.data) as WSMessage; + if (msg.event === 'SSE_CONNECTED') { + sseId = msg.payload?.sseId || null; + } + options.onMessage?.(msg); + } catch (e) { + console.error('SSE JSON error', e); + } + }; + + es.onerror = (e) => { + if (isClosed) return; + notify(false, 'none', 'SSE-Verbindungsabbruch'); + try { es?.close(); } catch {} + es = null; + clearTimeout(reconnectTimer); + reconnectTimer = setTimeout(() => { + if (!isClosed) connectWS(); + }, 5000); + }; + } catch (e: any) { + notify(false, 'none', e?.message || 'SSE Initialisierungsfehler'); + } + }; + + const connectWS = () => { + if (isClosed) return; + const token = getToken(); + if (!token) { + notify(false, 'none', 'Kein Authentifizierungs-Token vorhanden'); + return; + } + + if (ws) { + try { ws.close(); } catch {} + ws = null; + } + + clearTimeout(watchdogTimer); + + try { + const url = buildWebSocketURL(options.listId); + ws = new WebSocket(url); + + // 3.5s watchdog for corporate proxies silently discarding Upgrade header + watchdogTimer = setTimeout(() => { + if (ws && ws.readyState !== WebSocket.OPEN && !isClosed) { + console.warn('[MindSift Realtime] WebSocket connection timeout (Proxy Upgrade block). Falling back to SSE...'); + try { ws.close(); } catch {} + ws = null; + connectSSE(); + } + }, 3500); + + ws.onopen = () => { + if (isClosed) return; + clearTimeout(watchdogTimer); + reconnectDelay = 2000; + notify(true, 'ws'); + if (options.docId) { + try { + ws?.send(JSON.stringify({ event: 'CRDT_JOIN', docId: options.docId })); + } catch {} + } + }; + + ws.onmessage = (evt) => { + if (isClosed) return; + try { + const msg = JSON.parse(evt.data) as WSMessage; + options.onMessage?.(msg); + } catch (e) { + console.error('WS JSON error', e); + } + }; + + ws.onclose = (evt: CloseEvent) => { + if (isClosed) return; + clearTimeout(watchdogTimer); + ws = null; + + const isProxyError = evt?.code === 1006 || evt?.code === 1002 || evt?.code === 1008; + if (isProxyError) { + console.warn(`[MindSift Realtime] Proxy WebSocket rejection detected (Code ${evt.code}). Falling back to SSE...`); + connectSSE(); + return; + } + + const reason = evt.reason || (evt.code === 1000 ? 'Normal geschlossen' : `Code ${evt.code}`); + notify(false, 'none', reason); + + clearTimeout(reconnectTimer); + const delay = reconnectDelay; + reconnectDelay = Math.min(30000, Math.round(reconnectDelay * 1.5)); + reconnectTimer = setTimeout(connectWS, delay); + }; + + ws.onerror = (e) => { + if (isClosed) return; + clearTimeout(watchdogTimer); + try { ws?.close(); } catch {} + }; + } catch (e: any) { + clearTimeout(watchdogTimer); + console.warn('[MindSift Realtime] WebSocket constructor failed, trying SSE fallback:', e); + connectSSE(); + } + }; + + connectWS(); + + return { + getTransport: () => currentTransport, + isConnected: () => currentTransport !== 'none', + send: (msg: any) => { + const str = typeof msg === 'string' ? msg : JSON.stringify(msg); + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(str); + return; + } + // SSE Fallback: dispatch via REST endpoints + try { + const parsed = typeof msg === 'string' ? JSON.parse(msg) : msg; + if (parsed.event === 'CRDT_UPDATE' && parsed.docId) { + api.saveDocumentCRDT(parsed.docId, { update: parsed.payload?.update }); + } else if (parsed.event === 'CRDT_AWARENESS' && parsed.docId) { + api.sendDocumentAwareness(parsed.docId, parsed.payload); + } else if (parsed.event === 'CRDT_LEAVE' && parsed.docId) { + api.clearDocumentAwareness(parsed.docId); + if (sseId) api.joinSSERoom(sseId, `doc:${parsed.docId}`, 'leave'); + } else if (parsed.event === 'CRDT_JOIN' && parsed.docId) { + if (sseId) api.joinSSERoom(sseId, `doc:${parsed.docId}`, 'join'); + } else if (parsed.event === 'LOCK_DOC' && parsed.docId) { + api.lockDocument(parsed.docId); + } else if (parsed.event === 'UNLOCK_DOC' && parsed.docId) { + api.unlockDocument(parsed.docId); + } + } catch (err) { + console.error('Error dispatching fallback REST message for connection:', err); + } + }, + close: () => { + isClosed = true; + clearTimeout(watchdogTimer); + clearTimeout(reconnectTimer); + if (ws) { + try { ws.close(); } catch {} + ws = null; + } + if (es) { + try { es.close(); } catch {} + es = null; + } + notify(false, 'none'); + }, + }; +} diff --git a/internal/api/server.go b/internal/api/server.go index 47721d8..4e6e221 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -69,6 +69,8 @@ func (s *Server) Router() http.Handler { r.Get("/api/v1/ws", s.handleWebSocket) r.Get("/api/v1/ws/spaces/{spaceId}", s.handleWebSocket) r.Get("/api/v1/ws/lists/{listId}", s.handleWebSocket) + r.Get("/api/v1/events", s.handleSSE) + r.Post("/api/v1/events/rooms", s.handleSSERooms) // Authenticated API Routes r.Group(func(r chi.Router) { @@ -148,6 +150,8 @@ func (s *Server) Router() http.Handler { // Document CRDT Realtime Sync r.Get("/api/v1/documents/{id}/crdt", s.handleGetDocCRDT) r.Put("/api/v1/documents/{id}/crdt", s.handleSaveDocCRDT) + r.Post("/api/v1/documents/{id}/awareness", s.handlePostDocAwareness) + r.Delete("/api/v1/documents/{id}/awareness", s.handleDeleteDocAwareness) // Shared Lists & Realtime Collaboration r.Get("/api/v1/lists", s.handleGetAllLists) @@ -1423,12 +1427,29 @@ func (s *Server) handleSaveDocCRDT(w http.ResponseWriter, r *http.Request) { return } var req struct { - State string `json:"state"` + State string `json:"state"` + Update string `json:"update"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "Ungültiger Request-Body") return } + if req.Update != "" { + raw, err := base64.StdEncoding.DecodeString(req.Update) + if err != nil { + writeError(w, http.StatusBadRequest, "Ungültiges Base64 Format für Update") + return + } + if err := s.store.AppendDocumentCRDTUpdate(docID, raw, userID); err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + s.wsHub.BroadcastRoom("doc:"+docID, "CRDT_UPDATE", map[string]interface{}{ + "docId": docID, + "update": req.Update, + "fromUserId": userID, + }, nil) + } if req.State != "" { raw, err := base64.StdEncoding.DecodeString(req.State) if err != nil { @@ -1443,6 +1464,37 @@ func (s *Server) handleSaveDocCRDT(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]bool{"success": true}) } +func (s *Server) handlePostDocAwareness(w http.ResponseWriter, r *http.Request) { + userID := auth.GetUserID(r.Context()) + docID := chi.URLParam(r, "id") + doc, err := s.store.GetDocumentByID(userID, docID) + if err != nil { + writeError(w, http.StatusNotFound, "Dokument nicht gefunden") + return + } + _ = doc + + var payload json.RawMessage + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + writeError(w, http.StatusBadRequest, "Ungültiges JSON") + return + } + + s.wsHub.BroadcastRoom("doc:"+docID, "CRDT_AWARENESS", payload, nil) + writeJSON(w, http.StatusOK, map[string]bool{"success": true}) +} + +func (s *Server) handleDeleteDocAwareness(w http.ResponseWriter, r *http.Request) { + userID := auth.GetUserID(r.Context()) + docID := chi.URLParam(r, "id") + + s.wsHub.BroadcastRoom("doc:"+docID, "CRDT_AWARENESS_LEAVE", map[string]interface{}{ + "docId": docID, + "userId": userID, + }, nil) + writeJSON(w, http.StatusOK, map[string]bool{"success": true}) +} + // --- REALTIME LISTS & SHARING --- func (s *Server) handleGetAllLists(w http.ResponseWriter, r *http.Request) { @@ -1712,6 +1764,68 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) { s.wsHub.ServeWS(w, r, roomID, userID) } +func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { + roomID := r.URL.Query().Get("listId") + if roomID == "" { + roomID = r.URL.Query().Get("spaceId") + } + + token := r.URL.Query().Get("token") + if token == "" { + if authHdr := r.Header.Get("Authorization"); strings.HasPrefix(authHdr, "Bearer ") { + token = strings.TrimPrefix(authHdr, "Bearer ") + } + } + userID := "anonymous" + if token != "" { + if claims, err := auth.ValidateToken(s.cfg.JWTSecret, token); err == nil { + userID = claims.UserID + } + } + + var initialRooms []string + if roomsParam := r.URL.Query().Get("rooms"); roomsParam != "" { + for _, rm := range strings.Split(roomsParam, ",") { + rm = strings.TrimSpace(rm) + if rm != "" { + initialRooms = append(initialRooms, rm) + } + } + } + if docID := r.URL.Query().Get("docId"); docID != "" { + initialRooms = append(initialRooms, "doc:"+strings.TrimSpace(docID)) + } + + s.wsHub.ServeSSE(w, r, roomID, userID, initialRooms) +} + +func (s *Server) handleSSERooms(w http.ResponseWriter, r *http.Request) { + var req struct { + SSEID string `json:"sseId"` + Room string `json:"room"` + DocID string `json:"docId"` + Action string `json:"action"` // "join" or "leave" + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "Ungültiges JSON") + return + } + if req.Room == "" && req.DocID != "" { + req.Room = "doc:" + req.DocID + } + if req.SSEID == "" || req.Room == "" { + writeError(w, http.StatusBadRequest, "sseId und room erforderlich") + return + } + + if req.Action == "leave" { + s.wsHub.LeaveSSERoomByID(req.SSEID, req.Room) + } else { + s.wsHub.JoinSSERoomByID(req.SSEID, req.Room) + } + writeJSON(w, http.StatusOK, map[string]bool{"success": true}) +} + // --- API KEYS & WEBHOOK INGESTION --- func (s *Server) handleGetAPIKeys(w http.ResponseWriter, r *http.Request) { diff --git a/internal/api/server_test.go b/internal/api/server_test.go index e7fc354..82f6aed 100644 --- a/internal/api/server_test.go +++ b/internal/api/server_test.go @@ -2,6 +2,7 @@ package api import ( "bytes" + "context" "encoding/json" "net/http" "net/http/httptest" @@ -9,6 +10,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/sandrodach/mindsift/internal/auth" "github.com/sandrodach/mindsift/internal/config" @@ -908,3 +910,108 @@ func TestDocumentCRDTEndpoints(t *testing.T) { t.Fatalf("expected state 'AQIDBA==', got '%s'", crdtResp.State) } } + +func TestSSEAndAwarenessEndpoints(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test_sse.db") + store, err := storage.NewStore(dbPath) + if err != nil { + t.Fatalf("failed to init store: %v", err) + } + defer store.Close() + + hub := ws.NewHub() + hub.SetCRDTStore(store) + cfg := &config.Config{ + JWTSecret: "test-jwt-secret-key", + AllowRegistration: false, + } + srv := NewServer(cfg, store, hub, nil) + router := srv.Router() + + // 1. Setup user + setupBody := []byte(`{ + "email": "sse_user@example.com", + "password": "secretpassword", + "displayName": "SSE User", + "instanceName": "MindSift SSE", + "allowRegistration": false, + "seedSampleData": false + }`) + req := httptest.NewRequest(http.MethodPost, "/api/v1/auth/setup", bytes.NewReader(setupBody)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("setup failed: %s", rec.Body.String()) + } + var authResp struct { + Token string `json:"token"` + User *domain.User `json:"user"` + } + _ = json.Unmarshal(rec.Body.Bytes(), &authResp) + token := authResp.Token + + // 2. Create Document + doc, err := store.CreateDocument(authResp.User.ID, "", "SSE Collaborative Note", "📄", "", nil, nil, false) + if err != nil { + t.Fatalf("failed to create document: %v", err) + } + + // 3. Test Awareness POST and DELETE + awarenessBody := []byte(`{"userId":"` + authResp.User.ID + `","activeBlockId":"blk_123"}`) + req = httptest.NewRequest(http.MethodPost, "/api/v1/documents/"+doc.ID+"/awareness", bytes.NewReader(awarenessBody)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + rec = httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 OK from POST awareness, got %d: %s", rec.Code, rec.Body.String()) + } + + req = httptest.NewRequest(http.MethodDelete, "/api/v1/documents/"+doc.ID+"/awareness", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec = httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 OK from DELETE awareness, got %d", rec.Code) + } + + // 4. Test Delta CRDT Update via PUT /api/v1/documents/{id}/crdt + updateBody := []byte(`{"update":"AQIDBA=="}`) + req = httptest.NewRequest(http.MethodPut, "/api/v1/documents/"+doc.ID+"/crdt", bytes.NewReader(updateBody)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + rec = httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 OK from PUT crdt update, got %d: %s", rec.Code, rec.Body.String()) + } + + // 5. Test SSE Connection GET /api/v1/events + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + req = httptest.NewRequest(http.MethodGet, "/api/v1/events?token="+token+"&docId="+doc.ID, nil).WithContext(ctx) + rec = httptest.NewRecorder() + + go func() { + // Cancel context after short moment to terminate stream cleanly + time.Sleep(50 * time.Millisecond) + cancel() + }() + + router.ServeHTTP(rec, req) + + contentType := rec.Header().Get("Content-Type") + if !strings.Contains(contentType, "text/event-stream") { + t.Fatalf("expected Content-Type text/event-stream, got %s", contentType) + } + if rec.Header().Get("X-Accel-Buffering") != "no" { + t.Fatalf("expected X-Accel-Buffering: no, got %s", rec.Header().Get("X-Accel-Buffering")) + } + bodyStr := rec.Body.String() + if !strings.Contains(bodyStr, "SSE_CONNECTED") { + t.Fatalf("expected SSE_CONNECTED event in body, got: %s", bodyStr) + } +} + diff --git a/internal/ws/hub.go b/internal/ws/hub.go index 5748539..6863cbc 100644 --- a/internal/ws/hub.go +++ b/internal/ws/hub.go @@ -3,6 +3,7 @@ package ws import ( "encoding/base64" "encoding/json" + "fmt" "log" "net/http" "strings" @@ -30,6 +31,14 @@ type Client struct { subscribedRooms map[string]bool } +type SSEClient struct { + id string + userID string + roomID string + send chan []byte + subscribedRooms map[string]bool +} + type docLockItem struct { docID string userID string @@ -45,20 +54,26 @@ type CRDTStore interface { } type Hub struct { - mu sync.RWMutex - rooms map[string]map[*Client]bool - userRooms map[string]map[*Client]bool - allClients map[*Client]bool - locks map[string]*docLockItem - crdtStore CRDTStore + mu sync.RWMutex + rooms map[string]map[*Client]bool + userRooms map[string]map[*Client]bool + allClients map[*Client]bool + sseClients map[*SSEClient]bool + sseUserRooms map[string]map[*SSEClient]bool + sseRooms map[string]map[*SSEClient]bool + locks map[string]*docLockItem + crdtStore CRDTStore } func NewHub() *Hub { h := &Hub{ - rooms: make(map[string]map[*Client]bool), - userRooms: make(map[string]map[*Client]bool), - allClients: make(map[*Client]bool), - locks: make(map[string]*docLockItem), + rooms: make(map[string]map[*Client]bool), + userRooms: make(map[string]map[*Client]bool), + allClients: make(map[*Client]bool), + sseClients: make(map[*SSEClient]bool), + sseUserRooms: make(map[string]map[*SSEClient]bool), + sseRooms: make(map[string]map[*SSEClient]bool), + locks: make(map[string]*docLockItem), } go h.cleanupLocksLoop() return h @@ -85,7 +100,7 @@ func (h *Hub) Register(c *Client) { } h.userRooms[c.userID][c] = true } - onlineCount := len(h.allClients) + onlineCount := len(h.allClients) + len(h.sseClients) h.mu.Unlock() if c.roomID != "" { @@ -126,11 +141,14 @@ func (h *Hub) Unregister(c *Client) { if c.userID != "" && c.userID != "anonymous" { if uRoom, ok := h.userRooms[c.userID]; ok { delete(uRoom, c) - userRemaining = len(uRoom) - if userRemaining == 0 { + userRemaining += len(uRoom) + if len(uRoom) == 0 { delete(h.userRooms, c.userID) } } + if sseRoom, ok := h.sseUserRooms[c.userID]; ok { + userRemaining += len(sseRoom) + } } close(c.send) @@ -161,6 +179,142 @@ func (h *Hub) Unregister(c *Client) { } } +func (h *Hub) RegisterSSE(c *SSEClient) { + h.mu.Lock() + h.sseClients[c] = true + if c.roomID != "" { + if h.sseRooms[c.roomID] == nil { + h.sseRooms[c.roomID] = make(map[*SSEClient]bool) + } + h.sseRooms[c.roomID][c] = true + } + if c.userID != "" && c.userID != "anonymous" { + if h.sseUserRooms[c.userID] == nil { + h.sseUserRooms[c.userID] = make(map[*SSEClient]bool) + } + h.sseUserRooms[c.userID][c] = true + } + if c.subscribedRooms != nil { + for rID := range c.subscribedRooms { + if h.sseRooms[rID] == nil { + h.sseRooms[rID] = make(map[*SSEClient]bool) + } + h.sseRooms[rID][c] = true + } + } + onlineCount := len(h.allClients) + len(h.sseClients) + h.mu.Unlock() + + if c.roomID != "" { + h.Broadcast(c.roomID, "", "USER_PRESENCE", map[string]interface{}{ + "onlineCount": onlineCount, + }) + } +} + +func (h *Hub) UnregisterSSE(c *SSEClient) { + h.mu.Lock() + delete(h.sseClients, c) + if c.roomID != "" { + if room, ok := h.sseRooms[c.roomID]; ok { + delete(room, c) + if len(room) == 0 { + delete(h.sseRooms, c.roomID) + } + } + } + var docRoomsLeft []string + if c.subscribedRooms != nil { + for rID := range c.subscribedRooms { + if room, ok := h.sseRooms[rID]; ok { + delete(room, c) + if len(room) == 0 { + delete(h.sseRooms, rID) + } + } + if strings.HasPrefix(rID, "doc:") { + docRoomsLeft = append(docRoomsLeft, strings.TrimPrefix(rID, "doc:")) + } + } + } + userRemaining := 0 + if c.userID != "" && c.userID != "anonymous" { + if uRoom, ok := h.sseUserRooms[c.userID]; ok { + delete(uRoom, c) + userRemaining += len(uRoom) + if len(uRoom) == 0 { + delete(h.sseUserRooms, c.userID) + } + } + if wsRoom, ok := h.userRooms[c.userID]; ok { + userRemaining += len(wsRoom) + } + } + close(c.send) + + var releasedDocIDs []string + if userRemaining == 0 && c.userID != "" { + now := time.Now() + for docID, l := range h.locks { + if l.userID == c.userID || l.expiresAt.Before(now) { + releasedDocIDs = append(releasedDocIDs, docID) + delete(h.locks, docID) + } + } + } + h.mu.Unlock() + + for _, docID := range releasedDocIDs { + h.BroadcastToAll("DOC_UNLOCKED", map[string]string{"docId": docID, "doc_id": docID}) + } + + for _, docID := range docRoomsLeft { + h.BroadcastRoom("doc:"+docID, "CRDT_AWARENESS_LEAVE", map[string]interface{}{ + "docId": docID, + "userId": c.userID, + }, nil) + } +} + +func (h *Hub) JoinSSERoomByID(sseID, roomID string) bool { + h.mu.Lock() + defer h.mu.Unlock() + for client := range h.sseClients { + if client.id == sseID { + if h.sseRooms[roomID] == nil { + h.sseRooms[roomID] = make(map[*SSEClient]bool) + } + h.sseRooms[roomID][client] = true + if client.subscribedRooms == nil { + client.subscribedRooms = make(map[string]bool) + } + client.subscribedRooms[roomID] = true + return true + } + } + return false +} + +func (h *Hub) LeaveSSERoomByID(sseID, roomID string) bool { + h.mu.Lock() + defer h.mu.Unlock() + for client := range h.sseClients { + if client.id == sseID { + if room, ok := h.sseRooms[roomID]; ok { + delete(room, client) + if len(room) == 0 { + delete(h.sseRooms, roomID) + } + } + if client.subscribedRooms != nil { + delete(client.subscribedRooms, roomID) + } + return true + } + } + return false +} + // Document Locking Methods func (h *Hub) AcquireDocLock(docID, userID, userName string, ttl time.Duration) (*domain.DocLockInfo, bool, *domain.DocLockInfo) { @@ -300,6 +454,12 @@ func (h *Hub) BroadcastToAll(event string, payload interface{}) { default: } } + for sseClient := range h.sseClients { + select { + case sseClient.send <- data: + default: + } + } } func (h *Hub) BroadcastToUser(userID, event string, payload interface{}) { @@ -324,6 +484,14 @@ func (h *Hub) BroadcastToUser(userID, event string, payload interface{}) { } } } + if sseRoom, ok := h.sseUserRooms[userID]; ok { + for sseClient := range sseRoom { + select { + case sseClient.send <- data: + default: + } + } + } } func (h *Hub) BroadcastTaskEvent(userID, event string, payload interface{}) { @@ -363,6 +531,14 @@ func (h *Hub) Broadcast(spaceID, listID, event string, payload interface{}) { } } } + if sseRoom, ok := h.sseRooms[targetRoom]; ok { + for sseClient := range sseRoom { + select { + case sseClient.send <- data: + default: + } + } + } } func (h *Hub) BroadcastListEvent(listID, event string, payload interface{}) { @@ -421,6 +597,14 @@ func (h *Hub) BroadcastRoom(roomID, event string, payload interface{}, excludeCl } } } + if sseRoom, ok := h.sseRooms[roomID]; ok { + for sseClient := range sseRoom { + select { + case sseClient.send <- data: + default: + } + } + } } func (c *Client) SendEvent(event string, payload interface{}) { @@ -458,6 +642,106 @@ func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request, roomID, userID str go client.readPump() } +func (h *Hub) ServeSSE(w http.ResponseWriter, r *http.Request, roomID, userID string, initialRooms []string) { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "Streaming unsupported!", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache, no-transform") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + w.Header().Set("Access-Control-Allow-Origin", "*") + + client := &SSEClient{ + id: domain.NewID("sse"), + userID: userID, + roomID: roomID, + send: make(chan []byte, 128), + subscribedRooms: make(map[string]bool), + } + + for _, room := range initialRooms { + if strings.TrimSpace(room) != "" { + client.subscribedRooms[strings.TrimSpace(room)] = true + } + } + + h.RegisterSSE(client) + defer h.UnregisterSSE(client) + + // Send initial connected comment and event + fmt.Fprintf(w, ": connected (transport: sse, id: %s)\n\n", client.id) + connectedMsg, _ := json.Marshal(domain.WSMessage{ + Event: "SSE_CONNECTED", + Payload: map[string]interface{}{ + "sseId": client.id, + "transport": "sse", + "userId": userID, + }, + Timestamp: domain.NowISO(), + }) + fmt.Fprintf(w, "data: %s\n\n", connectedMsg) + flusher.Flush() + + // If initial rooms had doc rooms, send CRDT_INIT + if h.crdtStore != nil { + for rID := range client.subscribedRooms { + if strings.HasPrefix(rID, "doc:") { + docID := strings.TrimPrefix(rID, "doc:") + stateBlob, updates, err := h.crdtStore.GetDocumentCRDTState(docID) + if err == nil && (len(stateBlob) > 0 || len(updates) > 0) { + var b64Updates []string + for _, u := range updates { + b64Updates = append(b64Updates, base64.StdEncoding.EncodeToString(u)) + } + b64State := "" + if len(stateBlob) > 0 { + b64State = base64.StdEncoding.EncodeToString(stateBlob) + } + initMsg := domain.WSMessage{ + Event: "CRDT_INIT", + Payload: map[string]interface{}{ + "docId": docID, + "state": b64State, + "updates": b64Updates, + }, + Timestamp: domain.NowISO(), + } + if initBytes, err := json.Marshal(initMsg); err == nil { + fmt.Fprintf(w, "data: %s\n\n", initBytes) + flusher.Flush() + } + } + } + } + } + + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + + notify := r.Context().Done() + + for { + select { + case <-notify: + return + case msg, ok := <-client.send: + if !ok { + return + } + fmt.Fprintf(w, "data: %s\n\n", msg) + flusher.Flush() + case <-ticker.C: + // Periodic ping comment to keep HTTP connection alive and prevent proxy timeouts + fmt.Fprintf(w, ": ping\n\n") + flusher.Flush() + } + } +} + func (c *Client) readPump() { defer func() { c.hub.Unregister(c)