Enhance UpdateManager and SocketUser classes to improve subscription management and disconnection handling.

- Added `removeAllListeners` method in UpdateManager to clear all subscriptions efficiently.
- Updated SocketUser's `handleDisconnect` method to log user disconnection more accurately based on user presence.
- Introduced subscription tracking in UpdateManager to manage active subscriptions.

src/socket/socketuser.js

@@ -271,9 +271,17 @@ export class SocketUser {
 
   async handleDisconnect() {
     await this.actionManager.removeAllListeners();
+    await this.updateManager.removeAllListeners();
     await this.eventManager.removeAllListeners();
     await this.statsManager.removeAllListeners();
     await this.notificationManager.removeAllListeners();
-    logger.info('External user disconnected:', this.socket.user?.username);
+    if (this.user || this.socket.user) {
+      logger.info(
+        'External user disconnected:',
+        this.user.username || this.socket.user?.username
+      );
+    } else {
+      logger.info('External user disconnected.');
+    }
   }
 }

src/updates/__tests__/updatemanager.test.js

@@ -211,4 +211,19 @@ describe('UpdateManager', () => {
       );
     });
   });
+
+  describe('removeAllListeners', () => {
+    it('should remove all subscriptions', async () => {
+      await updateManager.subscribeToObjectNew('printer');
+      await updateManager.subscribeToObjectDelete('printer');
+      await updateManager.subscribeToObjectUpdate('123', 'printer');
+
+      expect(updateManager.subscriptions.size).toBe(3);
+
+      await updateManager.removeAllListeners();
+
+      expect(natsServer.removeSubscription).toHaveBeenCalledTimes(3);
+      expect(updateManager.subscriptions.size).toBe(0);
+    });
+  });
 });

src/updates/updatemanager.js

@@ -78,12 +78,15 @@ const stableStringify = value => {
 const getSubscriptionOwner = (socketId, filter) =>
   `${socketId}:${stableStringify(normalizeFilter(filter))}`;
 
+const getSubscriptionKey = (subject, owner) => `${subject}:${owner}`;
+
 /**
  * UpdateManager handles tracking object updates and broadcasts update events via websockets.
  */
 export class UpdateManager {
   constructor(socketClient) {
     this.socketClient = socketClient;
+    this.subscriptions = new Set();
   }
 
   matchesObjectTypeFilter(objectType, filter, value) {
@@ -116,10 +119,13 @@ export class UpdateManager {
 
   async subscribeToObjectNew(objectType, filter = {}) {
     const normalizedFilter = normalizeFilter(filter);
-    await natsServer.subscribe(
+    const subject = `${objectType}s.new`;
-      `${objectType}s.new`,
+    const owner = getSubscriptionOwner(
-      getSubscriptionOwner(this.socketClient.socketId, normalizedFilter),
+      this.socketClient.socketId,
-      async (key, value) => {
+      normalizedFilter
+    );
+
+    await natsServer.subscribe(subject, owner, async (key, value) => {
       logger.trace('Object new event:', value);
       this.emitObjectTypeEvent(
         'objectNew',
@@ -127,17 +133,21 @@ export class UpdateManager {
         normalizedFilter,
         value
       );
-      }
+    });
-    );
+
+    this.subscriptions.add(getSubscriptionKey(subject, owner));
     return { success: true };
   }
 
   async subscribeToObjectDelete(objectType, filter = {}) {
     const normalizedFilter = normalizeFilter(filter);
-    await natsServer.subscribe(
+    const subject = `${objectType}s.delete`;
-      `${objectType}s.delete`,
+    const owner = getSubscriptionOwner(
-      getSubscriptionOwner(this.socketClient.socketId, normalizedFilter),
+      this.socketClient.socketId,
-      async (key, value) => {
+      normalizedFilter
+    );
+
+    await natsServer.subscribe(subject, owner, async (key, value) => {
       logger.trace('Object delete event:', value);
       this.emitObjectTypeEvent(
         'objectDelete',
@@ -145,17 +155,18 @@ export class UpdateManager {
         normalizedFilter,
         value
       );
-      }
+    });
-    );
+
+    this.subscriptions.add(getSubscriptionKey(subject, owner));
     return { success: true };
   }
 
   async subscribeToObjectUpdate(id, objectType) {
     logger.debug('Subscribing to object update...', id, objectType);
-    await natsServer.subscribe(
+    const subject = `${objectType}s.${id}.object`;
-      `${objectType}s.${id}.object`,
+    const owner = this.socketClient.socketId;
-      this.socketClient.socketId,
+
-      (key, value) => {
+    await natsServer.subscribe(subject, owner, (key, value) => {
       const expandedValue = expandObjectIds(value);
       logger.trace('Object update event:', id);
       this.socketClient.socket.emit('objectUpdate', {
@@ -163,32 +174,53 @@ export class UpdateManager {
         objectType: objectType,
         object: { ...expandedValue }
       });
-      }
+    });
-    );
+
+    this.subscriptions.add(getSubscriptionKey(subject, owner));
     return { success: true };
   }
 
   async removeObjectNewListener(objectType, filter = {}) {
-    await natsServer.removeSubscription(
+    const subject = `${objectType}s.new`;
-      `${objectType}s.new`,
+    const owner = getSubscriptionOwner(this.socketClient.socketId, filter);
-      getSubscriptionOwner(this.socketClient.socketId, filter)
+
-    );
+    await natsServer.removeSubscription(subject, owner);
+    this.subscriptions.delete(getSubscriptionKey(subject, owner));
     return { success: true };
   }
 
   async removeObjectDeleteListener(objectType, filter = {}) {
-    await natsServer.removeSubscription(
+    const subject = `${objectType}s.delete`;
-      `${objectType}s.delete`,
+    const owner = getSubscriptionOwner(this.socketClient.socketId, filter);
-      getSubscriptionOwner(this.socketClient.socketId, filter)
+
-    );
+    await natsServer.removeSubscription(subject, owner);
+    this.subscriptions.delete(getSubscriptionKey(subject, owner));
     return { success: true };
   }
 
   async removeObjectUpdateListener(id, objectType) {
-    await natsServer.removeSubscription(
+    const subject = `${objectType}s.${id}.object`;
-      `${objectType}s.${id}.object`,
+    const owner = this.socketClient.socketId;
-      this.socketClient.socketId
+
+    await natsServer.removeSubscription(subject, owner);
+    this.subscriptions.delete(getSubscriptionKey(subject, owner));
+    return { success: true };
+  }
+
+  async removeAllListeners() {
+    logger.debug('Removing all update listeners...');
+    const removePromises = Array.from(this.subscriptions).map(
+      subscriptionKey => {
+        const separatorIndex = subscriptionKey.indexOf(':');
+        const subject = subscriptionKey.slice(0, separatorIndex);
+        const owner = subscriptionKey.slice(separatorIndex + 1);
+        return natsServer.removeSubscription(subject, owner);
+      }
     );
+
+    await Promise.all(removePromises);
+    this.subscriptions.clear();
+    logger.debug(`Removed ${removePromises.length} update listener(s)`);
     return { success: true };
   }
 }