search sync worker syncOrganizationMembers queue messages

Author: epipav

backend/src/database/repositories/organizationRepository.ts

@@ -808,7 +808,7 @@ class OrganizationRepository {
     fromOrganizationId: string,
     toOrganizationId: string,
     options: IRepositoryOptions,
-  ): Promise<string[]> {
+  ): Promise<void> {
     const seq = SequelizeRepository.getSequelize(options)
 
     const transaction = SequelizeRepository.getTransaction(options)
@@ -817,8 +817,6 @@ class OrganizationRepository {
 
     let addRoles: IMemberOrganization[] = []
 
-    const updatedMembers = new Set<string>()
-
     // first, handle members that belong to both organizations,
     // then make a full update on remaining org2 members (that doesn't belong to o1)
     const memberRolesWithBothOrganizations = await this.findMembersBelongToBothOrganizations(
@@ -917,50 +915,38 @@ class OrganizationRepository {
 
       for (const removeRole of removeRoles) {
         await this.removeMemberRole(removeRole, options)
-        updatedMembers.add(removeRole.memberId)
       }
 
       for (const addRole of addRoles) {
         await this.addMemberRole(addRole, options)
-        updatedMembers.add(addRole.memberId)
       }
 
       addRoles = []
       removeRoles = []
     }
 
     // update rest of the o2 members
-    const results = await seq.query(
+    await seq.query(
       `
-      WITH updated AS (
-        UPDATE "memberOrganizations"
+      UPDATE "memberOrganizations"
         SET "organizationId" = :toOrganizationId
         WHERE "organizationId" = :fromOrganizationId 
         AND "deletedAt" IS NULL
         AND "memberId" NOT IN (
             SELECT "memberId" 
             FROM "memberOrganizations" 
             WHERE "organizationId" = :toOrganizationId
-        )
-        RETURNING "memberId"
-      )
-      SELECT "memberId" FROM updated;
+        );
       `,
       {
         replacements: {
           toOrganizationId,
           fromOrganizationId,
         },
-        type: QueryTypes.SELECT,
+        type: QueryTypes.UPDATE,
         transaction,
       },
     )
-
-    for (const result of results) {
-      updatedMembers.add((result as any).memberId)
-    }
-
-    return Array.from(updatedMembers)
   }
 
   static async getOrganizationSegments(

backend/src/services/organizationService.ts

@@ -104,7 +104,7 @@ export default class OrganizationService extends LoggerBase {
       await txService.update(originalId, toUpdate, repoOptions.transaction)
 
       // update members that belong to source organization to destination org
-      const updatedMembers = await OrganizationRepository.moveMembersBetweenOrganizations(
+      await OrganizationRepository.moveMembersBetweenOrganizations(
         toMergeId,
         originalId,
         repoOptions,
@@ -136,10 +136,8 @@ export default class OrganizationService extends LoggerBase {
       await searchSyncEmitter.triggerOrganizationSync(this.options.currentTenant.id, originalId)
       await searchSyncEmitter.triggerRemoveOrganization(this.options.currentTenant.id, toMergeId)
 
-      while (updatedMembers.length > 0) {
-        const updatedMember = updatedMembers.shift()
-        await searchSyncEmitter.triggerMemberSync(this.options.currentTenant.id, updatedMember)
-      }
+      // sync organization members
+      await searchSyncEmitter.triggerOrganizationMembersSync(originalId)
 
       // sync organization activities
       await searchSyncEmitter.triggerOrganizationActivitiesSync(originalId)

services/apps/search_sync_worker/src/queue/index.ts

@@ -127,6 +127,14 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
               .catch((err) => this.log.error(err, 'Error while syncing tenant members!'))
           }
 
+          break
+        case SearchSyncWorkerQueueMessageType.SYNC_ORGANIZATION_MEMBERS:
+          if (data.organizationId) {
+            this.initMemberService()
+              .syncOrganizationMembers(data.organizationId)
+              .catch((err) => this.log.error(err, 'Error while syncing organization members!'))
+          }
+
           break
         // this one taks a while so we can't relly on it to be finished in time and the queue message might pop up again so we immediatelly return
         case SearchSyncWorkerQueueMessageType.CLEANUP_TENANT_MEMBERS:

services/apps/search_sync_worker/src/repo/member.repo.ts

@@ -89,6 +89,57 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
     return results.map((r) => r.id)
   }
 
+  public async getOrganizationMembersForSync(
+    organizationId: string,
+    perPage: number,
+    lastId?: string,
+  ): Promise<string[]> {
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    let results: any[]
+
+    if (lastId) {
+      results = await this.db().any(
+        `
+        select distinct mo."memberId"
+        from "memberOrganizations" mo
+        where mo."organizationId" = $(organizationId) and
+              mo."deletedAt" is null and
+              mo."memberId" > $(lastId) and
+              (
+                exists (select 1 from activities where "memberId" = mo."memberId") or
+                exists (select 1 from members where "memberId" = mo."memberId" and "manuallyCreated")
+              ) and
+              exists (select 1 from "memberIdentities" where "memberId" = mo."memberId")
+        order by mo."memberId"
+        limit ${perPage};`,
+        {
+          organizationId,
+          lastId,
+        },
+      )
+    } else {
+      results = await this.db().any(
+        `
+        select distinct mo."memberId"
+        from "memberOrganizations" mo
+        where mo."organizationId" = $(organizationId) and
+              mo."deletedAt" is null and
+              (
+                exists (select 1 from activities where "memberId" = mo."memberId") or
+                exists (select 1 from members where "memberId" = mo."memberId" and "manuallyCreated")
+              ) and
+              exists (select 1 from "memberIdentities" where "memberId" = mo."memberId")
+        order by mo."memberId"
+        limit ${perPage};`,
+        {
+          organizationId,
+        },
+      )
+    }
+
+    return results.map((r) => r.memberId)
+  }
+
   public async getRemainingTenantMembersForSync(
     tenantId: string,
     page: number,

services/apps/search_sync_worker/src/service/member.sync.service.ts

@@ -265,6 +265,50 @@ export class MemberSyncService extends LoggerBase {
     )
   }
 
+  public async syncOrganizationMembers(organizationId: string, batchSize = 200): Promise<void> {
+    this.log.debug({ organizationId }, 'Syncing all organization members!')
+    let docCount = 0
+    let memberCount = 0
+
+    const now = new Date()
+
+    await logExecutionTime(
+      async () => {
+        let memberIds = await this.memberRepo.getOrganizationMembersForSync(
+          organizationId,
+          batchSize,
+        )
+
+        while (memberIds.length > 0) {
+          const { membersSynced, documentsIndexed } = await this.syncMembers(memberIds)
+
+          docCount += documentsIndexed
+          memberCount += membersSynced
+
+          const diffInSeconds = (new Date().getTime() - now.getTime()) / 1000
+          this.log.info(
+            { organizationId },
+            `Synced ${memberCount} members! Speed: ${Math.round(
+              memberCount / diffInSeconds,
+            )} members/second!`,
+          )
+          memberIds = await this.memberRepo.getOrganizationMembersForSync(
+            organizationId,
+            batchSize,
+            memberIds[memberIds.length - 1],
+          )
+        }
+      },
+      this.log,
+      'sync-organization-members',
+    )
+
+    this.log.info(
+      { organizationId },
+      `Synced total of ${memberCount} members with ${docCount} documents!`,
+    )
+  }
+
   public async syncMembers(memberIds: string[]): Promise<IMemberSyncResult> {
     this.log.debug({ memberIds }, 'Syncing members!')
 

services/libs/sqs/src/instances/searchSyncWorker.ts

@@ -35,6 +35,16 @@ export class SearchSyncWorkerEmitter extends SqsQueueEmitter {
     })
   }
 
+  public async triggerOrganizationMembersSync(organizationId: string) {
+    if (!organizationId) {
+      throw new Error('organizationId is required!')
+    }
+    await this.sendMessage(organizationId, {
+      type: SearchSyncWorkerQueueMessageType.SYNC_ORGANIZATION_MEMBERS,
+      organizationId,
+    })
+  }
+
   public async triggerRemoveMember(tenantId: string, memberId: string) {
     if (!tenantId) {
       throw new Error('tenantId is required!')

services/libs/types/src/queue/search_sync_worker/index.ts

@@ -1,6 +1,7 @@
 export enum SearchSyncWorkerQueueMessageType {
   SYNC_MEMBER = 'sync_member',
   SYNC_TENANT_MEMBERS = 'sync_tenant_members',
+  SYNC_ORGANIZATION_MEMBERS = 'sync_organization_members',
   REMOVE_MEMBER = 'remove_member',
   CLEANUP_TENANT_MEMBERS = 'cleanup_tenant_members',