fix(nats): Implemented new system pub/sub core channel and moved sendAnalyticsData to use it.

jibon57 · jibon57 · commit f583fc8653f5 · 2026-04-02T21:17:46.000+02:00
diff --git a/src/components/whiteboard/helpers/handleRequests.ts b/src/components/whiteboard/helpers/handleRequests.ts
@@ -10,6 +10,7 @@ import {
   AnalyticsEventType,
   DataMsgBodyType,
 } from 'plugnmeet-protocol-js';
+import { throttle } from 'es-toolkit';
 
 import { store } from '../../../store';
 import { updateRequestedWhiteboardData } from '../../../store/slices/whiteboard';
@@ -23,7 +24,29 @@ const broadcastedElementVersions: Map<string, number> = new Map(),
   DELETED_ELEMENT_TIMEOUT = 3 * 60 * 60 * 1000; // 3 hours
 let preScrollX = 0,
   preScrollY = 0,
-  conn: ConnectNats;
+  conn: ConnectNats,
+  annotatedAnalyticsBatchCounter = 0;
+
+const flushAnnotatedAnalyticsData = () => {
+  if (annotatedAnalyticsBatchCounter > 0) {
+    if (!conn) {
+      conn = getNatsConn();
+    }
+    conn.sendAnalyticsData(
+      AnalyticsEvents.ANALYTICS_EVENT_USER_WHITEBOARD_ANNOTATED,
+      AnalyticsEventType.USER,
+      '',
+      '',
+      annotatedAnalyticsBatchCounter.toString(),
+    );
+    annotatedAnalyticsBatchCounter = 0; // Reset after sending
+  }
+};
+
+// Initialize the throttled function once at module load
+const throttledFlushAnalytics = throttle(flushAnnotatedAnalyticsData, 5000, {
+  edges: ['trailing'],
+});
 
 export const sendRequestedForWhiteboardData = async () => {
   if (!conn) {
@@ -171,13 +194,9 @@ export const broadcastScreenDataByNats = async (
     JSON.stringify(elements),
     sendTo,
   );
-  conn.sendAnalyticsData(
-    AnalyticsEvents.ANALYTICS_EVENT_USER_WHITEBOARD_ANNOTATED,
-    AnalyticsEventType.USER,
-    '',
-    '',
-    '1',
-  );
+  annotatedAnalyticsBatchCounter++;
+  // Trigger the throttled flush whenever there's activity
+  throttledFlushAnalytics();
 };
 
 export const broadcastCurrentPageNumber = async (
diff --git a/src/helpers/nats/ConnectNats.ts b/src/helpers/nats/ConnectNats.ts
@@ -58,7 +58,7 @@ const RENEW_TOKEN_FREQUENT = 3 * 60 * 1000,
   PING_INTERVAL = 10 * 1000,
   STATUS_CHECKER_INTERVAL = 500,
   USERS_SYNC_INTERVAL = 30 * 1000,
-  MAX_MISSED_PONGS = 6;
+  MAX_MISSED_PONGS = 12;
 export type PrivateDataDeliveryType = 'CHAT' | 'DATA_MSG';
 
 export default class ConnectNats {
@@ -263,6 +263,7 @@ export default class ConnectNats {
         pingInterval: PING_INTERVAL,
         noEcho: true,
       });
+      this.messageQueue.setNc(this._nc);
 
       console.info(`connected ${this._nc.getServer()}`);
     } catch (e) {
@@ -431,12 +432,19 @@ export default class ConnectNats {
     }
   }
 
+  /**
+   * Sends a message to the system worker with guaranteed delivery.
+   * This method uses JetStream to ensure the message is received by the server.
+   * It is used for critical messages like PINGs, token renewals, and private data delivery.
+   * @param data The message to send.
+   */
   public sendMessageToSystemWorker = (data: NatsMsgClientToServer) => {
     const subject =
       this._subjects.systemJsWorker + '.' + this._roomId + '.' + this._userId;
     this.messageQueue.addToQueue({
       subject,
       payload: toBinary(NatsMsgClientToServerSchema, data),
+      useJetStream: true,
     });
   };
 
@@ -495,6 +503,12 @@ export default class ConnectNats {
     return undefined;
   }
 
+  /**
+   * Sends a chat message.
+   * Private messages are sent via the system worker with JetStream's guaranteed delivery.
+   * Public messages are sent as fire-and-forget core NATS messages to the public chat subject.
+   * Both are managed by the MessageQueue.
+   */
   public sendChatMsg = async (to: string, msg: string) => {
     if (!this._nc || this._nc.isClosed()) {
       return;
@@ -551,7 +565,10 @@ export default class ConnectNats {
       this.sendPrivateData(payload, 'CHAT', to, false);
     } else {
       const subject = `${this._subjects.chat}.${this._roomId}`;
-      this._nc.publish(subject, payload);
+      this.messageQueue.addToQueue({
+        subject,
+        payload,
+      });
     }
 
     if (isPrivate) {
@@ -576,6 +593,11 @@ export default class ConnectNats {
     await this.subscriptionHandler.handleChat.handleMsg(chatMessage);
   };
 
+  /**
+   * Sends whiteboard data as a fire-and-forget message.
+   * This method uses the core NATS `publish` method via the `MessageQueue`
+   * to avoid blocking critical messages.
+   */
   public sendWhiteboardData = async (
     type: DataMsgBodyType,
     msg: string,
@@ -602,11 +624,17 @@ export default class ConnectNats {
     }
 
     const subject = `${this._subjects.whiteboard}.${this._roomId}`;
-    this._nc.publish(subject, payload);
+    this.messageQueue.addToQueue({
+      subject,
+      payload,
+    });
   };
 
   /**
-   * sendDataMessage method mostly use to communicate between clients
+   * Sends a generic data message.
+   * Private messages are sent via the system worker with JetStream's guaranteed delivery.
+   * Public messages are sent as fire-and-forget core NATS messages to the public data channel subject.
+   * Both are managed by the MessageQueue.
    */
   public sendDataMessage = async (
     type: DataMsgBodyType,
@@ -637,10 +665,19 @@ export default class ConnectNats {
       this.sendPrivateData(payload, 'DATA_MSG', to, false);
     } else {
       const subject = `${this._subjects.dataChannel}.${this._roomId}`;
-      this._nc.publish(subject, payload);
+      this.messageQueue.addToQueue({
+        subject,
+        payload,
+      });
     }
   };
 
+  /**
+   * Sends analytics data to the server as a lightweight, fire-and-forget message.
+   * This method uses the core NATS `publish` method via the `MessageQueue`
+   * by publishing to the `systemJsWorker` subject without the `useJetStream` flag.
+   * This avoids blocking critical JetStream messages.
+   */
   public sendAnalyticsData = (
     event_name: AnalyticsEvents,
     event_type: AnalyticsEventType = AnalyticsEventType.USER,
@@ -662,7 +699,13 @@ export default class ConnectNats {
       event: NatsMsgClientToServerEvents.PUSH_ANALYTICS_DATA,
       msg: toJsonString(AnalyticsDataMsgSchema, analyticsMsg),
     });
-    this.sendMessageToSystemWorker(data);
+
+    const subject =
+      this._subjects.systemJsWorker + '.' + this._roomId + '.' + this._userId;
+    this.messageQueue.addToQueue({
+      subject,
+      payload: toBinary(NatsMsgClientToServerSchema, data),
+    });
   };
 
   private startTokenRenewInterval() {
@@ -686,7 +729,7 @@ export default class ConnectNats {
 
   private startPingToServer() {
     const ping = async () => {
-      if (this.missedPongs === 3) {
+      if (this.missedPongs === 6) {
         this.pongMissedToastId = toast.loading(
           i18n.t('notifications.server-not-responding'),
           {
diff --git a/src/helpers/nats/MessageQueue.ts b/src/helpers/nats/MessageQueue.ts
@@ -1,5 +1,5 @@
 import { JetStreamClient, JetStreamError } from '@nats-io/jetstream';
-import { errors } from '@nats-io/nats-core';
+import { errors, NatsConnection } from '@nats-io/nats-core';
 
 import { formatNatsError } from '../utils';
 import { store } from '../../store';
@@ -9,10 +9,13 @@ import i18n from '../i18n';
 interface Message {
   subject: string;
   payload: any;
+  useJetStream?: boolean;
+  retries?: number;
 }
 
 // A Map where the key is the full, unique NATS subject.
 type SubjectQueues = Map<string, Array<Message>>;
+const MAX_RETRIES = 2; // Max number of retries for transient errors
 
 /**
  * A message queue that processes messages in independent, parallel queues
@@ -23,10 +26,10 @@ type SubjectQueues = Map<string, Array<Message>>;
 export default class MessageQueue {
   private _isConnected: boolean = false;
   private _js: JetStreamClient | undefined;
+  private _nc: NatsConnection | undefined;
 
   private readonly queues: SubjectQueues = new Map();
   private readonly processingSubjects: Set<string> = new Set();
-
   private _isHoldingNotificationShown = false;
 
   public setIsConnected = (value: boolean) => {
@@ -42,6 +45,10 @@ export default class MessageQueue {
     this._js = value;
   };
 
+  public setNc = (value: NatsConnection) => {
+    this._nc = value;
+  };
+
   public addToQueue = (message: Message) => {
     const { subject } = message;
 
@@ -65,62 +72,99 @@ export default class MessageQueue {
     }
 
     this.processingSubjects.add(subject);
-
     const request = subjectQueue[0];
 
     try {
-      if (!this._js) {
-        return;
+      if (request.useJetStream) {
+        // Assuming this._js is always available if _isConnected is true
+        await this._js!.publish(request.subject, request.payload);
+      } else {
+        // Assuming this._nc is always available if _isConnected is true
+        this._nc!.publish(request.subject, request.payload);
       }
 
-      await this._js.publish(request.subject, request.payload);
+      // Message sent successfully.
       subjectQueue.shift();
+      this.processingSubjects.delete(subject);
       this._isHoldingNotificationShown = false;
+      if (subjectQueue.length > 0) {
+        this.processSubjectQueue(subject).then();
+      }
     } catch (e: any) {
+      const formattedError = formatNatsError(e); // Format the error here
+
+      // Check for transient errors that are eligible for retry.
       if (
-        e instanceof errors.TimeoutError ||
-        e instanceof errors.NoRespondersError ||
-        e instanceof JetStreamError
+        request.useJetStream && // Only retry JetStream messages
+        (e instanceof errors.TimeoutError ||
+          e instanceof errors.NoRespondersError ||
+          e instanceof JetStreamError)
       ) {
-        console.error(
-          `NATS transient error for subject '${subject}': ${e.message}. Holding queue.`,
-        );
-        if (e.message.includes('connection draining')) {
-          return;
-        }
-        if (!this._isHoldingNotificationShown) {
-          const msg = formatNatsError(e);
+        request.retries = (request.retries || 0) + 1;
+
+        if (request.retries > MAX_RETRIES) {
+          // Exceeded max retries, discard the message.
+          console.error(
+            `NATS message for subject '${subject}' failed after ${MAX_RETRIES} retries. Discarding.`,
+            { error: formattedError, message: request },
+          );
           store.dispatch(
             addUserNotification({
-              message: i18n.t('notifications.queue-holding-messages', {
-                error: msg,
+              message: i18n.t('notifications.queue-discarded-message', {
+                error: formattedError,
               }),
-              typeOption: 'warning',
+              typeOption: 'error',
             }),
           );
-          this._isHoldingNotificationShown = true;
+
+          subjectQueue.shift(); // Discard
+          this.processingSubjects.delete(subject);
+          if (subjectQueue.length > 0) {
+            this.processSubjectQueue(subject).then();
+          }
+        } else {
+          // Retry with a delay.
+          console.warn(
+            `NATS transient error for subject '${subject}'. Retrying (${request.retries}/${MAX_RETRIES})...`,
+            { error: formattedError },
+          );
+          if (!this._isHoldingNotificationShown) {
+            store.dispatch(
+              addUserNotification({
+                message: i18n.t('notifications.queue-holding-messages', {
+                  error: formattedError,
+                }),
+                typeOption: 'warning',
+              }),
+            );
+            this._isHoldingNotificationShown = true;
+          }
+
+          this.processingSubjects.delete(subject);
+          setTimeout(() => {
+            this.processSubjectQueue(subject).then();
+          }, 500 * request.retries); // Simple exponential backoff
         }
-        return;
       } else {
+        // This is a non-recoverable or non-JetStream error. Discard immediately.
         console.error(
-          `Found poison message for subject '${subject}'. Discarding.`,
-          { error: e.message, message: request },
+          `Found poison message or non-retryable error for subject '${subject}'. Discarding.`,
+          { error: formattedError, message: request }, // Use formatted error
         );
         store.dispatch(
           addUserNotification({
             message: i18n.t('notifications.queue-discarded-message', {
-              error: e.message,
+              error: formattedError, // Use formatted error
             }),
             typeOption: 'error',
           }),
         );
-        subjectQueue.shift();
-      }
-    } finally {
-      this.processingSubjects.delete(subject);
 
-      if (subjectQueue.length > 0) {
-        this.processSubjectQueue(subject).then();
+        subjectQueue.shift(); // Discard
+        this.processingSubjects.delete(subject);
+        if (subjectQueue.length > 0) {
+          this.processSubjectQueue(subject).then();
+        }
       }
     }
   }
