chore: try to insert identities and merge until we are done

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

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

@@ -1196,7 +1196,6 @@ export default class ActivityService extends LoggerBase {
               reach: value.member.reach,
             },
             value.platform,
-            undefined,
             orgPromiseCache,
             value.timestamp,
           )
@@ -1342,7 +1341,6 @@ export default class ActivityService extends LoggerBase {
                 payload.dbMember,
                 dbMemberIdentities.get(payload.dbMember.id),
                 payload.platform,
-                undefined,
                 orgPromiseCache,
                 payload.activity.timestamp,
               )
@@ -1404,7 +1402,6 @@ export default class ActivityService extends LoggerBase {
                 payload.dbObjectMember,
                 dbMemberIdentities.get(payload.dbObjectMember.id),
                 payload.platform,
-                undefined,
                 orgPromiseCache,
                 payload.activity.timestamp,
               )
@@ -1828,6 +1825,17 @@ export default class ActivityService extends LoggerBase {
       return metadata
     }
 
+    if (error instanceof ApplicationError && error.metadata?.mergeCount !== undefined) {
+      return {
+        ...error.metadata,
+        errorMessage: error.message,
+        memberType,
+        memberIdToUpdate: dbMember?.id,
+        memberSource:
+          memberType === 'member' ? payload.dbMemberSource : payload.dbObjectMemberSource,
+      }
+    }
+
     if (error instanceof ApplicationError) {
       let nextError: any = error.originalError
 

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

@@ -184,12 +184,65 @@ export default class MemberService extends LoggerBase {
     }
   }
 
+  /**
+   * After an initial identity conflict redirects to a surviving member, inserts all incoming
+   * identities that the surviving member doesn't already have. If a further conflict triggers
+   * another merge the loop follows the new survivor. Throws ApplicationError once maxMerges
+   * is reached so the error surfaces in integration.results.
+   */
+  private async syncIdentitiesAfterRedirect(
+    survivingId: string,
+    integrationId: string,
+    incomingIdentities: IMemberIdentity[],
+    maxMerges = 2,
+  ): Promise<string> {
+    for (let mergeCount = 0; ; mergeCount++) {
+      const freshMap = await findIdentitiesForMembers(this.pgQx, [survivingId])
+      const freshIdentities = freshMap.get(survivingId) ?? []
+
+      const toInsert = incomingIdentities.filter(
+        (incoming) =>
+          !freshIdentities.some(
+            (existing) =>
+              existing.platform === incoming.platform &&
+              existing.value === incoming.value &&
+              existing.type === incoming.type &&
+              existing.verified === incoming.verified,
+          ),
+      )
+
+      if (toInsert.length === 0) {
+        return survivingId
+      }
+
+      const nextRedirectId = await this.insertIdentitiesWithConflictResolution(
+        survivingId,
+        integrationId,
+        toInsert,
+        true,
+      )
+
+      if (!nextRedirectId) {
+        return survivingId
+      }
+
+      if (mergeCount >= maxMerges) {
+        throw new ApplicationError('identity sync exceeded merge limit', undefined, {
+          mergeCount: mergeCount + 1,
+          survivingId,
+          maxMerges,
+        })
+      }
+
+      survivingId = nextRedirectId
+    }
+  }
+
   public async create(
     segmentIds: string[],
     integrationId: string,
     data: IMemberCreateData,
     platform: PlatformType,
-    releaseMemberLock?: () => Promise<void>,
     orgPromiseCache?: Map<string, Promise<string | undefined>>,
     activityTimestamp?: string,
   ): Promise<string> {
@@ -295,16 +348,18 @@ export default class MemberService extends LoggerBase {
           }
 
           if (redirectId) {
+            await this.scheduleOrphanMemberDeletion(id)
+            const finalMemberId = await this.syncIdentitiesAfterRedirect(
+              redirectId,
+              integrationId,
+              data.identities,
+            )
             await logExecutionTimeV2(
-              () => this.memberRepo.addToSegments(redirectId, segmentIds),
+              () => this.memberRepo.addToSegments(finalMemberId, segmentIds),
               this.log,
               'memberService -> create -> addToSegments (conflict redirect)',
             )
-            if (releaseMemberLock) {
-              await releaseMemberLock()
-            }
-            await this.scheduleOrphanMemberDeletion(id)
-            return redirectId
+            return finalMemberId
           }
 
           try {
@@ -323,10 +378,6 @@ export default class MemberService extends LoggerBase {
             throw err
           }
 
-          if (releaseMemberLock) {
-            await releaseMemberLock()
-          }
-
           // we should prevent organization creation for bot members
           if (botDetection === MemberBotDetection.CONFIRMED_BOT) {
             this.log.debug('Skipping organization creation for bot member')
@@ -442,7 +493,6 @@ export default class MemberService extends LoggerBase {
     original: IDbMember,
     originalIdentities: IMemberIdentity[],
     platform: PlatformType,
-    releaseMemberLock?: () => Promise<void>,
     orgPromiseCache?: Map<string, Promise<string | undefined>>,
     activityTimestamp?: string,
   ): Promise<string | void> {
@@ -543,7 +593,11 @@ export default class MemberService extends LoggerBase {
               'memberService -> update -> insertIdentitiesWithConflictResolution',
             )
             if (redirectId) {
-              effectiveMemberId = redirectId
+              effectiveMemberId = await this.syncIdentitiesAfterRedirect(
+                redirectId,
+                integrationId,
+                identitiesToCreate,
+              )
             }
           }
 
@@ -563,10 +617,6 @@ export default class MemberService extends LoggerBase {
             }
           }
 
-          if (releaseMemberLock) {
-            await releaseMemberLock()
-          }
-
           if (this.botDetectionService.isFlaggedAsBot(toUpdate.attributes)) {
             this.log.debug({ memberId: id }, 'Skipping organization creation for bot member')
             return effectiveMemberId !== id ? effectiveMemberId : undefined