opensearch sync issues fixed when syncing unmerged members with no activities

Author: epipav

services/apps/entity_merging_worker/src/activities/members.ts

@@ -10,6 +10,8 @@ import {
   findMemberById,
   moveActivitiesToNewMember,
   moveIdentityActivitiesToNewMember,
+  findMemberSegments,
+  markMemberAsManuallyCreated,
 } from '@crowd/data-access-layer/src/old/apps/entity_merging_worker'
 
 export async function deleteMember(memberId: string): Promise<void> {
@@ -78,12 +80,29 @@ export async function recalculateActivityAffiliations(
   })
 }
 
-export async function syncMember(memberId: string): Promise<void> {
+export async function syncMember(memberId: string, secondaryMemberId: string): Promise<void> {
   const syncApi = new SearchSyncApiClient({
     baseUrl: process.env['CROWD_SEARCH_SYNC_API_URL'],
   })
 
-  await syncApi.triggerMemberSync(memberId)
+  // check if member has any activities
+  const result = await findMemberSegments(svc.postgres.writer, memberId)
+
+  if (result.segmentIds) {
+    // segment information can be deduced from activities, no need to send segmentIds explicitly on merging
+    await syncApi.triggerMemberSync(memberId)
+    return
+  }
+
+  // check if secondary member has any activities
+  const secondaryResult = await findMemberSegments(svc.postgres.writer, secondaryMemberId)
+
+  if (secondaryResult.segmentIds) {
+    // mark member as manually created
+    await markMemberAsManuallyCreated(svc.postgres.writer, memberId)
+    // member doesn't have any activity to deduce segmentIds for syncing, use the secondary member's activity segments
+    await syncApi.triggerMemberSync(memberId, secondaryResult.segmentIds)
+  }
 }
 
 export async function notifyFrontendMemberUnmergeSuccessful(

services/apps/entity_merging_worker/src/workflows/all.ts

@@ -39,8 +39,8 @@ export async function finishMemberUnmerging(
   userId: string,
 ): Promise<void> {
   await moveActivitiesWithIdentityToAnotherMember(primaryId, secondaryId, identities, tenantId)
-  await syncMember(primaryId)
-  await syncMember(secondaryId)
+  await syncMember(primaryId, secondaryId)
+  await syncMember(secondaryId, primaryId)
   await recalculateActivityAffiliations(primaryId, tenantId)
   await recalculateActivityAffiliations(secondaryId, tenantId)
   await setMergeActionState(primaryId, secondaryId, tenantId, 'unmerged' as MergeActionState)

services/apps/search_sync_api/src/routes/member.ts

@@ -18,10 +18,10 @@ router.post(
       serviceConfig,
     )
 
-    const { memberIds } = req.body
+    const { memberIds, segmentIds } = req.body
     try {
       req.log.trace(`Calling memberSyncService.syncMembers for ${memberIds}`)
-      await memberSyncService.syncMembers(memberIds)
+      await memberSyncService.syncMembers(memberIds, segmentIds)
       res.sendStatus(200)
     } catch (error) {
       res.status(500).send(error.message)

services/libs/data-access-layer/src/old/apps/entity_merging_worker/index.ts

@@ -1,5 +1,6 @@
 import { DbStore } from '@crowd/database'
 import { MergeActionState } from '@crowd/types'
+import { ISegmentIds } from './types'
 
 export async function deleteMemberSegments(db: DbStore, memberId: string) {
   return db.connection().query(
@@ -90,3 +91,25 @@ export async function moveIdentityActivitiesToNewMember(
     [toId, fromId, tenantId, username, platform],
   )
 }
+
+export async function findMemberSegments(db: DbStore, memberId: string): Promise<ISegmentIds> {
+  const result = await db.connection().one(
+    `
+      SELECT array_agg(distinct "segmentId") as "segmentIds"
+      FROM activities
+      WHERE "memberId" = $1
+    `,
+    [memberId],
+  )
+  return result as ISegmentIds
+}
+
+export async function markMemberAsManuallyCreated(db: DbStore, memberId: string): Promise<void> {
+  return db.connection().query(
+    `
+      UPDATE members set "manuallyCreated" = true
+      WHERE "id" = $1
+    `,
+    [memberId],
+  )
+}

services/libs/data-access-layer/src/old/apps/entity_merging_worker/types.ts

@@ -0,0 +1,3 @@
+export interface ISegmentIds {
+  segmentIds: string[]
+}

services/libs/opensearch/src/apiClient.ts

@@ -12,13 +12,14 @@ export class SearchSyncApiClient {
     })
   }
 
-  public async triggerMemberSync(memberId: string): Promise<void> {
+  public async triggerMemberSync(memberId: string, segmentIds?: string[]): Promise<void> {
     if (!memberId) {
       throw new Error('memberId is required!')
     }
 
     await this.searchSyncApi.post('/sync/members', {
       memberIds: [memberId],
+      segmentIds,
     })
   }
 

services/libs/opensearch/src/repo/member.data.ts

@@ -102,3 +102,8 @@ export interface IMemberSegmentMatrixItem {
 export interface IMemberSegmentMatrix {
   [key: string]: IMemberSegmentMatrixItem[]
 }
+
+export interface IMemberSegment {
+  memberId: string
+  segmentId: string
+}

services/libs/opensearch/src/repo/member.repo.ts

@@ -2,7 +2,7 @@ import { DbStore, RepositoryBase } from '@crowd/database'
 import { Logger } from '@crowd/logging'
 import { RedisCache, RedisClient } from '@crowd/redis'
 import { IMemberAttribute } from '@crowd/types'
-import { IDbMemberSyncData, IMemberSegmentMatrix } from './member.data'
+import { IDbMemberSyncData, IMemberSegment, IMemberSegmentMatrix } from './member.data'
 
 export class MemberRepository extends RepositoryBase<MemberRepository> {
   private readonly cache: RedisCache
@@ -586,40 +586,58 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
     return results.map((r) => r.id)
   }
 
-  public async getMemberSegmentCouples(ids): Promise<IMemberSegmentMatrix> {
-    let results = await this.db().any(
-      `
-      select distinct a."segmentId", a."memberId"
-      from activities a
-      where a."memberId" in ($(ids:csv));
-      `,
-      {
-        ids,
-      },
-    )
-
-    // Manually created members don't have any activities,
-    // filter out those memberIds that are not in the results
-    const manuallyCreatedIds = ids.filter((id) => !results.some((r) => r.memberId === id))
-
-    // memberSegments aren't maintained well, so use as a fallback for manuallyCreated members
-    if (manuallyCreatedIds.length > 0) {
-      const missingResults = await this.db().any(
+  public async getMemberSegmentCouples(
+    memberIds: string[],
+    segmentIds?: string[],
+  ): Promise<IMemberSegmentMatrix> {
+    let memberSegments: IMemberSegment[] = []
+
+    if (segmentIds && segmentIds.length > 0) {
+      for (const memberId of memberIds) {
+        for (const segmentId of segmentIds) {
+          memberSegments.push({
+            memberId,
+            segmentId,
+          })
+        }
+      }
+    } else {
+      memberSegments = await this.db().any(
         `
-        select distinct ms."segmentId", ms."memberId"
-        from "memberSegments" ms
-        where ms."memberId" in ($(manuallyCreatedIds:csv));
+        select distinct a."segmentId", a."memberId"
+        from activities a
+        where a."memberId" in ($(ids:csv));
         `,
         {
-          manuallyCreatedIds,
+          ids: memberIds,
         },
       )
-      results = [...results, ...missingResults]
+
+      // Manually created members don't have any activities,
+      // filter out those memberIds that are not in the results
+      const manuallyCreatedIds = memberIds.filter(
+        (id) => !memberSegments.some((r) => r.memberId === id),
+      )
+
+      // memberSegments aren't maintained well, so use as a fallback for manuallyCreated members
+      if (manuallyCreatedIds.length > 0) {
+        const missingResults = await this.db().any(
+          `
+          select distinct ms."segmentId", ms."memberId"
+          from "memberSegments" ms
+          where ms."memberId" in ($(manuallyCreatedIds:csv));
+          `,
+          {
+            manuallyCreatedIds,
+          },
+        )
+        memberSegments = [...memberSegments, ...missingResults]
+      }
     }
 
     const matrix = {}
 
-    for (const memberSegment of results) {
+    for (const memberSegment of memberSegments) {
       if (!matrix[memberSegment.memberId]) {
         matrix[memberSegment.memberId] = [
           {

services/libs/opensearch/src/service/member.sync.service.ts

@@ -312,13 +312,13 @@ export class MemberSyncService {
     )
   }
 
-  public async syncMembers(memberIds: string[]): Promise<IMemberSyncResult> {
+  public async syncMembers(memberIds: string[], segmentIds?: string[]): Promise<IMemberSyncResult> {
     const CONCURRENT_DATABASE_QUERIES = 25
     const BULK_INDEX_DOCUMENT_BATCH_SIZE = 2500
 
     // get all memberId-segmentId couples
     const memberSegmentCouples: IMemberSegmentMatrix =
-      await this.memberRepo.getMemberSegmentCouples(memberIds)
+      await this.memberRepo.getMemberSegmentCouples(memberIds, segmentIds)
     let databaseStream = []
     let syncStream = []
     let documentsIndexed = 0