fix: Send updates to correct users

decentraland · Jan 29, 2025 · aeca9c4 · aeca9c4
1 parent 1adb94c
commit aeca9c4
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 6 deletions.
diff --git a/src/adapters/rpc-server/services/subscribe-to-friend-connectivity-updates.ts b/src/adapters/rpc-server/services/subscribe-to-friend-connectivity-updates.ts
@@ -35,6 +35,8 @@ export async function subscribeToFriendConnectivityUpdatesService({
           logger
         },
         getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendConnectivityUpdate']) => update.address,
+        shouldHandleUpdate: (update: SubscriptionEventsEmitter['friendConnectivityUpdate']) =>
+          update.address !== context.address,
         parser: parseEmittedUpdateToFriendConnectivityUpdate,
         parseArgs: [profileImagesUrl]
       })

diff --git a/src/adapters/rpc-server/services/subscribe-to-friendship-updates.ts b/src/adapters/rpc-server/services/subscribe-to-friendship-updates.ts
@@ -18,8 +18,10 @@ export async function subscribeToFriendshipUpdatesService({
         logger,
         catalystClient
       },
-      getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.to,
+      getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from,
       parser: parseEmittedUpdateToFriendshipUpdate,
+      shouldHandleUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) =>
+        update.from !== context.address && update.to === context.address,
       parseArgs: [profileImagesUrl]
     })
   }

diff --git a/src/logic/updates.ts b/src/logic/updates.ts
@@ -20,6 +20,7 @@ interface SubscriptionHandlerParams<T, U> {
     catalystClient: ICatalystClientComponent
   }
   getAddressFromUpdate: (update: U) => string
+  shouldHandleUpdate: (update: U) => boolean
   parser: UpdateParser<T, U>
   parseArgs: any[]
 }
@@ -65,6 +66,7 @@ export async function* handleSubscriptionUpdates<T, U>({
   eventName,
   components: { catalystClient, logger },
   getAddressFromUpdate,
+  shouldHandleUpdate,
   parser,
   parseArgs
 }: SubscriptionHandlerParams<T, U>): AsyncGenerator<T> {
@@ -80,6 +82,11 @@ export async function* handleSubscriptionUpdates<T, U>({
     const eventNameString = String(eventName)
     logger.debug(`${eventNameString} received:`, { update: JSON.stringify(update) })
 
+    if (!shouldHandleUpdate(update as U)) {
+      logger.debug(`Skipping update ${eventNameString} for ${rpcContext.address}`, { update: JSON.stringify(update) })
+      continue
+    }
+
     const profile = await catalystClient.getEntityByPointer(getAddressFromUpdate(update as U))
     const parsedUpdate = await parser(update as U, profile, ...parseArgs)
 

diff --git a/test/unit/logic/updates.spec.ts b/test/unit/logic/updates.spec.ts
@@ -171,7 +171,8 @@ describe('updates handlers', () => {
           catalystClient: mockCatalystClient,
           logger
         },
-        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.to,
+        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from,
+        shouldHandleUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from === '0x123',
         parser,
         parseArgs: [PROFILE_IMAGES_URL]
       })
@@ -202,7 +203,8 @@ describe('updates handlers', () => {
           catalystClient: mockCatalystClient,
           logger
         },
-        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.to,
+        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from,
+        shouldHandleUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from === '0x123',
         parser,
         parseArgs: [PROFILE_IMAGES_URL]
       })
@@ -230,7 +232,8 @@ describe('updates handlers', () => {
           catalystClient: mockCatalystClient,
           logger
         },
-        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.to,
+        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from,
+        shouldHandleUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from === '0x123',
         parser,
         parseArgs: [PROFILE_IMAGES_URL]
       })
@@ -251,7 +254,8 @@ describe('updates handlers', () => {
           catalystClient: mockCatalystClient,
           logger
         },
-        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.to,
+        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from,
+        shouldHandleUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from === '0x123',
         parser,
         parseArgs: [PROFILE_IMAGES_URL]
       })
@@ -272,7 +276,8 @@ describe('updates handlers', () => {
         rpcContext,
         eventName: 'friendshipUpdate',
         components: { catalystClient: mockCatalystClient, logger },
-        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.to,
+        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from,
+        shouldHandleUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from === '0x123',
         parser,
         parseArgs: [PROFILE_IMAGES_URL]
       })
@@ -286,5 +291,28 @@ describe('updates handlers', () => {
         update: JSON.stringify(friendshipUpdate)
       })
     })
+
+    it('should skip update if shouldHandleUpdate returns false', async () => {
+      parser.mockResolvedValueOnce({ parsed: true })
+
+      const generator = handleSubscriptionUpdates({
+        rpcContext,
+        eventName: 'friendshipUpdate',
+        components: { catalystClient: mockCatalystClient, logger },
+        getAddressFromUpdate: (update: SubscriptionEventsEmitter['friendshipUpdate']) => update.from,
+        shouldHandleUpdate: () => false,
+        parser,
+        parseArgs: [PROFILE_IMAGES_URL]
+      })
+
+      const resultPromise = generator.next()
+      rpcContext.subscribers['0x123'].emit('friendshipUpdate', friendshipUpdate)
+
+      await sleep(100) // could be flaky
+
+      expect(logger.debug).toHaveBeenCalledWith('Skipping update friendshipUpdate for 0x123', {
+        update: JSON.stringify(friendshipUpdate)
+      })
+    })
   })
 })