realtime

pjw7536 · pjw7536 · commit 97c84f03c92e · 2025-10-07T10:22:53.000+09:00
diff --git a/apps/api/plane/bgtasks/issue_activities_task.py b/apps/api/plane/bgtasks/issue_activities_task.py
@@ -42,6 +42,67 @@ def extract_ids(data: dict | None, primary_key: str, fallback_key: str) -> set[s
     return {str(x) for x in data.get(fallback_key, [])}
 
 
+def _deserialize_for_broadcast(raw_value):
+    if raw_value is None:
+        return None
+    if isinstance(raw_value, (dict, list)):
+        return raw_value
+    if isinstance(raw_value, str):
+        try:
+            return json.loads(raw_value)
+        except ValueError:
+            return raw_value
+    return str(raw_value)
+
+
+def broadcast_issue_event(
+    *,
+    project_id,
+    issue_id,
+    event_type,
+    actor_id,
+    epoch,
+    requested_data,
+    current_instance,
+):
+    if not issue_id or not project_id:
+        return
+
+    try:
+        redis_client = redis_instance()
+    except Exception as redis_error:  # pragma: no cover - defensive guard
+        log_exception(redis_error)
+        return
+
+    if not redis_client:
+        return
+
+    channel_id = str(project_id)
+    payload = {
+        "type": event_type,
+        "issue_id": str(issue_id),
+        "project_id": channel_id,
+        "actor_id": str(actor_id) if actor_id else None,
+        "timestamp": int(epoch) if isinstance(epoch, (int, float)) else None,
+    }
+
+    requested_payload = _deserialize_for_broadcast(requested_data)
+    if requested_payload is not None:
+        payload["requested_data"] = requested_payload
+
+    current_payload = _deserialize_for_broadcast(current_instance)
+    if current_payload is not None:
+        payload["current_instance"] = current_payload
+
+    try:
+        redis_client.publish(
+            f"issue_events:{channel_id}",
+            json.dumps(payload, default=str),
+        )
+    except Exception as publish_error:  # pragma: no cover - defensive guard
+        log_exception(publish_error)
+
+
 # Track Changes in name
 def track_name(
     requested_data,
@@ -1580,45 +1641,15 @@ def issue_activity(
         issue_activities_created = IssueActivity.objects.bulk_create(issue_activities)
 
         # Broadcast project issue updates for realtime listeners
-        if issue_id:
-            try:
-                ri = redis_instance()
-                project_channel = str(project_id)
-                payload = {
-                    "type": type,
-                    "issue_id": str(issue_id),
-                    "project_id": project_channel,
-                    "actor_id": str(actor_id) if actor_id else None,
-                    "timestamp": int(epoch) if isinstance(epoch, (int, float)) else None,
-                }
-
-                def _deserialize(raw_value):
-                    if raw_value is None:
-                        return None
-                    if isinstance(raw_value, (dict, list)):
-                        return raw_value
-                    if isinstance(raw_value, str):
-                        try:
-                            return json.loads(raw_value)
-                        except ValueError:
-                            return raw_value
-                    return str(raw_value)
-
-                requested_payload = _deserialize(requested_data)
-                if requested_payload is not None:
-                    payload["requested_data"] = requested_payload
-
-                current_payload = _deserialize(current_instance)
-                if current_payload is not None:
-                    payload["current_instance"] = current_payload
-
-                ri.publish(
-                    f"issue_events:{project_channel}",
-                    json.dumps(payload, default=str),
-                )
-            except Exception as publish_error:
-                # Failure to broadcast should not block primary flow
-                log_exception(publish_error)
+        broadcast_issue_event(
+            project_id=project_id,
+            issue_id=issue_id,
+            event_type=type,
+            actor_id=actor_id,
+            epoch=epoch,
+            requested_data=requested_data,
+            current_instance=current_instance,
+        )
 
         if notification:
             notifications.delay(
diff --git a/apps/live/src/controllers/issue-events.controller.ts b/apps/live/src/controllers/issue-events.controller.ts
@@ -1,3 +1,4 @@
+import type Redis from "ioredis";
 import type { Request } from "express";
 import type WebSocket from "ws";
 // plane imports
@@ -8,104 +9,169 @@ import { redisManager } from "@/redis";
 // auth
 import { handleAuthentication } from "@/lib/auth";
 
-@Controller("/issues")
-export class IssueEventsController {
-  @WSDecorator("/")
-  async handleConnection(ws: WebSocket, req: Request) {
-    const query = req.query as Record<string, string | string[]>;
+type TokenPayload = {
+  id?: string;
+  cookie?: string;
+};
 
-    const projectIdRaw = query.projectId;
-    const workspaceSlugRaw = query.workspaceSlug;
-    const tokenRaw = query.token;
+type ConnectionParams = {
+  projectId: string;
+  workspaceSlug: string;
+  token: string;
+};
 
-    const projectId = Array.isArray(projectIdRaw) ? projectIdRaw[0] : projectIdRaw;
-    const workspaceSlug = Array.isArray(workspaceSlugRaw) ? workspaceSlugRaw[0] : workspaceSlugRaw;
-    const token = Array.isArray(tokenRaw) ? tokenRaw[0] : tokenRaw;
+const getFirstQueryValue = (value?: string | string[]) => (Array.isArray(value) ? value[0] : value);
 
-    if (!projectId || !workspaceSlug || !token) {
-      ws.close(4001, "Missing required parameters");
-      return;
+const extractConnectionParams = (req: Request): ConnectionParams | null => {
+  const query = req.query as Record<string, string | string[]>;
+  const projectId = getFirstQueryValue(query.projectId);
+  const workspaceSlug = getFirstQueryValue(query.workspaceSlug);
+  const token = getFirstQueryValue(query.token);
+
+  if (!projectId || !workspaceSlug || !token) {
+    return null;
+  }
+
+  return { projectId, workspaceSlug, token };
+};
+
+const parseToken = (rawToken: string): TokenPayload | null => {
+  try {
+    const parsed: unknown = JSON.parse(rawToken);
+    if (!parsed || typeof parsed !== "object") {
+      return null;
     }
+    return parsed as TokenPayload;
+  } catch (error) {
+    logger.error("Invalid token payload for issue events", error);
+    return null;
+  }
+};
 
-    let parsedToken: { id?: string; cookie?: string } | undefined;
+const closeSocket = (ws: WebSocket, code: number, reason: string) => {
+  if (ws.readyState === ws.CLOSED || ws.readyState === ws.CLOSING) {
+    return;
+  }
+
+  try {
+    ws.close(code, reason);
+  } catch (error) {
+    logger.error("Issue events websocket close failure", error);
+  }
+};
+
+const ensureAuthenticated = async (ws: WebSocket, token: TokenPayload, req: Request) => {
+  const cookie = token?.cookie || req.headers.cookie || "";
+  if (cookie) {
     try {
-      parsedToken = JSON.parse(token);
+      await handleAuthentication({
+        cookie,
+        userId: token?.id ?? "",
+      });
+      return true;
     } catch (error) {
-      logger.error("Invalid token payload for issue events", error);
-      ws.close(4002, "Invalid token");
+      logger.error("Failed to authenticate issue events connection", error);
+      closeSocket(ws, 4003, "Unauthorized");
+      return false;
+    }
+  }
+
+  if (!token?.id) {
+    closeSocket(ws, 4003, "Unauthorized");
+    return false;
+  }
+
+  return true;
+};
+
+@Controller("/issues")
+export class IssueEventsController {
+  @WSDecorator("/")
+  async handleConnection(ws: WebSocket, req: Request) {
+    const params = extractConnectionParams(req);
+    if (!params) {
+      closeSocket(ws, 4001, "Missing required parameters");
       return;
     }
 
-    const cookieString = parsedToken?.cookie || req.headers.cookie || "";
-    if (cookieString) {
-      try {
-        await handleAuthentication({
-          cookie: cookieString,
-          userId: parsedToken?.id ?? "",
-        });
-      } catch (error) {
-        logger.error("Failed to authenticate issue events connection", error);
-        ws.close(4003, "Unauthorized");
-        return;
-      }
-    } else if (!parsedToken?.id) {
-      ws.close(4003, "Unauthorized");
+    const tokenPayload = parseToken(params.token);
+    if (!tokenPayload) {
+      closeSocket(ws, 4002, "Invalid token");
+      return;
+    }
+
+    const authenticated = await ensureAuthenticated(ws, tokenPayload, req);
+    if (!authenticated) {
       return;
     }
 
     const redisClient = redisManager.getClient();
     if (!redisClient) {
-      ws.close(1011, "Realtime service unavailable");
+      closeSocket(ws, 1011, "Realtime service unavailable");
       return;
     }
 
-    const channel = `issue_events:${projectId}`;
-    const subscriber = redisClient.duplicate();
-    let cleanupInitiated = false;
+    let subscriber: Redis;
+    try {
+      subscriber = redisClient.duplicate();
+    } catch (error) {
+      logger.error("Failed to create issue events redis subscriber", error);
+      closeSocket(ws, 1011, "Realtime service unavailable");
+      return;
+    }
+
+    const channel = `issue_events:${params.projectId}`;
+    let cleanupStarted = false;
 
     const cleanup = async () => {
-      if (cleanupInitiated) return;
-      cleanupInitiated = true;
+      if (cleanupStarted) return;
+      cleanupStarted = true;
+
       subscriber.removeAllListeners("message");
       subscriber.removeAllListeners("error");
+
       try {
         await subscriber.unsubscribe(channel);
       } catch (error) {
         logger.error("Failed to unsubscribe issue events channel", error);
       }
-      subscriber.disconnect();
+
+      try {
+        subscriber.disconnect();
+      } catch (error) {
+        logger.error("Failed to disconnect issue events subscriber", error);
+      }
     };
 
     try {
       subscriber.on("error", (error) => {
         logger.error("Issue events redis subscriber error", error);
-        if (ws.readyState === ws.OPEN || ws.readyState === ws.CONNECTING) {
-          ws.close(1011, "Realtime service unavailable");
-        }
+        closeSocket(ws, 1011, "Realtime service unavailable");
         void cleanup();
       });
+
       await subscriber.connect();
+
       subscriber.on("message", (incomingChannel, message) => {
         if (incomingChannel === channel && ws.readyState === ws.OPEN) {
           ws.send(message);
         }
       });
+
       await subscriber.subscribe(channel);
+
       ws.on("close", () => {
         void cleanup();
       });
+
       ws.on("error", (error) => {
         logger.error("Issue events websocket error", error);
-        if (ws.readyState === ws.OPEN || ws.readyState === ws.CONNECTING) {
-          ws.close(1011, "Issue events websocket error");
-        }
+        closeSocket(ws, 1011, "Issue events websocket error");
         void cleanup();
       });
     } catch (error) {
       logger.error("Failed to subscribe to issue events channel", error);
-      if (ws.readyState === ws.OPEN || ws.readyState === ws.CONNECTING) {
-        ws.close(1011, "Subscription failure");
-      }
+      closeSocket(ws, 1011, "Subscription failure");
       void cleanup();
     }
   }
diff --git a/apps/web/core/hooks/use-issue-channel.ts b/apps/web/core/hooks/use-issue-channel.ts
