fix: resolve activity members from verified identities instead of stale relations (CM-819) (#3689)

Author: skwowet

services/apps/data_sink_worker/src/service/activity.service.ts

@@ -662,7 +662,6 @@ export default class ActivityService extends LoggerBase {
     )
 
     // map existing activities to payloads for further processing
-    const memberIdsToLoad = new Set<string>()
     const payloadsNotInDb: IActivityProcessData[] = []
     for (const payload of relevantPayloads) {
       const existingRelation = singleOrDefault(existingActivityRelations.rows, (a) => {
@@ -693,73 +692,16 @@ export default class ActivityService extends LoggerBase {
         return aTimestamp === pTimestamp
       })
 
-      // if we have member ids we can use them to load members from db
       if (existingRelation) {
         payload.activityId = existingRelation.activityId
         payload.dbActivityRelation = existingRelation
-
-        memberIdsToLoad.add(existingRelation.memberId)
-
-        if (existingRelation.objectMemberId) {
-          memberIdsToLoad.add(existingRelation.objectMemberId)
-        }
-      } else {
-        payloadsNotInDb.push(payload)
       }
-    }
-
-    if (memberIdsToLoad.size > 0) {
-      // load members by member ids
-      const dbMembers = await logExecutionTimeV2(
-        async () => this.memberRepo.findByIds(Array.from(memberIdsToLoad)),
-        this.log,
-        'processActivities -> memberRepo.findByIds',
-      )
-
-      // and map them to payloads
-      for (const payload of relevantPayloads.filter((p) => p.dbActivityRelation)) {
-        let addToPayloadsNotInDb = false
-        payload.dbMember = singleOrDefault(
-          dbMembers,
-          (m) => m.id === payload.dbActivityRelation.memberId,
-        )
-        if (!payload.dbMember) {
-          this.log.warn(
-            {
-              memberId: payload.dbActivityRelation.memberId,
-            },
-            'Member not found! We will try to find an existing one or create a new one!',
-          )
 
-          addToPayloadsNotInDb = true
-        } else {
-          payload.dbMemberSource = 'activity'
-        }
-
-        if (payload.dbActivityRelation.objectMemberId) {
-          payload.dbObjectMember = singleOrDefault(
-            dbMembers,
-            (m) => m.id === payload.dbActivityRelation.objectMemberId,
-          )
-
-          if (!payload.dbObjectMember) {
-            this.log.warn(
-              {
-                objectMemberId: payload.dbActivityRelation.objectMemberId,
-              },
-              'Object member not found! We will try to find an existing one or create a new one!',
-            )
-
-            addToPayloadsNotInDb = true
-          } else {
-            payload.dbObjectMemberSource = 'activity'
-          }
-        }
-
-        if (addToPayloadsNotInDb) {
-          payloadsNotInDb.push(payload)
-        }
-      }
+      // Regardless of whether the activity already exists, we always resolve the
+      // owning member from identities (username/email/etc.) instead of trusting
+      // the existing relation.memberId. This ensures activities always follow
+      // the current owner of the identity.
+      payloadsNotInDb.push(payload)
     }
 
     if (payloadsNotInDb.length > 0) {

services/libs/data-access-layer/src/member_identities/index.ts

@@ -249,7 +249,7 @@ export async function findMembersByVerifiedEmails(
     with matching_identities as (
       select mi."memberId", mi.value
       from "memberIdentities" mi
-      where mi.type = $(type) and lower(mi.value) in ($(emails:csv))
+      where mi.verified = true and mi.type = $(type) and lower(mi.value) in ($(emails:csv))
       limit ${emails.length}
     )
     select mi.value as "identityValue", ${MEMBER_SELECT_COLUMNS.map((c) => `m."${c}"`).join(', ')}
@@ -295,7 +295,7 @@ export async function findMembersByVerifiedUsernames(
       with matching_identities as (
         select mi."memberId", mi.platform, mi.value
         from "memberIdentities" mi
-        where mi.type = $(type) and (${orConditions.join(' or ')})
+        where mi.verified = true and mi.type = $(type) and (${orConditions.join(' or ')})
         limit ${params.length}
       )
       select mi.platform as "identityPlatform", mi.value as "identityValue", ${MEMBER_SELECT_COLUMNS.map((c) => `m."${c}"`).join(', ')}

services/libs/integrations/src/integrations/github/processWebhookStream.ts

@@ -204,7 +204,7 @@ const parseWebhookPullRequestEvents = async (
       break
     }
     case 'assigned': {
-      objectMember = await handleWebhookSender(payload?.requested_reviewer, ctx)
+      objectMember = await handleWebhookSender(payload?.assignee, ctx)
 
       if (member && objectMember) {
         await ctx.processData<GithubWebhookData>({