chore: bugfix for two way no merge and better error handling when merging

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

Author: themarolt

services/apps/cron_service/src/jobs/integrationResultsReporting.job.ts

@@ -62,23 +62,34 @@ const job: IJobDefinition = {
       )
     ).count
 
-    // Break down errors by errorMessage + location, enriched with platform info
+    // Break down errors by errorMessage + location, enriched with platform info.
+    // When a mergeError is present in metadata, prefer its errorMessage for grouping
+    // so merge crashes surface as distinct groups rather than collapsing into the
+    // generic outer errorMessage.
     const errorGroups = await dbConnection.any<IErrorGroup>(
       `
       SELECT
-        COALESCE(r.error->>'errorMessage', '[no errorMessage]') AS "errorMessage",
-        COALESCE(r.error->>'location',     '[no location]')     AS location,
-        count(*)::int                                           AS count,
-        round(avg(r.retries), 1)::float                         AS "avgRetries",
-        max(r.retries)::int                                     AS "maxRetries",
-        min(r."createdAt")                                      AS oldest,
-        max(r."updatedAt")                                      AS newest,
+        COALESCE(
+          r.error->'metadata'->>'errorMessage',
+          r.error->>'errorMessage',
+          '[no errorMessage]'
+        ) AS "errorMessage",
+        COALESCE(r.error->>'location', '[no location]') AS location,
+        count(*)::int                                   AS count,
+        round(avg(r.retries), 1)::float                 AS "avgRetries",
+        max(r.retries)::int                             AS "maxRetries",
+        min(r."createdAt")                              AS oldest,
+        max(r."updatedAt")                              AS newest,
         string_agg(DISTINCT i.platform, ', ' ORDER BY i.platform) AS platforms
       FROM integration.results r
       LEFT JOIN integrations i ON i.id = r."integrationId"
       WHERE r.state = 'error'
       GROUP BY
-        r.error->>'errorMessage',
+        COALESCE(
+          r.error->'metadata'->>'errorMessage',
+          r.error->>'errorMessage',
+          '[no errorMessage]'
+        ),
         r.error->>'location'
       ORDER BY count DESC
       LIMIT 20

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

@@ -1768,6 +1768,20 @@ export default class ActivityService extends LoggerBase {
         }
       }
 
+      if (
+        metadata.memberWithIdentity &&
+        metadata.memberIdToUpdate &&
+        metadata.memberWithIdentity === metadata.memberIdToUpdate
+      ) {
+        // The member already owns the conflicting identity — stale prefetch race.
+        // The identity is already present so treat this as a no-op success.
+        this.log.warn(
+          { memberId: metadata.memberIdToUpdate, identity: metadata.erroredVerifiedIdentity },
+          'Verified identity already belongs to this member (stale prefetch) — treating as success',
+        )
+        return metadata.memberIdToUpdate as string
+      }
+
       if (
         metadata.memberWithIdentity &&
         metadata.memberIdToUpdate &&
@@ -1789,16 +1803,22 @@ export default class ActivityService extends LoggerBase {
             return originalId
           } else {
             metadata.noMerge = true
+            metadata.errorMessage = 'noMerge blocked — verified identity conflict'
           }
         } catch (err) {
           metadata.mergeError = {
             errorMessage: err?.message ?? '<no error message>',
             errorStack: err?.stack,
             err,
           }
+          metadata.errorMessage = 'merge failed — auto-merge threw an error'
         }
       }
 
+      if (!metadata.errorMessage) {
+        metadata.errorMessage = 'verified identity conflict — identity owner not found'
+      }
+
       return metadata
     }
 

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

@@ -84,8 +84,7 @@ export async function mergeIfAllowed(
   secondaryId: string,
 ): Promise<boolean> {
   const noMergeMemberIds = await getMemberNoMerge(pgQx, [primaryId, secondaryId])
-  const noMerge = singleOrDefault(
-    noMergeMemberIds,
+  const noMerge = noMergeMemberIds.some(
     (m) =>
       (m.memberId === primaryId && m.noMergeId === secondaryId) ||
       (m.memberId === secondaryId && m.noMergeId === primaryId),