ФУНКЦИИ - МЕЖПРОЕКТНАЯ КОММУНИКАЦИЯ: presence автора и унификация карточек

2026-04-29 14:22:42 +03:00
parent d53fa2b38c
commit 4dbb7b500c
11 changed files with 466 additions and 89 deletions
@@ -5,6 +5,7 @@
 */

 import type { Request } from "express";
+import { randomUUID } from "crypto";
 import type Redis from "ioredis";
 import type { WebSocket as WSSocket } from "ws";
 // plane imports
@@ -17,7 +18,11 @@ import { ProjectMemberService } from "@/services/project-member.service";
 import { UserService } from "@/services/user.service";

 const ISSUE_EVENT_CHANNEL_PREFIX = "plane:issue-events:project";
+const PRESENCE_CHANNEL_PREFIX = "plane:presence:workspace";
+const PRESENCE_KEY_PREFIX = "plane:presence:workspace";
 const HEARTBEAT_INTERVAL_MS = 25_000;
+const PRESENCE_TTL_SECONDS = 70;
+const OFFLINE_GRACE_MS = 4_000;

 type TIssueRealtimeEvent = {
  event_id?: string;
@@ -32,6 +37,69 @@ const sendJson = (ws: WSSocket, payload: Record<string, unknown>) => {
  ws.send(JSON.stringify(payload));
 };

+const presenceChannel = (workspaceSlug: string) => `${PRESENCE_CHANNEL_PREFIX}:${workspaceSlug}`;
+
+const presenceConnectionKey = (workspaceSlug: string, userId: string, connectionId: string) =>
+  `${PRESENCE_KEY_PREFIX}:${workspaceSlug}:user:${userId}:connection:${connectionId}`;
+
+const presenceConnectionPattern = (workspaceSlug: string, userId?: string) =>
+  userId
+    ? `${PRESENCE_KEY_PREFIX}:${workspaceSlug}:user:${userId}:connection:*`
+    : `${PRESENCE_KEY_PREFIX}:${workspaceSlug}:user:*:connection:*`;
+
+const parseUserIdFromPresenceKey = (workspaceSlug: string, key: string) => {
+  const prefix = `${PRESENCE_KEY_PREFIX}:${workspaceSlug}:user:`;
+  if (!key.startsWith(prefix)) return undefined;
+
+  const userId = key.slice(prefix.length).split(":connection:")[0];
+  return userId || undefined;
+};
+
+const scanPresenceKeys = async (redisClient: Redis, pattern: string) => {
+  const keys: string[] = [];
+  let cursor = "0";
+
+  do {
+    const [nextCursor, matchedKeys] = await redisClient.scan(cursor, "MATCH", pattern, "COUNT", 250);
+    cursor = nextCursor;
+    keys.push(...matchedKeys);
+  } while (cursor !== "0");
+
+  return keys;
+};
+
+const getPresenceConnectionCount = async (redisClient: Redis, workspaceSlug: string, userId: string) =>
+  (await scanPresenceKeys(redisClient, presenceConnectionPattern(workspaceSlug, userId))).length;
+
+const getOnlineUserIds = async (redisClient: Redis, workspaceSlug: string) => {
+  const keys = await scanPresenceKeys(redisClient, presenceConnectionPattern(workspaceSlug));
+  const userIds = new Set<string>();
+
+  keys.forEach((key) => {
+    const userId = parseUserIdFromPresenceKey(workspaceSlug, key);
+    if (userId) userIds.add(userId);
+  });
+
+  return [...userIds];
+};
+
+const publishPresenceEvent = async (
+  redisClient: Redis,
+  workspaceSlug: string,
+  type: "presence.user.online" | "presence.user.offline" | "presence.user.heartbeat",
+  userId: string
+) =>
+  redisClient.publish(
+    presenceChannel(workspaceSlug),
+    JSON.stringify({
+      event_id: randomUUID(),
+      type,
+      workspace_slug: workspaceSlug,
+      user_id: userId,
+      server_ts: new Date().toISOString(),
+    })
+  );
+
@Controller("/issues")
 export class IssueStreamController {
  [key: string]: unknown;
@@ -53,8 +121,14 @@ export class IssueStreamController {

    let subscriber: Redis | undefined;
    let heartbeat: NodeJS.Timeout | undefined;
+    let presenceKey: string | undefined;
+    let presenceUserId: string | undefined;
+    let isCleanedUp = false;

    const cleanup = async () => {
+      if (isCleanedUp) return;
+      isCleanedUp = true;
+
      if (heartbeat) clearInterval(heartbeat);

      if (subscriber) {
@@ -65,6 +139,26 @@ export class IssueStreamController {
          logger.error("ISSUE_STREAM_CONTROLLER: Redis cleanup failed:", error);
        }
      }
+
+      const redisClient = redisManager.getClient();
+      if (!redisClient || !presenceKey || !presenceUserId || !workspaceSlug) return;
+
+      try {
+        const currentPresenceUserId = presenceUserId;
+        const currentPresenceKey = presenceKey;
+
+        await redisClient.del(currentPresenceKey);
+        setTimeout(() => {
+          void (async () => {
+            const activeConnections = await getPresenceConnectionCount(redisClient, workspaceSlug, currentPresenceUserId);
+            if (activeConnections === 0) {
+              await publishPresenceEvent(redisClient, workspaceSlug, "presence.user.offline", currentPresenceUserId);
+            }
+          })();
+        }, OFFLINE_GRACE_MS);
+      } catch (error) {
+        logger.error("ISSUE_STREAM_CONTROLLER: Presence cleanup failed:", error);
+      }
    };

    try {
@@ -81,13 +175,41 @@ export class IssueStreamController {
      }

      const channel = `${ISSUE_EVENT_CHANNEL_PREFIX}:${projectId}`;
+      const workspacePresenceChannel = presenceChannel(workspaceSlug);
      subscriber = redisClient.duplicate({ lazyConnect: true });
      await subscriber.connect();
-      await subscriber.subscribe(channel);
+      await subscriber.subscribe(channel, workspacePresenceChannel);
+
+      const connectionId = randomUUID();
+      presenceUserId = user.id;
+      presenceKey = presenceConnectionKey(workspaceSlug, user.id, connectionId);
+
+      const activeConnections = await getPresenceConnectionCount(redisClient, workspaceSlug, user.id);
+      await redisClient.setex(
+        presenceKey,
+        PRESENCE_TTL_SECONDS,
+        JSON.stringify({
+          connected_at: new Date().toISOString(),
+          project_id: projectId,
+          user_id: user.id,
+          workspace_slug: workspaceSlug,
+        })
+      );
+
+      if (activeConnections === 0) {
+        await publishPresenceEvent(redisClient, workspaceSlug, "presence.user.online", user.id);
+      }

      subscriber.on("message", (_channel, message) => {
        try {
          const event = JSON.parse(message) as TIssueRealtimeEvent;
+
+          if (_channel === workspacePresenceChannel) {
+            if (!event.type?.startsWith("presence.")) return;
+            sendJson(ws, event as Record<string, unknown>);
+            return;
+          }
+
          if (
            event.project_id !== projectId ||
            (!event.type?.startsWith("issue.") && !event.type?.startsWith("external_contour."))
@@ -106,6 +228,21 @@ export class IssueStreamController {
      });

      heartbeat = setInterval(() => {
+        if (presenceKey) {
+          void redisClient
+            .setex(
+              presenceKey,
+              PRESENCE_TTL_SECONDS,
+              JSON.stringify({
+                heartbeat_at: new Date().toISOString(),
+                project_id: projectId,
+                user_id: user.id,
+                workspace_slug: workspaceSlug,
+              })
+            )
+            .then(() => publishPresenceEvent(redisClient, workspaceSlug, "presence.user.heartbeat", user.id))
+            .catch((error) => logger.error("ISSUE_STREAM_CONTROLLER: Presence heartbeat failed:", error));
+        }
        sendJson(ws, { type: "issue.stream.ping", server_ts: new Date().toISOString() });
      }, HEARTBEAT_INTERVAL_MS);

@@ -114,6 +251,12 @@ export class IssueStreamController {
        project_id: projectId,
        user_id: user.id,
      });
+      sendJson(ws, {
+        type: "presence.snapshot",
+        workspace_slug: workspaceSlug,
+        online_user_ids: await getOnlineUserIds(redisClient, workspaceSlug),
+        server_ts: new Date().toISOString(),
+      });
    } catch (error) {
      logger.error("ISSUE_STREAM_CONTROLLER: WebSocket authentication failed:", error);
      ws.close(1008, "Issue stream authentication failed");