ФУНКЦИИ - МЕЖПРОЕКТНАЯ КОММУНИКАЦИЯ: realtime внешних контуров и действия исходящих карточек

2026-04-29 12:53:04 +03:00
parent b2a710a7ec
commit 248292bd52
14 changed files with 819 additions and 105 deletions
@@ -0,0 +1,137 @@
+/**
+ * Copyright (c) 2023-present Plane Software, Inc. and contributors
+ * SPDX-License-Identifier: AGPL-3.0-only
+ * See the LICENSE file for details.
+ */
+
+import { useEffect, useRef } from "react";
+import { LIVE_BASE_PATH, LIVE_BASE_URL } from "@plane/constants";
+
+type TExternalContourRealtimeEvent = {
+  event_id?: string;
+  type?: string;
+  workspace_slug?: string;
+  project_id?: string;
+};
+
+const SYNC_DEBOUNCE_MS = 350;
+
+const buildIssueStreamUrl = (workspaceSlug: string, projectId: string) => {
+  const liveBaseUrl = LIVE_BASE_URL?.trim() || window.location.origin;
+  const liveBasePath = LIVE_BASE_PATH?.trim() || "/live";
+  const url = new URL(liveBaseUrl);
+
+  url.protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
+  url.pathname = `${liveBasePath.replace(/\/$/, "")}/issues/stream`;
+  url.searchParams.set("workspaceSlug", workspaceSlug);
+  url.searchParams.set("projectId", projectId);
+
+  return url.toString();
+};
+
+export const useExternalContoursRealtimeEvents = (
+  workspaceSlug: string | undefined,
+  projectId: string | undefined,
+  syncBoard: (workspaceSlug: string, projectId: string) => Promise<void>
+) => {
+  const syncBoardRef = useRef(syncBoard);
+  const processedEventIdsRef = useRef<string[]>([]);
+  const processedEventSetRef = useRef(new Set<string>());
+  const syncTimerRef = useRef<ReturnType<typeof setTimeout> | undefined>();
+
+  useEffect(() => {
+    syncBoardRef.current = syncBoard;
+  }, [syncBoard]);
+
+  useEffect(() => {
+    if (!workspaceSlug || !projectId || typeof window === "undefined") return;
+
+    let socket: WebSocket | undefined;
+    let reconnectTimer: ReturnType<typeof setTimeout> | undefined;
+    let cancelled = false;
+    let reconnectAttempt = 0;
+
+    const rememberEvent = (eventId?: string) => {
+      if (!eventId) return true;
+      if (processedEventSetRef.current.has(eventId)) return false;
+
+      processedEventIdsRef.current.push(eventId);
+      processedEventSetRef.current.add(eventId);
+
+      if (processedEventIdsRef.current.length > 250) {
+        const removedEventId = processedEventIdsRef.current.shift();
+        if (removedEventId) processedEventSetRef.current.delete(removedEventId);
+      }
+
+      return true;
+    };
+
+    const scheduleSync = () => {
+      if (cancelled) return;
+      if (syncTimerRef.current) clearTimeout(syncTimerRef.current);
+
+      syncTimerRef.current = setTimeout(() => {
+        void syncBoardRef.current(workspaceSlug, projectId);
+      }, SYNC_DEBOUNCE_MS);
+    };
+
+    const scheduleReconnect = () => {
+      if (cancelled) return;
+      const delay = Math.min(1000 * 2 ** reconnectAttempt, 15000);
+      reconnectAttempt += 1;
+      reconnectTimer = setTimeout(connect, delay);
+    };
+
+    const connect = () => {
+      try {
+        socket = new WebSocket(buildIssueStreamUrl(workspaceSlug, projectId));
+
+        socket.onopen = () => {
+          reconnectAttempt = 0;
+          scheduleSync();
+        };
+
+        socket.onmessage = (message) => {
+          try {
+            const event = JSON.parse(message.data) as TExternalContourRealtimeEvent;
+
+            if (event.type === "issue.stream.ping") {
+              socket?.send(JSON.stringify({ type: "issue.stream.pong" }));
+              return;
+            }
+
+            if (event.type === "issue.stream.ready") return;
+            if (event.workspace_slug && event.workspace_slug !== workspaceSlug) return;
+            if (event.project_id && event.project_id !== projectId) return;
+            if (!event.type?.startsWith("external_contour.") && !event.type?.startsWith("issue.")) return;
+            if (!rememberEvent(event.event_id)) return;
+
+            scheduleSync();
+          } catch (error) {
+            console.error("Failed to process external contour realtime event", error);
+          }
+        };
+
+        socket.onclose = () => {
+          scheduleReconnect();
+        };
+
+        socket.onerror = () => {
+          socket?.close();
+        };
+      } catch (error) {
+        console.error("Failed to connect external contour realtime stream", error);
+        scheduleReconnect();
+      }
+    };
+
+    connect();
+
+    return () => {
+      cancelled = true;
+      if (syncTimerRef.current) clearTimeout(syncTimerRef.current);
+      if (reconnectTimer) clearTimeout(reconnectTimer);
+      socket?.close();
+    };
+  }, [workspaceSlug, projectId]);
+};
@@ -41,6 +41,7 @@ type TIssueFilterSnapshot = {

 const REALTIME_STORE_TYPES = new Set<EIssuesStoreType>([EIssuesStoreType.PROJECT, EIssuesStoreType.PROJECT_VIEW]);
 const MAX_PROCESSED_EVENTS = 250;
+const INITIAL_CATCH_UP_WINDOW_MS = 5 * 60 * 1000;

 const hasIssueId = (value: unknown, issueId: string): boolean => {
  if (Array.isArray(value)) return value.includes(issueId);
@@ -91,7 +92,6 @@ export const useIssueRealtimeEvents = (storeType: EIssuesStoreType, workspaceSlu
    let reconnectTimer: ReturnType<typeof setTimeout> | undefined;
    let cancelled = false;
    let reconnectAttempt = 0;
-    let hasConnectedOnce = false;

    const getFilterParams = () => {
      const filters = { ...(issueFilterRef.current?.appliedFilters ?? {}) };
@@ -103,6 +103,32 @@ export const useIssueRealtimeEvents = (storeType: EIssuesStoreType, workspaceSlu
      return filters;
    };

+    const rememberUpdatedAt = (updatedAt?: string) => {
+      if (!updatedAt) return;
+
+      const currentUpdatedAt = lastSeenUpdatedAtRef.current;
+      if (!currentUpdatedAt || Date.parse(updatedAt) > Date.parse(currentUpdatedAt)) {
+        lastSeenUpdatedAtRef.current = updatedAt;
+      }
+    };
+
+    const getInitialCatchUpStart = () => {
+      const latestKnownUpdatedAt = Object.values(issueMapRef.current ?? {}).reduce<string | undefined>(
+        (latestUpdatedAt, issue) => {
+          if (!issue?.updated_at) return latestUpdatedAt;
+          if (!latestUpdatedAt || Date.parse(issue.updated_at) > Date.parse(latestUpdatedAt)) return issue.updated_at;
+
+          return latestUpdatedAt;
+        },
+        undefined
+      );
+
+      const fallbackUpdatedAt = new Date(Date.now() - INITIAL_CATCH_UP_WINDOW_MS).toISOString();
+      if (!latestKnownUpdatedAt) return fallbackUpdatedAt;
+
+      return Date.parse(latestKnownUpdatedAt) < Date.parse(fallbackUpdatedAt) ? latestKnownUpdatedAt : fallbackUpdatedAt;
+    };
+
    const rememberEvent = (eventId: string) => {
      if (processedEventSetRef.current.has(eventId)) return false;

@@ -146,7 +172,7 @@ export const useIssueRealtimeEvents = (storeType: EIssuesStoreType, workspaceSlu
    const handleIssueEvent = async (event: TIssueRealtimeEvent) => {
      if (!event.event_id || !event.issue_id) return;
      if (!rememberEvent(event.event_id)) return;
-      if (event.updated_at) lastSeenUpdatedAtRef.current = event.updated_at;
+      rememberUpdatedAt(event.updated_at);

      if (event.type === "issue.deleted") {
        removeIssue(event.issue_id, true);
@@ -175,7 +201,10 @@ export const useIssueRealtimeEvents = (storeType: EIssuesStoreType, workspaceSlu
      const results = response?.results;
      if (!Array.isArray(results)) return;

-      results.forEach(applyIssue);
+      results.forEach((issue) => {
+        rememberUpdatedAt(issue.updated_at);
+        applyIssue(issue);
+      });
    };

    const scheduleReconnect = () => {
@@ -191,8 +220,8 @@ export const useIssueRealtimeEvents = (storeType: EIssuesStoreType, workspaceSlu

        socket.onopen = () => {
          reconnectAttempt = 0;
-          if (hasConnectedOnce) void catchUpMissedEvents();
-          hasConnectedOnce = true;
+          lastSeenUpdatedAtRef.current = lastSeenUpdatedAtRef.current ?? getInitialCatchUpStart();
+          void catchUpMissedEvents();
        };

        socket.onmessage = (message) => {
@@ -205,6 +234,7 @@ export const useIssueRealtimeEvents = (storeType: EIssuesStoreType, workspaceSlu
            }

            if (event.type === "issue.stream.ready") return;
+            if (!event.type?.startsWith("issue.")) return;
            if (event.workspace_slug && event.workspace_slug !== workspaceSlug) return;
            if (event.project_id && event.project_id !== projectId) return;