organization aggs sync improvements (#2860)

Author: skwowet

backend/src/services/member/memberOrganizationsService.ts

@@ -169,6 +169,7 @@ export default class MemberOrganizationsService extends LoggerBase {
         memberId,
         [memberOrganizationToBeDeleted.organizationId],
         repositoryOptions,
+        true,
       )
 
       const result = await this.list(memberId)

services/apps/profiles_worker/src/activities/member/memberUpdate.ts

@@ -21,10 +21,10 @@ export async function updateMemberAffiliations(input: MemberUpdateInput): Promis
   }
 }
 
-export async function syncMember(memberId: string): Promise<void> {
+export async function syncMember(memberId: string, withAggs: boolean): Promise<void> {
   const syncApi = new SearchSyncApiClient({
     baseUrl: process.env['CROWD_SEARCH_SYNC_API_URL'],
   })
 
-  await syncApi.triggerMemberSync(memberId)
+  await syncApi.triggerMemberSync(memberId, { withAggs })
 }

services/apps/profiles_worker/src/workflows/member/memberUpdate.ts

@@ -20,7 +20,7 @@ export async function memberUpdate(input: MemberUpdateInput): Promise<void> {
     await updateMemberAffiliations(input)
     if (input.syncToOpensearch) {
       // sync member
-      await syncMember(input.member.id)
+      await syncMember(input.member.id, input.syncToOpensearch)
       // sync all member organizations
       const organizationIds = input.memberOrganizationIds || []
       for (const orgId of organizationIds) {

services/apps/script_executor_worker/src/activities.ts

@@ -28,12 +28,9 @@ import {
   markActivitiesAsIndexed,
   resetIndexedIdentities,
 } from './activities/populate-activity-relations'
-import {
-  deleteIndexedEntities,
-  getMembersForSync,
-  markEntitiesIndexed,
-  syncMembersBatch,
-} from './activities/sync/member'
+import { deleteIndexedEntities, markEntitiesIndexed } from './activities/sync/entity-index'
+import { getMembersForSync, syncMembersBatch } from './activities/sync/member'
+import { getOrganizationsForSync, syncOrganizationsBatch } from './activities/sync/organization'
 
 export {
   findMembersWithSameVerifiedEmailsInDifferentPlatforms,
@@ -58,6 +55,8 @@ export {
   markActivitiesAsIndexed,
   syncMembersBatch,
   getMembersForSync,
+  getOrganizationsForSync,
+  syncOrganizationsBatch,
   deleteIndexedEntities,
   markEntitiesIndexed,
 }

services/apps/script_executor_worker/src/activities/sync/entity-index.ts

@@ -0,0 +1,27 @@
+import { IndexedEntityType } from '@crowd/opensearch/src/repo/indexing.data'
+import { IndexingRepository } from '@crowd/opensearch/src/repo/indexing.repo'
+
+import { svc } from '../../main'
+
+export async function deleteIndexedEntities(entityType: IndexedEntityType): Promise<void> {
+  try {
+    const indexingRepo = new IndexingRepository(svc.postgres.writer, svc.log)
+    await indexingRepo.deleteIndexedEntities(entityType)
+  } catch (error) {
+    svc.log.error(error, 'Error deleting indexed entities')
+    throw error
+  }
+}
+
+export async function markEntitiesIndexed(
+  entityType: IndexedEntityType,
+  entityIds: string[],
+): Promise<void> {
+  try {
+    const indexingRepo = new IndexingRepository(svc.postgres.writer, svc.log)
+    await indexingRepo.markEntitiesIndexed(entityType, entityIds)
+  } catch (error) {
+    svc.log.error(error, 'Error marking entities indexed')
+    throw error
+  }
+}

services/apps/script_executor_worker/src/activities/sync/member.ts

@@ -1,28 +1,18 @@
 import { sumBy } from '@crowd/common'
 import { DbStore } from '@crowd/data-access-layer/src/database'
 import { MemberSyncService } from '@crowd/opensearch'
-import { IndexedEntityType } from '@crowd/opensearch/src/repo/indexing.data'
-import { IndexingRepository } from '@crowd/opensearch/src/repo/indexing.repo'
 import { MemberRepository } from '@crowd/opensearch/src/repo/member.repo'
 
 import { svc } from '../../main'
 
-export async function deleteIndexedEntities(entityType: IndexedEntityType): Promise<void> {
-  const indexingRepo = new IndexingRepository(svc.postgres.writer, svc.log)
-  await indexingRepo.deleteIndexedEntities(entityType)
-}
-
-export async function markEntitiesIndexed(
-  entityType: IndexedEntityType,
-  entityIds: string[],
-): Promise<void> {
-  const indexingRepo = new IndexingRepository(svc.postgres.writer, svc.log)
-  await indexingRepo.markEntitiesIndexed(entityType, entityIds)
-}
-
 export async function getMembersForSync(batchSize: number): Promise<string[]> {
-  const memberRepo = new MemberRepository(svc.redis, svc.postgres.reader, svc.log)
-  return memberRepo.getMembersForSync(batchSize)
+  try {
+    const memberRepo = new MemberRepository(svc.redis, svc.postgres.reader, svc.log)
+    return memberRepo.getMembersForSync(batchSize)
+  } catch (error) {
+    svc.log.error(error, 'Error getting members for sync')
+    throw error
+  }
 }
 
 export async function syncMembersBatch(
@@ -41,7 +31,7 @@ export async function syncMembersBatch(
 
     const CHUNK_SIZE = chunkSize || 10
 
-    svc.log.info(`Syncing members in chunks of ${CHUNK_SIZE} members!`)
+    svc.log.info(`Syncing members in chunks of ${CHUNK_SIZE}!`)
 
     const results = []
     for (let i = 0; i < memberIds.length; i += CHUNK_SIZE) {

services/apps/script_executor_worker/src/activities/sync/organization.ts

@@ -0,0 +1,51 @@
+import { sumBy } from '@crowd/common'
+import { DbStore } from '@crowd/data-access-layer/src/database'
+import { OrganizationSyncService } from '@crowd/opensearch'
+import { OrganizationRepository } from '@crowd/opensearch/src/repo/organization.repo'
+
+import { svc } from '../../main'
+
+export async function getOrganizationsForSync(batchSize: number): Promise<string[]> {
+  try {
+    const organizationRepo = new OrganizationRepository(svc.postgres.reader, svc.log)
+    return organizationRepo.getOrganizationsForSync(batchSize)
+  } catch (error) {
+    svc.log.error(error, 'Error getting organizations for sync')
+    throw error
+  }
+}
+
+export async function syncOrganizationsBatch(
+  organizationIds: string[],
+  withAggs: boolean,
+  chunkSize?: number,
+): Promise<{ docCount: number; organizationCount: number }> {
+  try {
+    const service = new OrganizationSyncService(
+      new DbStore(svc.log, svc.questdbSQL),
+      svc.postgres.writer,
+      svc.opensearch,
+      svc.log,
+    )
+
+    const CHUNK_SIZE = chunkSize || 10
+
+    svc.log.info(`Syncing orgs in chunks of ${CHUNK_SIZE}!`)
+
+    const results = []
+    for (let i = 0; i < organizationIds.length; i += CHUNK_SIZE) {
+      const chunk = organizationIds.slice(i, i + CHUNK_SIZE)
+      const chunkResults = await Promise.all(
+        chunk.map((organizationId) => service.syncOrganizations([organizationId], { withAggs })),
+      )
+      results.push(...chunkResults)
+    }
+
+    return {
+      docCount: sumBy(results, (r) => r.documentsIndexed),
+      organizationCount: sumBy(results, (r) => r.organizationsSynced),
+    }
+  } catch (err) {
+    throw new Error(err)
+  }
+}

services/apps/script_executor_worker/src/main.ts

@@ -26,6 +26,9 @@ const options: Options = {
   opensearch: {
     enabled: true,
   },
+  queue: {
+    enabled: true,
+  },
 }
 
 export const svc = new ServiceWorker(config, options)

services/apps/script_executor_worker/src/types.ts

@@ -25,10 +25,14 @@ export interface IPopulateActivityRelationsArgs {
   deleteIndexedEntities?: boolean
   latestSyncedActivityTimestamp?: string
 }
-export interface ISyncMembersArgs {
+export interface ISyncArgs {
   batchSize?: number
   chunkSize?: number
   clean?: boolean
   withAggs?: boolean
   testRun?: boolean
 }
+
+export interface IUnlinkDeletedMemberOrgArgs {
+  memberId: string
+}

services/apps/script_executor_worker/src/workflows.ts

@@ -4,6 +4,7 @@ import { findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms } from '.
 import { fixOrgIdentitiesWithWrongUrls } from './workflows/fixOrgIdentitiesWithWrongUrls'
 import { populateActivityRelations } from './workflows/populateActivityRelations'
 import { syncMembers } from './workflows/syncMembers'
+import { syncOrganizations } from './workflows/syncOrganizations'
 
 export {
   findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms,
@@ -12,4 +13,5 @@ export {
   fixOrgIdentitiesWithWrongUrls,
   populateActivityRelations,
   syncMembers,
+  syncOrganizations,
 }