feat: improve SSE broadcasting for weather sessions with enhanced error handling and connection management

src/app/api/weather/[id]/subscribe/route.ts

@@ -98,25 +98,38 @@ export async function GET(request: Request, { params }: { params: Promise<{ id:
 
 // Helper to broadcast to all connections (called from actions)
 export function broadcastToWeatherSession(sessionId: string, event: object) {
-  const sessionConnections = connections.get(sessionId);
-  if (!sessionConnections || sessionConnections.size === 0) {
-    return;
-  }
-
-  const encoder = new TextEncoder();
-  const message = encoder.encode(`data: ${JSON.stringify(event)}\n\n`);
-
-  for (const controller of sessionConnections) {
-    try {
-      controller.enqueue(message);
-    } catch {
-      // Connection might be closed, remove it
-      sessionConnections.delete(controller);
+  try {
+    const sessionConnections = connections.get(sessionId);
+    if (!sessionConnections || sessionConnections.size === 0) {
+      // No active connections, event will be picked up by polling
+      console.log(`[SSE Broadcast] No connections for session ${sessionId}, will be picked up by polling`);
+      return;
     }
-  }
 
-  // Clean up empty sets
-  if (sessionConnections.size === 0) {
-    connections.delete(sessionId);
+    console.log(`[SSE Broadcast] Broadcasting to ${sessionConnections.size} connections for session ${sessionId}`);
+
+    const encoder = new TextEncoder();
+    const message = encoder.encode(`data: ${JSON.stringify(event)}\n\n`);
+
+    let sentCount = 0;
+    for (const controller of sessionConnections) {
+      try {
+        controller.enqueue(message);
+        sentCount++;
+      } catch (error) {
+        // Connection might be closed, remove it
+        console.log(`[SSE Broadcast] Failed to send, removing connection:`, error);
+        sessionConnections.delete(controller);
+      }
+    }
+
+    console.log(`[SSE Broadcast] Sent to ${sentCount} connections`);
+
+    // Clean up empty sets
+    if (sessionConnections.size === 0) {
+      connections.delete(sessionId);
+    }
+  } catch (error) {
+    console.error('[SSE Broadcast] Error broadcasting:', error);
   }
 }