live sync working

apps/admin/src/app/chat/page.tsx

@@ -57,6 +57,11 @@ export default function AdminChatPage() {
 
   const typingTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
   const socketRef = useRef<Socket | null>(null);
+  const getTokenRef = useRef(getToken);
+  const activeThreadIdRef = useRef<string | null>(null);
+  const subscribedThreadIdsRef = useRef<Set<string>>(new Set());
+  const loadingThreadsRef = useRef(false);
+  const loadingMessagesRef = useRef<Record<string, boolean>>({});
 
   const activeMessages = useMemo(
     () => (activeThreadId ? (messagesByThreadId[activeThreadId] ?? []) : []),
@@ -69,14 +74,28 @@ export default function AdminChatPage() {
   );
 
   useEffect(() => {
-    const loadThreads = async () => {
+    getTokenRef.current = getToken;
-      const token = await getToken();
+  }, [getToken]);
+
+  useEffect(() => {
+    activeThreadIdRef.current = activeThreadId;
+  }, [activeThreadId]);
+
+  const loadThreads = async (showLoader = true) => {
+    if (loadingThreadsRef.current) {
+      return;
+    }
+
+    const token = await getTokenRef.current();
     if (!token) {
       return;
     }
 
+    loadingThreadsRef.current = true;
     try {
+      if (showLoader) {
         setLoadingThreads(true);
+      }
       const response = await axios.get<{ threads: ChatThread[] }>(
         "/api/chat/threads",
         {
@@ -85,33 +104,33 @@ export default function AdminChatPage() {
       );
 
       setThreads(response.data.threads);
-        setActiveThreadId(
+      setActiveThreadId((prev) => prev ?? response.data.threads[0]?.id ?? null);
-          (prev) => prev ?? response.data.threads[0]?.id ?? null,
-        );
       setReadByThreadId({});
     } finally {
+      if (showLoader) {
         setLoadingThreads(false);
       }
+      loadingThreadsRef.current = false;
+    }
   };
 
-    void loadThreads();
+  const loadMessages = async (threadId: string, showLoader = true) => {
-  }, [getToken]);
+    if (loadingMessagesRef.current[threadId]) {
-
-  useEffect(() => {
-    if (!activeThreadId) {
       return;
     }
 
-    const loadMessages = async () => {
+    const token = await getTokenRef.current();
-      const token = await getToken();
     if (!token) {
       return;
     }
 
+    loadingMessagesRef.current[threadId] = true;
     try {
+      if (showLoader) {
         setLoadingMessages(true);
+      }
       const response = await axios.get<{ messages: ChatMessage[] }>(
-          `/api/chat/threads/${activeThreadId}/messages`,
+        `/api/chat/threads/${threadId}/messages`,
         {
           headers: { Authorization: `Bearer ${token}` },
         },
@@ -119,21 +138,33 @@ export default function AdminChatPage() {
 
       setMessagesByThreadId((prev) => ({
         ...prev,
-          [activeThreadId]: response.data.messages,
+        [threadId]: response.data.messages,
       }));
     } finally {
+      if (showLoader) {
         setLoadingMessages(false);
       }
+      loadingMessagesRef.current[threadId] = false;
+    }
   };
 
-    void loadMessages();
+  useEffect(() => {
-  }, [activeThreadId, getToken]);
+    void loadThreads(true);
+  }, []);
+
+  useEffect(() => {
+    if (!activeThreadId) {
+      return;
+    }
+
+    void loadMessages(activeThreadId, true);
+  }, [activeThreadId]);
 
   useEffect(() => {
     let mounted = true;
 
     const setupSocket = async () => {
-      const token = await getToken();
+      const token = await getTokenRef.current();
       if (!token || !mounted) {
         return;
       }
@@ -151,14 +182,14 @@ export default function AdminChatPage() {
 
       socket.on("connect", () => {
         setSocketConnected(true);
-        if (activeThreadId) {
+        subscribedThreadIdsRef.current.clear();
-          socket.emit("chat:subscribe", { threadId: activeThreadId });
+        void loadThreads(false);
-        }
       });
 
       socket.on("disconnect", () => {
         setSocketConnected(false);
         setTypingByThreadId({});
+        subscribedThreadIdsRef.current.clear();
       });
 
       socket.on(
@@ -185,7 +216,7 @@ export default function AdminChatPage() {
                       lastMessageAt: event.message.createdAt,
                       lastMessageBody: event.message.body,
                       unreadCount:
-                        event.threadId === activeThreadId ||
+                        event.threadId === activeThreadIdRef.current ||
                         event.message.senderUserId === userId
                           ? thread.unreadCount
                           : thread.unreadCount + 1,
@@ -263,6 +294,8 @@ export default function AdminChatPage() {
             [event.userId!]: event.lastReadMessageId!,
           },
         }));
+
+        void loadThreads(false);
       });
 
       socket.on("chat:error", (event: { code?: string; message?: string }) => {
@@ -284,28 +317,57 @@ export default function AdminChatPage() {
         socketRef.current = null;
       }
     };
-  }, [activeThreadId, getToken, userId]);
+  }, [userId]);
 
   useEffect(() => {
-    if (!activeThreadId || !socketRef.current || !socketConnected) {
+    if (!socketRef.current || !socketConnected) {
+      return;
+    }
+
+    const socket = socketRef.current;
+    const nextIds = new Set(threads.map((thread) => thread.id));
+
+    threads.forEach((thread) => {
+      if (!subscribedThreadIdsRef.current.has(thread.id)) {
+        socket.emit("chat:subscribe", { threadId: thread.id });
+        subscribedThreadIdsRef.current.add(thread.id);
+      }
+    });
+
+    Array.from(subscribedThreadIdsRef.current).forEach((threadId) => {
+      if (!nextIds.has(threadId)) {
+        socket.emit("chat:unsubscribe", { threadId });
+        subscribedThreadIdsRef.current.delete(threadId);
+      }
+    });
+  }, [socketConnected, threads]);
+
+  useEffect(() => {
+    if (!activeThreadId) {
       return;
     }
 
-    socketRef.current.emit("chat:subscribe", { threadId: activeThreadId });
     setThreads((prev) =>
       prev.map((thread) =>
         thread.id === activeThreadId ? { ...thread, unreadCount: 0 } : thread,
       ),
     );
+  }, [activeThreadId]);
+
+  useEffect(() => {
+    if (!socketConnected) {
+      const interval = setInterval(() => {
+        void loadThreads(false);
+        if (activeThreadIdRef.current) {
+          void loadMessages(activeThreadIdRef.current, false);
+        }
+      }, 5000);
 
       return () => {
-      if (socketRef.current) {
+        clearInterval(interval);
-        socketRef.current.emit("chat:unsubscribe", {
-          threadId: activeThreadId,
-        });
-      }
       };
-  }, [activeThreadId, socketConnected]);
+    }
+  }, [socketConnected]);
 
   useEffect(() => {
     if (!activeThreadId) {
@@ -319,7 +381,7 @@ export default function AdminChatPage() {
     }
 
     const markRead = async () => {
-      const token = await getToken();
+      const token = await getTokenRef.current();
       if (!token) {
         return;
       }
@@ -351,7 +413,7 @@ export default function AdminChatPage() {
     };
 
     void markRead();
-  }, [activeThreadId, getToken, messagesByThreadId, socketConnected, userId]);
+  }, [activeThreadId, messagesByThreadId, socketConnected, userId]);
 
   const onChangeDraft = (value: string) => {
     setDraft(value);
@@ -419,7 +481,7 @@ export default function AdminChatPage() {
       return;
     }
 
-    const token = await getToken();
+    const token = await getTokenRef.current();
     if (!token) {
       return;
     }

apps/mobile/src/contexts/ChatContext.tsx

@@ -74,16 +74,26 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
   >({});
 
   const socketRef = useRef<Socket | null>(null);
+  const getTokenRef = useRef(getToken);
   const lastMarkedReadByThreadRef = useRef<Record<string, string>>({});
   const activeThreadIdRef = useRef<string | null>(null);
   const currentUserIdRef = useRef<string | undefined>(undefined);
-  const refreshThreadsRef = useRef<() => Promise<void>>(async () => {});
+  const subscribedThreadIdsRef = useRef<Set<string>>(new Set());
-  const refreshMessagesRef = useRef<(threadId: string) => Promise<void>>(
+  const refreshThreadsRef = useRef<(showLoader?: boolean) => Promise<void>>(
     async () => {},
   );
+  const refreshMessagesRef = useRef<
+    (threadId: string, showLoader?: boolean) => Promise<void>
+  >(async () => {});
   const markThreadReadRef = useRef<
     (threadId: string, lastReadMessageId?: string) => Promise<void>
   >(async () => {});
+  const refreshingThreadsRef = useRef(false);
+  const refreshingMessagesRef = useRef<Record<string, boolean>>({});
+
+  useEffect(() => {
+    getTokenRef.current = getToken;
+  }, [getToken]);
 
   const clearAll = useCallback(() => {
     setThreads([]);
@@ -97,6 +107,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
     setSocketConnected(false);
     setTypingByThreadId({});
     lastMarkedReadByThreadRef.current = {};
+    subscribedThreadIdsRef.current.clear();
 
     if (socketRef.current) {
       socketRef.current.disconnect();
@@ -104,18 +115,28 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
     }
   }, []);
 
-  const refreshThreads = useCallback(async () => {
+  const refreshThreads = useCallback(
+    async (showLoader = true) => {
       if (!isSignedIn) {
         return;
       }
 
-    const token = await getToken();
+      if (refreshingThreadsRef.current) {
+        return;
+      }
+
+      refreshingThreadsRef.current = true;
+
+      const token = await getTokenRef.current();
       if (!token) {
+        refreshingThreadsRef.current = false;
         return;
       }
 
       try {
+        if (showLoader) {
           setLoadingThreads(true);
+        }
         if (isClientUser) {
           try {
             await chatApi.getMyDmThread(token);
@@ -137,23 +158,37 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
       } catch (error) {
         log.warn("Failed to refresh chat threads", { error });
       } finally {
+        if (showLoader) {
           setLoadingThreads(false);
         }
-  }, [getToken, isSignedIn, isClientUser]);
+        refreshingThreadsRef.current = false;
+      }
+    },
+    [isSignedIn, isClientUser],
+  );
 
   const refreshMessages = useCallback(
-    async (threadId: string) => {
+    async (threadId: string, showLoader = true) => {
       if (!isSignedIn) {
         return;
       }
 
-      const token = await getToken();
+      if (refreshingMessagesRef.current[threadId]) {
+        return;
+      }
+
+      refreshingMessagesRef.current[threadId] = true;
+
+      const token = await getTokenRef.current();
       if (!token) {
+        refreshingMessagesRef.current[threadId] = false;
         return;
       }
 
       try {
+        if (showLoader) {
           setLoadingMessages(true);
+        }
         const response = await chatApi.getThreadMessages(threadId, token);
         setMessagesByThreadId((prev) => ({
           ...prev,
@@ -172,10 +207,13 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
       } catch (error) {
         log.warn("Failed to refresh chat messages", { threadId, error });
       } finally {
+        if (showLoader) {
           setLoadingMessages(false);
         }
+        refreshingMessagesRef.current[threadId] = false;
+      }
     },
-    [getToken, isSignedIn],
+    [isSignedIn],
   );
 
   const loadOlderMessages = useCallback(
@@ -193,7 +231,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
         return;
       }
 
-      const token = await getToken();
+      const token = await getTokenRef.current();
       if (!token) {
         return;
       }
@@ -235,7 +273,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
         setLoadingOlderByThreadId((prev) => ({ ...prev, [threadId]: false }));
       }
     },
-    [getToken, isSignedIn, loadingOlderByThreadId, nextCursorByThreadId],
+    [isSignedIn, loadingOlderByThreadId, nextCursorByThreadId],
   );
 
   const markThreadRead = useCallback(
@@ -244,7 +282,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
         return;
       }
 
-      const token = await getToken();
+      const token = await getTokenRef.current();
       if (!token) {
         return;
       }
@@ -271,7 +309,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
         log.warn("Failed to mark thread read", { threadId, error });
       }
     },
-    [getToken, isSignedIn, user?.id],
+    [isSignedIn, user?.id],
   );
 
   useEffect(() => {
@@ -301,7 +339,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
         return;
       }
 
-      const token = await getToken();
+      const token = await getTokenRef.current();
       if (!token) {
         return;
       }
@@ -360,7 +398,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
         throw error;
       }
     },
-    [getToken, isSignedIn, socketConnected, user?.id],
+    [isSignedIn, socketConnected, user?.id],
   );
 
   useEffect(() => {
@@ -372,7 +410,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
     let mounted = true;
 
     const setupSocket = async () => {
-      const token = await getToken();
+      const token = await getTokenRef.current();
       if (!token || !mounted) {
         return;
       }
@@ -382,7 +420,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
 
       socket.on("connect", () => {
         setSocketConnected(true);
-        void refreshThreadsRef.current();
+        void refreshThreadsRef.current(false);
 
         if (activeThreadIdRef.current) {
           socket.emit("chat:subscribe", {
@@ -394,6 +432,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
       socket.on("disconnect", () => {
         setSocketConnected(false);
         setTypingByThreadId({});
+        subscribedThreadIdsRef.current.clear();
       });
 
       socket.on(
@@ -460,6 +499,8 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
                   : thread,
               ),
             );
+          } else if (event.threadId !== activeThreadIdRef.current) {
+            void refreshThreadsRef.current(false);
           }
         },
       );
@@ -523,7 +564,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
             },
           }));
 
-          void refreshThreadsRef.current();
+          void refreshThreadsRef.current(false);
         },
       );
 
@@ -575,14 +616,14 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
         socketRef.current = null;
       }
     };
-  }, [clearAll, getToken, isSignedIn]);
+  }, [clearAll, isSignedIn]);
 
   useEffect(() => {
     if (!isSignedIn) {
       return;
     }
 
-    void refreshThreadsRef.current();
+    void refreshThreadsRef.current(true);
   }, [isSignedIn, userRole]);
 
   useEffect(() => {
@@ -594,7 +635,7 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
       socketRef.current.emit("chat:subscribe", { threadId: activeThreadId });
     }
 
-    void refreshMessagesRef.current(activeThreadId);
+    void refreshMessagesRef.current(activeThreadId, true);
 
     setThreads((prev) =>
       prev.map((thread) =>
@@ -611,6 +652,46 @@ export function ChatProvider({ children }: { children: React.ReactNode }) {
     };
   }, [activeThreadId, socketConnected]);
 
+  useEffect(() => {
+    if (!socketRef.current || !socketConnected) {
+      return;
+    }
+
+    const socket = socketRef.current;
+    const nextIds = new Set(threads.map((thread) => thread.id));
+
+    threads.forEach((thread) => {
+      if (!subscribedThreadIdsRef.current.has(thread.id)) {
+        socket.emit("chat:subscribe", { threadId: thread.id });
+        subscribedThreadIdsRef.current.add(thread.id);
+      }
+    });
+
+    Array.from(subscribedThreadIdsRef.current).forEach((threadId) => {
+      if (!nextIds.has(threadId)) {
+        socket.emit("chat:unsubscribe", { threadId });
+        subscribedThreadIdsRef.current.delete(threadId);
+      }
+    });
+  }, [socketConnected, threads]);
+
+  useEffect(() => {
+    if (!isSignedIn || socketConnected) {
+      return;
+    }
+
+    const interval = setInterval(() => {
+      void refreshThreadsRef.current(false);
+      if (activeThreadIdRef.current) {
+        void refreshMessagesRef.current(activeThreadIdRef.current, false);
+      }
+    }, 5000);
+
+    return () => {
+      clearInterval(interval);
+    };
+  }, [isSignedIn, socketConnected]);
+
   useEffect(() => {
     if (!activeThreadId) {
       return;