feat: real-time thread synchronization and event listeners (#1350)

packages/api/src/EmbeddedChatApi.ts

@@ -14,8 +14,14 @@ export default class EmbeddedChatApi {
   rid: string;
   sdk: DDPSDK;
   onMessageCallbacks: ((message: any) => void)[];
-  onMessageDeleteCallbacks: ((messageId: string) => void)[];
+  onMessageDeleteCallbacks: ((messageId: string, threadId?: string) => void)[];
+  onThreadMessageCallbacks: ((message: any) => void)[];
+  onThreadMessageDeleteCallbacks: ((
+    messageId: string,
+    threadId?: string
+  ) => void)[];
   onTypingStatusCallbacks: ((users: string[]) => void)[];
+
   onActionTriggeredCallbacks: ((data: any) => void)[];
   onUiInteractionCallbacks: ((data: any) => void)[];
   typingUsers: string[];
@@ -37,7 +43,10 @@ export default class EmbeddedChatApi {
     });
     this.onMessageCallbacks = [];
     this.onMessageDeleteCallbacks = [];
+    this.onThreadMessageCallbacks = [];
+    this.onThreadMessageDeleteCallbacks = [];
     this.onTypingStatusCallbacks = [];
+
     this.typingUsers = [];
     this.onActionTriggeredCallbacks = [];
     this.onUiInteractionCallbacks = [];
@@ -186,7 +195,7 @@ export default class EmbeddedChatApi {
         await this.updateUserUsername(data.userId, data.me.name);
       }
       return { status: "success", me: data.me };
-    } catch (error) {
+    } catch (error: any) {
       if (error instanceof ApiError && error.response?.status === 401) {
         const authErrorRes = await error.response.json();
         return { error: authErrorRes?.error };
@@ -235,8 +244,8 @@ export default class EmbeddedChatApi {
   }
 
   /**
-   * All subscriptions are implemented here.
-   * TODO: Add logic to call thread message event listeners. To be done after thread implementation
+   * Initializes all realtime subscriptions (room messages, thread messages,
+   * typing notifications, delete events, and user interactions).
    */
   async connect(isChannelPrivate = false) {
     // Guard against concurrent connect() calls (e.g. React StrictMode double-invoke)
@@ -285,14 +294,19 @@ export default class EmbeddedChatApi {
         await this.joinRoom(isChannelPrivate);
       }
 
-      // Subscribe to room messages
+      // Subscribe to room messages and thread replies
       const roomMsgSub = this.sdk.stream(
         "room-messages",
         [this.rid, false],
         (data: any) => {
           const message = this._normalizeMessage(data);
           if (message) {
             this.onMessageCallbacks.forEach((callback) => callback(message));
+            if (message.tmid) {
+              this.onThreadMessageCallbacks.forEach((callback) =>
+                callback(message)
+              );
+            }
           }
         }
       );
@@ -321,11 +335,17 @@ export default class EmbeddedChatApi {
         "notify-room",
         [`${this.rid}/deleteMessage`, false],
         (...args: any[]) => {
-          const messageId = args[0]?._id;
+          const messageData = args[0];
+          const messageId = messageData?._id;
           if (messageId) {
             this.onMessageDeleteCallbacks.forEach((callback) =>
-              callback(messageId)
+              callback(messageId, messageData?.tmid)
             );
+            if (messageData?.tmid) {
+              this.onThreadMessageDeleteCallbacks.forEach((callback) =>
+                callback(messageId, messageData.tmid)
+              );
+            }
           }
         }
       );
@@ -379,7 +399,9 @@ export default class EmbeddedChatApi {
     );
   }
 
-  async addMessageDeleteListener(callback: (messageId: string) => void) {
+  async addMessageDeleteListener(
+    callback: (messageId: string, threadId?: string) => void
+  ) {
     const idx = this.onMessageDeleteCallbacks.findIndex((c) => c === callback);
     if (idx !== -1) {
       this.onMessageDeleteCallbacks[idx] = callback;
@@ -388,12 +410,49 @@ export default class EmbeddedChatApi {
     }
   }
 
-  async removeMessageDeleteListener(callback: (messageId: string) => void) {
+  async removeMessageDeleteListener(
+    callback: (messageId: string, threadId?: string) => void
+  ) {
     this.onMessageDeleteCallbacks = this.onMessageDeleteCallbacks.filter(
       (c) => c !== callback
     );
   }
 
+  async addThreadMessageListener(callback: (message: any) => void) {
+    const idx = this.onThreadMessageCallbacks.findIndex((c) => c === callback);
+    if (idx !== -1) {
+      this.onThreadMessageCallbacks[idx] = callback;
+    } else {
+      this.onThreadMessageCallbacks.push(callback);
+    }
+  }
+
+  async removeThreadMessageListener(callback: (message: any) => void) {
+    this.onThreadMessageCallbacks = this.onThreadMessageCallbacks.filter(
+      (c) => c !== callback
+    );
+  }
+
+  async addThreadMessageDeleteListener(
+    callback: (messageId: string, threadId?: string) => void
+  ) {
+    const idx = this.onThreadMessageDeleteCallbacks.findIndex(
+      (c) => c === callback
+    );
+    if (idx !== -1) {
+      this.onThreadMessageDeleteCallbacks[idx] = callback;
+    } else {
+      this.onThreadMessageDeleteCallbacks.push(callback);
+    }
+  }
+
+  async removeThreadMessageDeleteListener(
+    callback: (messageId: string, threadId?: string) => void
+  ) {
+    this.onThreadMessageDeleteCallbacks =
+      this.onThreadMessageDeleteCallbacks.filter((c) => c !== callback);
+  }
+
   async addTypingStatusListener(callback: (users: string[]) => void) {
     const idx = this.onTypingStatusCallbacks.findIndex((c) => c === callback);
     if (idx !== -1) {

packages/react/package.json

@@ -16,6 +16,7 @@
     "build": "cross-env NODE_OPTIONS=--max-old-space-size=4096 rollup -c --context=window --environment NODE_ENV:production",
     "watch": "cross-env NODE_OPTIONS=--max-old-space-size=4096 rollup -c --watch --context=window",
     "test:lint": "eslint src/**/*.js",
+    "test": "node --test src/store/messageStore.test.mjs",
     "format": "prettier --write 'src/' ",
     "format:check": "prettier --check 'src/' ",
     "lint": "eslint 'src/**/*.js'",

packages/react/src/store/messageStore.js

@@ -36,55 +36,154 @@ const useMessageStore = create((set, get) => ({
     }),
   upsertMessage: (message, enableThreads = false) => {
     if (message.tmid && enableThreads) {
-      if (get().threadMainMessage?._id === message.tmid) {
-        set((state) => ({
-          threadMessages: upsertMessage(state.threadMessages, message),
-        }));
-      }
+      const parentId = message.tmid;
+      const isOpenThread = get().threadMainMessage?._id === parentId;
+      const isExistingInThread = get().threadMessages.some(
+        (m) => m._id === message._id
+      );
+
+      set((state) => {
+        const nextState = {};
+
+        if (isOpenThread) {
+          nextState.threadMessages = upsertMessage(
+            state.threadMessages,
+            message
+          );
+          if (state.threadMainMessage) {
+            nextState.threadMainMessage = {
+              ...state.threadMainMessage,
+              tcount:
+                (state.threadMainMessage.tcount || 0) +
+                (isExistingInThread ? 0 : 1),
+              tlm: message.ts || new Date().toISOString(),
+            };
+          }
+        }
+
+        const parentIdx = state.messages.findIndex((m) => m._id === parentId);
+        if (parentIdx !== -1) {
+          const parent = state.messages[parentIdx];
+          const newReplies = parent.replies
+            ? message.u?._id && !parent.replies.includes(message.u._id)
+              ? [...parent.replies, message.u._id]
+              : parent.replies
+            : message.u?._id
+            ? [message.u._id]
+            : [];
+
+          const updatedParent = {
+            ...parent,
+            tcount: (parent.tcount || 0) + (isExistingInThread ? 0 : 1),
+            tlm: message.ts || new Date().toISOString(),
+            replies: newReplies,
+          };
+
+          const newMessages = cloneArray(state.messages);
+          newMessages[parentIdx] = updatedParent;
+          nextState.messages = newMessages;
+        }
+
+        return nextState;
+      });
     } else {
-      set((state) => ({
-        messages: upsertMessage(state.messages, message),
-      }));
+      set((state) => {
+        const isMainThreadMessage =
+          state.threadMainMessage?._id === message._id;
+        return {
+          messages: upsertMessage(state.messages, message),
+          ...(isMainThreadMessage
+            ? {
+                threadMainMessage: {
+                  ...state.threadMainMessage,
+                  ...message,
+                },
+              }
+            : {}),
+        };
+      });
     }
   },
-  removeMessage: (messageId) => {
+  removeMessage: (messageId, parentId) => {
     const threadMessage = get().threadMessages.find((m) => m._id === messageId);
     const message = get().messages.find((m) => m._id === messageId);
-    if (threadMessage) {
+    const effectiveParentId = parentId || threadMessage?.tmid;
+
+    if (effectiveParentId) {
       return set((state) => ({
-        deletedMessage: threadMessage,
+        deletedMessage: threadMessage || message || { _id: messageId },
         threadMessages: cloneArray(state.threadMessages).filter(
           (m) => m._id !== messageId
         ),
+        threadMainMessage:
+          state.threadMainMessage?._id === effectiveParentId
+            ? {
+                ...state.threadMainMessage,
+                tcount: Math.max(0, (state.threadMainMessage.tcount || 1) - 1),
+              }
+            : state.threadMainMessage,
+        messages: cloneArray(state.messages)
+          .filter((m) => m._id !== messageId)
+          .map((m) => {
+            if (m._id !== effectiveParentId) return m;
+            return {
+              ...m,
+              tcount: Math.max(0, (m.tcount || 1) - 1),
+            };
+          }),
       }));
     }
-    if (message) {
+
+    if (message || get().threadMainMessage?._id === messageId) {
+      const shouldCloseThread = get().threadMainMessage?._id === messageId;
       return set((state) => ({
-        deletedMessage: message,
+        deletedMessage: message || { _id: messageId },
         messages: cloneArray(state.messages).filter((m) => m._id !== messageId),
+        ...(shouldCloseThread
+          ? {
+              isThreadOpen: false,
+              threadMainMessage: null,
+              threadMessages: [],
+            }
+          : {}),
       }));
     }
   },
   replaceMessage: (oldMessageId, newMessage) => {
-    const threadMessage = get().threadMessages.find(
-      (m) => m._id === oldMessageId
-    );
-    const message = get().messages.find((m) => m._id === oldMessageId);
-    if (threadMessage) {
-      return set((state) => ({
-        threadMessages: cloneArray(state.threadMessages).map((m) =>
-          m._id === oldMessageId ? newMessage : m
-        ),
-      }));
-    }
-    if (message) {
-      return set((state) => ({
-        messages: cloneArray(state.messages).map((m) =>
-          m._id === oldMessageId ? newMessage : m
-        ),
-      }));
-    }
+    set((state) => {
+      const isThreadMsg = state.threadMessages.some(
+        (m) => m._id === oldMessageId
+      );
+      const isMainMsg = state.messages.some((m) => m._id === oldMessageId);
+      const isMainThreadMsg = state.threadMainMessage?._id === oldMessageId;
+
+      return {
+        ...(isThreadMsg
+          ? {
+              threadMessages: cloneArray(state.threadMessages).map((m) =>
+                m._id === oldMessageId ? newMessage : m
+              ),
+            }
+          : {}),
+        ...(isMainMsg
+          ? {
+              messages: cloneArray(state.messages).map((m) =>
+                m._id === oldMessageId ? newMessage : m
+              ),
+            }
+          : {}),
+        ...(isMainThreadMsg
+          ? {
+              threadMainMessage: {
+                ...state.threadMainMessage,
+                ...newMessage,
+              },
+            }
+          : {}),
+      };
+    });
   },
+
   setEditMessage: (editMessage) => set(() => ({ editMessage })),
   setMessagesOffset: (newOffset) => set(() => ({ messagesOffset: newOffset })),
   editMessagePermissions: {},