fix: ws double messages (#2700)

src/backend/src/modules/broadcast/BroadcastService.js

@@ -103,7 +103,7 @@ export class BroadcastService extends BaseService {
         const safeMeta = this.#normalizeMeta(meta);
         const outboundEvent = { key, data, meta: safeMeta };
 
-        // Mirror local outer.gui/pub events to Redis so same-cluster replicas
+        // Mirror local outer.pub events to Redis so same-cluster replicas
         // receive them even when this instance is the originator.
         this.#publishWebhookEventsToRedis([outboundEvent]).catch(error => {
             console.warn('local redis pubsub publish failed', { error, key });
@@ -171,6 +171,12 @@ export class BroadcastService extends BaseService {
         return meta;
     }
 
+    #resolveLocalPeerId () {
+        const localPeerId = this.config?.webhook?.peerId ?? this.config?.webhook?.key;
+        if ( typeof localPeerId !== 'string' || localPeerId.trim() === '' ) return null;
+        return localPeerId.trim();
+    }
+
     #resolvePeerId (peerConfig) {
         if ( !peerConfig || typeof peerConfig !== 'object' ) return null;
         const peerId = peerConfig.peerId ?? peerConfig.key;
@@ -213,9 +219,7 @@ export class BroadcastService extends BaseService {
 
     #isRedisWebhookEventKey (key) {
         if ( typeof key !== 'string' ) return false;
-        return key === 'outer.gui' ||
-            key.startsWith('outer.gui.') ||
-            key === 'outer.pub' ||
+        return key === 'outer.pub' ||
             key.startsWith('outer.pub.');
     }
 
@@ -377,6 +381,11 @@ export class BroadcastService extends BaseService {
             res.status(403).send({ error: { message: 'Missing X-Broadcast-Peer-Id' } });
             return;
         }
+        const localPeerId = this.#resolveLocalPeerId();
+        if ( localPeerId && peerId === localPeerId ) {
+            res.status(200).send({ ok: true, ignored: 'self-peer' });
+            return;
+        }
 
         const peer = this.#peersByKey[peerId];
         if ( !peer || !peer.webhook_secret ) {
@@ -437,7 +446,7 @@ export class BroadcastService extends BaseService {
         this.#incomingLastNonceByPeer.set(peerId, { timestamp, nonce });
 
         await this.#publishWebhookEventsToRedis(incomingEvents);
-        this.#emitIncomingEventsSequentially(incomingEvents);
+        await this.#emitIncomingEventsSequentially(incomingEvents);
 
         res.status(200).send({ ok: true });
     }

src/backend/src/modules/broadcast/BroadcastService.redisPubSub.test.js

@@ -49,7 +49,7 @@ describe('BroadcastService redis pubsub', () => {
         eventService.emit.mockClear();
     });
 
-    it('re-emits only outer.gui/pub events from redis pubsub payloads', async () => {
+    it('re-emits only outer.pub events from redis pubsub payloads', async () => {
         await redisClient.publish('broadcast.webhook.events', JSON.stringify({
             sourceId: 'other-instance',
             events: [
@@ -61,15 +61,9 @@ describe('BroadcastService redis pubsub', () => {
 
         await wait();
 
-        expect(eventService.emit).toHaveBeenCalledTimes(2);
+        expect(eventService.emit).toHaveBeenCalledTimes(1);
         expect(eventService.emit).toHaveBeenNthCalledWith(
             1,
-            'outer.gui.notif.message',
-            { id: 'gui-1' },
-            expect.objectContaining({ from_outside: true }),
-        );
-        expect(eventService.emit).toHaveBeenNthCalledWith(
-            2,
             'outer.pub.notice',
             { id: 'pub-1' },
             expect.objectContaining({ from_outside: true }),
@@ -89,10 +83,10 @@ describe('BroadcastService redis pubsub', () => {
         expect(eventService.emit).not.toHaveBeenCalled();
     });
 
-    it('publishes local outer.gui/pub events to redis pubsub for replicas', async () => {
+    it('publishes local outer.pub events to redis pubsub for replicas', async () => {
         const publishSpy = vi.spyOn(redisClient, 'publish');
         try {
-            await service.outBroadcastEventHandler('outer.gui.notif.message', { id: 'gui-local' }, {});
+            await service.outBroadcastEventHandler('outer.pub.notice', { id: 'pub-local' }, {});
             await wait();
 
             const publishCall = publishSpy.mock.calls.find(([channel]) => channel === 'broadcast.webhook.events');
@@ -104,8 +98,8 @@ describe('BroadcastService redis pubsub', () => {
             expect(parsedPayload.sourceId).toBeDefined();
             expect(parsedPayload.events).toEqual([
                 {
-                    key: 'outer.gui.notif.message',
-                    data: { id: 'gui-local' },
+                    key: 'outer.pub.notice',
+                    data: { id: 'pub-local' },
                     meta: {},
                 },
             ]);
@@ -113,4 +107,51 @@ describe('BroadcastService redis pubsub', () => {
             publishSpy.mockRestore();
         }
     });
+
+    it('does not publish local outer.gui events to redis pubsub', async () => {
+        const publishSpy = vi.spyOn(redisClient, 'publish');
+        try {
+            await service.outBroadcastEventHandler('outer.gui.notif.message', { id: 'gui-local' }, {});
+            await wait();
+
+            const publishCall = publishSpy.mock.calls.find(([channel]) => channel === 'broadcast.webhook.events');
+            expect(publishCall).toBeUndefined();
+        } finally {
+            publishSpy.mockRestore();
+        }
+    });
+
+    it('does not rebroadcast events marked from_outside', async () => {
+        const publishSpy = vi.spyOn(redisClient, 'publish');
+        try {
+            await service.outBroadcastEventHandler('outer.gui.notif.message', { id: 'outside' }, {
+                from_outside: true,
+            });
+            await wait();
+
+            expect(publishSpy).not.toHaveBeenCalled();
+        } finally {
+            publishSpy.mockRestore();
+        }
+    });
+
+    it('ignores redis pubsub payloads with this instance sourceId', async () => {
+        const publishSpy = vi.spyOn(redisClient, 'publish');
+        try {
+            await service.outBroadcastEventHandler('outer.pub.notice', { id: 'self-source' }, {});
+            await wait();
+
+            const publishCall = publishSpy.mock.calls.find(([channel]) => channel === 'broadcast.webhook.events');
+            expect(publishCall).toBeDefined();
+            const [_channel, payload] = publishCall;
+
+            eventService.emit.mockClear();
+            await redisClient.publish('broadcast.webhook.events', payload);
+            await wait();
+
+            expect(eventService.emit).not.toHaveBeenCalled();
+        } finally {
+            publishSpy.mockRestore();
+        }
+    });
 });

src/backend/src/services/WSPushService.js

@@ -59,12 +59,10 @@ class WSPushService extends BaseService {
             from_new_service: true,
         };
 
-        {
-            const svc_operationTrace = context.get('services').get('operationTrace');
-            const frame = context.get(svc_operationTrace.ckey('frame'));
-            const gui_metadata = frame.get_attr('gui_metadata') || {};
-            Object.assign(metadata, gui_metadata);
-        }
+        const svc_operationTrace = context.get('services').get('operationTrace');
+        const frame = context.get(svc_operationTrace.ckey('frame'));
+        const gui_metadata = frame.get_attr('gui_metadata') || {};
+        Object.assign(metadata, gui_metadata);
 
         const response = await node.getSafeEntry({ thumbnail: true });
 
@@ -109,12 +107,10 @@ class WSPushService extends BaseService {
             from_new_service: true,
         };
 
-        {
-            const svc_operationTrace = context.get('services').get('operationTrace');
-            const frame = context.get(svc_operationTrace.ckey('frame'));
-            const gui_metadata = frame?.get_attr?.('gui_metadata') || {};
-            Object.assign(metadata, gui_metadata);
-        }
+        const svc_operationTrace = context.get('services').get('operationTrace');
+        const frame = context.get(svc_operationTrace.ckey('frame'));
+        const gui_metadata = frame?.get_attr?.('gui_metadata') || {};
+        Object.assign(metadata, gui_metadata);
 
         const response = await node.getSafeEntry({ debug: 'hi', thumbnail: true });
 
@@ -157,12 +153,10 @@ class WSPushService extends BaseService {
             from_new_service: true,
         };
 
-        {
-            const svc_operationTrace = context.get('services').get('operationTrace');
-            const frame = context.get(svc_operationTrace.ckey('frame'));
-            const gui_metadata = frame.get_attr('gui_metadata') || {};
-            Object.assign(metadata, gui_metadata);
-        }
+        const svc_operationTrace = context.get('services').get('operationTrace');
+        const frame = context.get(svc_operationTrace.ckey('frame'));
+        const gui_metadata = frame.get_attr('gui_metadata') || {};
+        Object.assign(metadata, gui_metadata);
 
         const response = await moved.getSafeEntry();
 
@@ -205,12 +199,10 @@ class WSPushService extends BaseService {
 
         const response = { ...fsentry };
 
-        {
-            const svc_operationTrace = context.get('services').get('operationTrace');
-            const frame = context.get(svc_operationTrace.ckey('frame'));
-            const gui_metadata = frame.get_attr('gui_metadata') || {};
-            Object.assign(metadata, gui_metadata);
-        }
+        const svc_operationTrace = context.get('services').get('operationTrace');
+        const frame = context.get(svc_operationTrace.ckey('frame'));
+        const gui_metadata = frame.get_attr('gui_metadata') || {};
+        Object.assign(metadata, gui_metadata);
 
         const user_id_list = await (async () => {
             const user_id_set = new Set();
@@ -230,7 +222,7 @@ class WSPushService extends BaseService {
     }
 
     /**
-    * Emits an upload or download progress event to the relevant socket.
+    * Emits an upload or download progress event to the relevant user room.
     *
     * @param {string} key - The event key that triggered this method.
     * @param {Object} data - Contains upload_tracker, context, and meta information.
@@ -238,7 +230,7 @@ class WSPushService extends BaseService {
     * @param {Object} data.context - Context of the operation.
     * @param {Object} data.meta - Additional metadata for the event.
     *
-    * It emits a progress event to the socket if it exists, otherwise, it does nothing.
+    * It emits a progress event to the room if it exists, otherwise, it does nothing.
     */
     async _on_upload_progress (key, data) {
         this.log.info('got upload progress event');
@@ -249,17 +241,15 @@ class WSPushService extends BaseService {
             from_new_service: true,
         };
 
-        {
-            const svc_operationTrace = context.get('services').get('operationTrace');
-            const frame = context.get(svc_operationTrace.ckey('frame'));
-            const gui_metadata = frame.get_attr('gui_metadata') || {};
-            Object.assign(metadata, gui_metadata);
-        }
+        const svc_operationTrace = context.get('services').get('operationTrace');
+        const frame = context.get(svc_operationTrace.ckey('frame'));
+        const gui_metadata = frame.get_attr('gui_metadata') || {};
+        Object.assign(metadata, gui_metadata);
 
-        const { socket_id } = metadata;
+        const roomId = metadata.user_id ?? metadata.userId;
 
-        if ( ! socket_id ) {
-            console.warn('missing socket id', { metadata });
+        if ( ! roomId ) {
+            console.warn('missing room id for upload progress', { metadata });
             return;
         }
 
@@ -270,7 +260,7 @@ class WSPushService extends BaseService {
 
         upload_tracker.sub(delta => {
             this.log.info('emitting progress event');
-            svc_socketio.send({ socket: socket_id }, ws_event_name, {
+            svc_socketio.send({ room: roomId }, ws_event_name, {
                 ...metadata,
                 total: upload_tracker.total_,
                 loaded: upload_tracker.progress_,