added auth

package-lock.json

@@ -8,6 +8,7 @@
       "name": "chat_app_frontend",
       "version": "0.1.0",
       "dependencies": {
+        "@microsoft/fetch-event-source": "^2.0.1",
         "@radix-ui/react-slot": "^1.2.4",
         "class-variance-authority": "^0.7.1",
         "clsx": "^2.1.1",
@@ -1523,6 +1524,12 @@
         "@jridgewell/sourcemap-codec": "^1.4.14"
       }
     },
+    "node_modules/@microsoft/fetch-event-source": {
+      "version": "2.0.1",
+      "resolved": "https://registry.npmjs.org/@microsoft/fetch-event-source/-/fetch-event-source-2.0.1.tgz",
+      "integrity": "sha512-W6CLUJ2eBMw3Rec70qrsEW0jOm/3twwJv21mrmj2yORiaVmVYGS4sSS5yUwvQc1ZlDLYGPnClVWmUUMagKNsfA==",
+      "license": "MIT"
+    },
     "node_modules/@modelcontextprotocol/sdk": {
       "version": "1.29.0",
       "resolved": "https://registry.npmjs.org/@modelcontextprotocol/sdk/-/sdk-1.29.0.tgz",

package.json

@@ -9,6 +9,7 @@
     "lint": "eslint"
   },
   "dependencies": {
+    "@microsoft/fetch-event-source": "^2.0.1",
     "@radix-ui/react-slot": "^1.2.4",
     "class-variance-authority": "^0.7.1",
     "clsx": "^2.1.1",

src/app/chat/page.tsx

@@ -180,11 +180,11 @@ export default function ChatPage() {
 
   // ─── SSE: real-time group invite / removal ───────────────────────────────
 useEffect(() => {
-  if (!user?.userId) return;
+  if (!user?.token) return;
 
-  console.log("[SSE HOOK] Initializing for user:", user.userId);
+  console.log("[SSE HOOK] Initializing authenticated SSE for user:", user.username);
 
-  const cleanup = subscribeToSSE(user.userId, {
+  const cleanup = subscribeToSSE(user.token, {
     // Someone added this user to a group conversation
     onGroupInvite: ({ conversation }) => {
       console.log("[SSE HOOK] onGroupInvite TRIGGERED");
@@ -291,7 +291,7 @@ useEffect(() => {
   });
 
   return cleanup;
-}, [user?.userId, joinRoom, leaveRoom, setCurrentRoom, setMessages]);
+}, [user?.token, joinRoom, leaveRoom, setCurrentRoom, setMessages]);
 
 
   // React to P2P Ready event

src/lib/api.ts

@@ -1,3 +1,6 @@
+import { fetchEventSource } from '@microsoft/fetch-event-source';
+
+
 export const API_BASE_URL = process.env.NEXT_PUBLIC_API_URL;
 
 export enum ConversationType {
@@ -83,64 +86,98 @@ export async function fetchMessages(convId: string): Promise<any[]> {
   return res.json();
 }
 
+
+
+
 export function subscribeToSSE(
-  userId: string,
+  token: string,
   handlers: {
     onGroupInvite?: (data: { conversation: any }) => void;
     onGroupRemoved?: (data: { conversation: any }) => void;
   }
 ): () => void {
-  const url = `${API_BASE_URL}/conversations/stream?userId=${userId}`;
+  const url = `${API_BASE_URL}/conversations/stream`;
-  console.log("[SSE] Opening EventSource:", url);
+  console.log("[SSE] Opening authenticated stream:", url);
 
-  const es = new EventSource(url);
+  const ctrl = new AbortController();
 
-  es.onopen = () => {
+  fetchEventSource(url, {
-    console.log("[SSE] Connection OPENED. readyState:", es.readyState);
+    method: 'GET',
-  };
+    headers: {
+      Authorization: `Bearer ${token}`,
+    },
+    signal: ctrl.signal,
 
-  // Catch-all: any unnamed event (i.e. events without `event:` field in the stream)
+    // --- connection opened ---
-  es.onmessage = (e: MessageEvent) => {
+    onopen: async (response) => {
-    console.warn("[SSE] Received UNNAMED message (no event type). This means the backend is NOT setting the event name correctly.");
+      if (response.ok) {
-    console.warn("[SSE] Unnamed message data:", e.data);
+        console.log("[SSE] Connection OPENED. Status:", response.status);
-  };
+        return; // everything is fine
+      }
+
+      // non-retriable errors (401, 403, etc.)
+      const text = await response.text().catch(() => '');
+      console.error(`[SSE] Server responded with ${response.status}: ${text}`);
+      throw new Error(`SSE connection rejected: ${response.status}`);
+    },
+
+    // --- incoming event ---
+    onmessage: (ev) => {
+      const { event, data } = ev;
+
+      // Unnamed / heartbeat messages
+      if (!event || event === 'message') {
+        console.log("[SSE] heartbeat / unnamed message:", data);
+        return;
+      }
+
+      console.log(`[SSE] Received event: ${event}`);
+      console.log(`[SSE] Raw data:`, data);
 
-  es.addEventListener("groupInvite", (e: MessageEvent) => {
-    console.log("[SSE] Received named event: groupInvite");
-    console.log("[SSE] groupInvite raw data:", e.data);
       try {
-      const parsed = JSON.parse(e.data);
+        const parsed = JSON.parse(data);
+
+        switch (event) {
+          case 'groupInvite':
             console.log("[SSE] groupInvite parsed:", parsed);
             handlers.onGroupInvite?.(parsed);
-    } catch (err) {
+            break;
-      console.error("[SSE] Failed to parse groupInvite data:", err);
-    }
-  });
 
-  es.addEventListener("groupRemoved", (e: MessageEvent) => {
+          case 'groupRemoved':
-    console.log("[SSE] Received named event: groupRemoved");
-    console.log("[SSE] groupRemoved raw data:", e.data);
-    try {
-      const parsed = JSON.parse(e.data);
             console.log("[SSE] groupRemoved parsed:", parsed);
             handlers.onGroupRemoved?.(parsed);
-    } catch (err) {
+            break;
-      console.error("[SSE] Failed to parse groupRemoved data:", err);
+
+          default:
+            console.warn("[SSE] Unhandled event type:", event, parsed);
         }
+      } catch (err) {
+        console.error(`[SSE] Failed to parse ${event} data:`, err);
+      }
+    },
+
+    // --- error handling / reconnection ---
+    onerror: (err) => {
+      console.warn("[SSE] Connection error:", err);
+
+      // If we intentionally aborted, don't retry
+      if (ctrl.signal.aborted) {
+        console.log("[SSE] Aborted – no retry.");
+        throw err; // throwing inside onerror stops reconnection
+      }
+
+      // For any other error, fetchEventSource will automatically retry.
+      // Returning nothing (undefined) lets the library reconnect.
+      console.log("[SSE] Will auto-retry...");
+    },
+
+    // Keep the connection alive — don't close on page-hide (tab switch, etc.)
+    openWhenHidden: true,
   });
 
-  es.onerror = (err) => {
+  // Return cleanup: abort the stream
-    console.warn("[SSE] Connection error. readyState:", es.readyState, "| 0=CONNECTING, 1=OPEN, 2=CLOSED");
-    console.warn("[SSE] Error object:", err);
-    if (es.readyState === EventSource.CLOSED) {
-      console.error("[SSE] Connection CLOSED permanently. Server may have rejected the request.");
-    } else {
-      console.log("[SSE] Will auto-retry (readyState=CONNECTING)...");
-    }
-  };
-
   return () => {
-    console.log("[SSE] Closing EventSource for userId:", userId);
+    console.log("[SSE] Aborting SSE stream.");
-    es.close();
+    ctrl.abort();
   };
 }