real-time

Author: pjw7536

apps/api/plane/bgtasks/issue_activities_task.py

@@ -1579,6 +1579,47 @@ def issue_activity(
         # Save all the values to database
         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)
+
         if notification:
             notifications.delay(
                 type=type,

apps/live/src/controllers/index.ts

@@ -1,5 +1,6 @@
 import { CollaborationController } from "./collaboration.controller";
 import { ConvertDocumentController } from "./convert-document.controller";
 import { HealthController } from "./health.controller";
+import { IssueEventsController } from "./issue-events.controller";
 
-export const CONTROLLERS = [CollaborationController, ConvertDocumentController, HealthController];
+export const CONTROLLERS = [CollaborationController, ConvertDocumentController, HealthController, IssueEventsController];

apps/live/src/controllers/issue-events.controller.ts

@@ -0,0 +1,112 @@
+import type { Request } from "express";
+import type WebSocket from "ws";
+// plane imports
+import { Controller, WebSocket as WSDecorator } from "@plane/decorators";
+import { logger } from "@plane/logger";
+// redis
+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[]>;
+
+    const projectIdRaw = query.projectId;
+    const workspaceSlugRaw = query.workspaceSlug;
+    const tokenRaw = query.token;
+
+    const projectId = Array.isArray(projectIdRaw) ? projectIdRaw[0] : projectIdRaw;
+    const workspaceSlug = Array.isArray(workspaceSlugRaw) ? workspaceSlugRaw[0] : workspaceSlugRaw;
+    const token = Array.isArray(tokenRaw) ? tokenRaw[0] : tokenRaw;
+
+    if (!projectId || !workspaceSlug || !token) {
+      ws.close(4001, "Missing required parameters");
+      return;
+    }
+
+    let parsedToken: { id?: string; cookie?: string } | undefined;
+    try {
+      parsedToken = JSON.parse(token);
+    } catch (error) {
+      logger.error("Invalid token payload for issue events", error);
+      ws.close(4002, "Invalid token");
+      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");
+      return;
+    }
+
+    const redisClient = redisManager.getClient();
+    if (!redisClient) {
+      ws.close(1011, "Realtime service unavailable");
+      return;
+    }
+
+    const channel = `issue_events:${projectId}`;
+    const subscriber = redisClient.duplicate();
+    let cleanupInitiated = false;
+
+    const cleanup = async () => {
+      if (cleanupInitiated) return;
+      cleanupInitiated = 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.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");
+        }
+        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");
+        }
+        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");
+      }
+      void cleanup();
+    }
+  }
+}