fix: keep merge suggestion UI in sync when pairs are rescored

services/apps/members_enrichment_worker/src/activities/enrichment.ts

@@ -59,7 +59,6 @@ import {
   MemberIdentityType,
   OrganizationAttributeSource,
   OrganizationIdentityType,
-  OrganizationMergeSuggestionTable,
   PlatformType,
 } from '@crowd/types'
 
@@ -587,14 +586,7 @@ export async function updateMemberUsingSquashedPayload(
             if (mergeSuggestions.length > 0) {
               // A shared verified identity is a strong merge signal, unless the pair was
               // explicitly marked as no-merge by a reviewer.
-              await mergeSuggestionsRepo.addToMerge(
-                mergeSuggestions,
-                OrganizationMergeSuggestionTable.ORGANIZATION_TO_MERGE_RAW,
-              )
-              await mergeSuggestionsRepo.addToMerge(
-                mergeSuggestions,
-                OrganizationMergeSuggestionTable.ORGANIZATION_TO_MERGE_FILTERED,
-              )
+              await mergeSuggestionsRepo.addToMerge(mergeSuggestions)
             }
           }
         }

services/apps/merge_suggestions_worker/src/activities/memberMergeSuggestions.ts

@@ -13,7 +13,6 @@ import {
   IMemberIdentity,
   IMemberMergeSuggestion,
   MemberIdentityType,
-  MemberMergeSuggestionTable,
   OpenSearchIndex,
   PlatformType,
 } from '@crowd/types'
@@ -379,14 +378,14 @@ export async function getMemberMergeSuggestions(
 
 export async function addMemberToMerge(
   suggestions: IMemberMergeSuggestion[],
-  table: MemberMergeSuggestionTable,
+  similarityThreshold = 0.75,
 ): Promise<void> {
   if (suggestions.length > 0) {
     const memberMergeSuggestionsRepo = new MemberMergeSuggestionsRepository(
       svc.postgres.writer.connection(),
       svc.log,
     )
-    await memberMergeSuggestionsRepo.addToMerge(suggestions, table)
+    await memberMergeSuggestionsRepo.addToMerge(suggestions, similarityThreshold)
   }
 }
 

services/apps/merge_suggestions_worker/src/activities/organizationMergeSuggestions.ts

@@ -20,7 +20,6 @@ import {
   IOrganizationOpensearch,
   OpenSearchIndex,
   OrganizationIdentityType,
-  OrganizationMergeSuggestionTable,
 } from '@crowd/types'
 
 import { svc } from '../main'
@@ -391,14 +390,14 @@ export async function getOrganizationMergeSuggestions(
 
 export async function addOrganizationToMerge(
   suggestions: IOrganizationMergeSuggestion[],
-  table: OrganizationMergeSuggestionTable,
+  similarityThreshold = 0.75,
 ): Promise<void> {
   if (suggestions.length > 0) {
     const organizationMergeSuggestionsRepo = new OrganizationMergeSuggestionsRepository(
       svc.postgres.writer.connection(),
       svc.log,
     )
-    await organizationMergeSuggestionsRepo.addToMerge(suggestions, table)
+    await organizationMergeSuggestionsRepo.addToMerge(suggestions, similarityThreshold)
   }
 }
 

services/apps/merge_suggestions_worker/src/workflows/generateMemberMergeSuggestions.ts

@@ -1,10 +1,6 @@
 import { continueAsNew, proxyActivities } from '@temporalio/workflow'
 
-import {
-  IMemberBaseForMergeSuggestions,
-  IMemberMergeSuggestion,
-  MemberMergeSuggestionTable,
-} from '@crowd/types'
+import { IMemberBaseForMergeSuggestions, IMemberMergeSuggestion } from '@crowd/types'
 
 import * as activities from '../activities/memberMergeSuggestions'
 import { IProcessGenerateMemberMergeSuggestionsArgs } from '../types'
@@ -52,17 +48,8 @@ export async function generateMemberMergeSuggestions(
     allMergeSuggestions.push(...mergeSuggestionsResults.flat())
   }
 
-  // Add all merge suggestions to add to merge
   if (allMergeSuggestions.length > 0) {
-    await activity.addMemberToMerge(
-      allMergeSuggestions,
-      MemberMergeSuggestionTable.MEMBER_TO_MERGE_RAW,
-    )
-
-    await activity.addMemberToMerge(
-      allMergeSuggestions.filter((s) => s.similarity > SIMILARITY_CONFIDENCE_SCORE_THRESHOLD),
-      MemberMergeSuggestionTable.MEMBER_TO_MERGE_FILTERED,
-    )
+    await activity.addMemberToMerge(allMergeSuggestions, SIMILARITY_CONFIDENCE_SCORE_THRESHOLD)
   }
 
   await continueAsNew<typeof generateMemberMergeSuggestions>({ tenantId: args.tenantId, lastUuid })

services/apps/merge_suggestions_worker/src/workflows/generateOrganizationMergeSuggestions.ts

@@ -1,10 +1,6 @@
 import { continueAsNew, proxyActivities } from '@temporalio/workflow'
 
-import {
-  IOrganizationBaseForMergeSuggestions,
-  IOrganizationMergeSuggestion,
-  OrganizationMergeSuggestionTable,
-} from '@crowd/types'
+import { IOrganizationBaseForMergeSuggestions, IOrganizationMergeSuggestion } from '@crowd/types'
 
 import * as activities from '../activities/organizationMergeSuggestions'
 import { IProcessGenerateOrganizationMergeSuggestionsArgs } from '../types'
@@ -55,16 +51,10 @@ export async function generateOrganizationMergeSuggestions(
     allMergeSuggestions.push(...mergeSuggestionsResults.flat())
   }
 
-  // Add all merge suggestions to add to merge
   if (allMergeSuggestions.length > 0) {
     await activity.addOrganizationToMerge(
       allMergeSuggestions,
-      OrganizationMergeSuggestionTable.ORGANIZATION_TO_MERGE_RAW,
-    )
-
-    await activity.addOrganizationToMerge(
-      allMergeSuggestions.filter((s) => s.similarity > SIMILARITY_CONFIDENCE_SCORE_THRESHOLD),
-      OrganizationMergeSuggestionTable.ORGANIZATION_TO_MERGE_FILTERED,
+      SIMILARITY_CONFIDENCE_SCORE_THRESHOLD,
     )
   }
 

services/libs/data-access-layer/src/old/apps/merge_suggestions_worker/memberMergeSuggestions.repo.ts

@@ -82,7 +82,7 @@ class MemberMergeSuggestionsRepository {
 
   async addToMerge(
     suggestions: IMemberMergeSuggestion[],
-    table: MemberMergeSuggestionTable,
+    similarityThreshold = 0.75,
   ): Promise<void> {
     // Remove possible duplicates
     suggestions = removeDuplicateSuggestions<IMemberMergeSuggestion>(
@@ -99,9 +99,9 @@ class MemberMergeSuggestionsRepository {
       }, new Set<string>()),
     )
 
-    // filter non existing member ids from suggestions
     const nonExistingIds = await this.findNonExistingIds(uniqueMemberIds)
 
+    // filter non existing member ids from suggestions
     suggestions = suggestions.filter(
       (s) => !nonExistingIds.includes(s.members[0]) && !nonExistingIds.includes(s.members[1]),
     )
@@ -135,9 +135,66 @@ class MemberMergeSuggestionsRepository {
       }
     }
 
+    const upsertTable = async (
+      table: MemberMergeSuggestionTable,
+      placeholders: string[],
+      replacements: Record<string, unknown>,
+      onlyAboveThreshold: boolean,
+    ) => {
+      const thresholdFilter = onlyAboveThreshold ? 'and v.similarity > $(similarityThreshold)' : ''
+
+      // Update existing rows if they already exist
+      await this.connection.none(
+        `
+            update "${table}" t
+            set
+              similarity = v.similarity,
+              "activityEstimate" = v."activityEstimate",
+              "updatedAt" = now()
+            from (
+              values
+                ${placeholders.join(', ')}
+            ) as v("memberId", "toMergeId", "similarity", "activityEstimate", "createdAt", "updatedAt")
+            where
+              (
+                (t."memberId" = v."memberId"::uuid and t."toMergeId" = v."toMergeId"::uuid)
+                or
+                (t."memberId" = v."toMergeId"::uuid and t."toMergeId" = v."memberId"::uuid)
+              )
+              ${thresholdFilter};
+        `,
+        replacements,
+      )
+
+      // Insert only new rows and enforce bidirectional uniqueness
+      await this.connection.none(
+        `
+            insert into "${table}"
+              ("memberId", "toMergeId", "similarity", "activityEstimate", "createdAt", "updatedAt")
+            select v.*
+            from (
+              values
+                ${placeholders.join(', ')}
+            ) as v("memberId", "toMergeId", "similarity", "activityEstimate", "createdAt", "updatedAt")
+            where not exists (
+              select 1
+              from "${table}" t
+              where
+                (t."memberId" = v."memberId"::uuid and t."toMergeId" = v."toMergeId"::uuid)
+                or
+                (t."memberId" = v."toMergeId"::uuid and t."toMergeId" = v."memberId"::uuid)
+            )
+            ${thresholdFilter};
+        `,
+        replacements,
+      )
+    }
+
     for (const suggestionChunk of suggestionChunks) {
       const placeholders: string[] = []
-      let replacements: Record<string, unknown> = {}
+      let replacements: Record<string, unknown> = {
+        similarityThreshold,
+      }
 
       suggestionChunk.forEach((suggestion, index) => {
         const { query, replacements: chunkReplacements } = insertValues(
@@ -152,45 +209,37 @@ class MemberMergeSuggestionsRepository {
       })
 
       try {
-        // 1. Update existing rows if they already exist
-        await this.connection.none(
-          `
-            update "${table}" t
-            set 
-              similarity = v.similarity,
-              "activityEstimate" = v."activityEstimate",
-              "updatedAt" = now()
-            from (
-              values
-                ${placeholders.join(', ')}
-            ) as v("memberId", "toMergeId", "similarity", "activityEstimate", "createdAt", "updatedAt")
-            where 
-              (t."memberId" = v."memberId"::uuid and t."toMergeId" = v."toMergeId"::uuid)
-              or
-              (t."memberId" = v."toMergeId"::uuid and t."toMergeId" = v."memberId"::uuid);
-        `,
+        // memberToMerge is the high-confidence slice of raw (score > similarityThreshold).
+        // A later run can score the same pair lower, so drop it from UI when it falls to or below the cutoff.
+        await upsertTable(
+          MemberMergeSuggestionTable.MEMBER_TO_MERGE_RAW,
+          placeholders,
           replacements,
+          false,
+        )
+
+        await upsertTable(
+          MemberMergeSuggestionTable.MEMBER_TO_MERGE_FILTERED,
+          placeholders,
+          replacements,
+          true,
         )
 
-        // 2. Insert only new rows and enforce bidirectional uniqueness
         await this.connection.none(
           `
-            insert into "${table}" 
-              ("memberId", "toMergeId", "similarity", "activityEstimate", "createdAt", "updatedAt")
-            select v.*
-            from (
+            delete from "memberToMerge" t
+            using (
               values
                 ${placeholders.join(', ')}
             ) as v("memberId", "toMergeId", "similarity", "activityEstimate", "createdAt", "updatedAt")
-            where not exists (
-              select 1
-              from "${table}" t
-              where 
+            where
+              (
                 (t."memberId" = v."memberId"::uuid and t."toMergeId" = v."toMergeId"::uuid)
                 or
                 (t."memberId" = v."toMergeId"::uuid and t."toMergeId" = v."memberId"::uuid)
-            );
-        `,
+              )
+              and v.similarity <= $(similarityThreshold);
+          `,
           replacements,
         )
       } catch (error) {