= ({
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)