member aggs sync improvements

backend/src/database/migrations/V1738589807__rmTenantIdinIndexedEntities.sql

@@ -0,0 +1 @@
+alter table indexed_entities drop column tenant_id;

backend/src/services/searchSyncService.ts

@@ -63,16 +63,6 @@ export default class SearchSyncService extends LoggerBase {
     }
   }
 
-  async triggerTenantMembersSync(tenantId: string) {
-    const client = await this.getSearchSyncClient()
-
-    if (client instanceof SearchSyncApiClient || client instanceof SearchSyncWorkerEmitter) {
-      await client.triggerTenantMembersSync(tenantId)
-    } else {
-      throw new Error('Unexpected search client type!')
-    }
-  }
-
   async triggerOrganizationMembersSync(tenantId: string, organizationId: string) {
     const client = await this.getSearchSyncClient()
 
@@ -126,16 +116,6 @@ export default class SearchSyncService extends LoggerBase {
     }
   }
 
-  async triggerTenantOrganizationSync(tenantId: string) {
-    const client = await this.getSearchSyncClient()
-
-    if (client instanceof SearchSyncApiClient || client instanceof SearchSyncWorkerEmitter) {
-      await client.triggerTenantOrganizationSync(tenantId)
-    } else {
-      throw new Error('Unexpected search client type!')
-    }
-  }
-
   async triggerRemoveOrganization(tenantId: string, organizationId: string) {
     const client = await this.getSearchSyncClient()
 

services/apps/script_executor_worker/package.json

@@ -14,8 +14,10 @@
   "dependencies": {
     "@crowd/archetype-standard": "workspace:*",
     "@crowd/archetype-worker": "workspace:*",
+    "@crowd/common": "workspace:*",
     "@crowd/data-access-layer": "workspace:*",
     "@crowd/logging": "workspace:*",
+    "@crowd/opensearch": "workspace:*",
     "@crowd/redis": "workspace:*",
     "@crowd/types": "workspace:*",
     "@temporalio/workflow": "~1.11.1",

services/apps/script_executor_worker/src/activities.ts

@@ -21,6 +21,12 @@ import {
   findMembersWithSamePlatformIdentitiesDifferentCapitalization,
   findMembersWithSameVerifiedEmailsInDifferentPlatforms,
 } from './activities/merge-members-with-similar-identities'
+import {
+  deleteIndexedEntities,
+  getMembersForSync,
+  markEntitiesIndexed,
+  syncMembersBatch,
+} from './activities/sync/member'
 
 export {
   findMembersWithSameVerifiedEmailsInDifferentPlatforms,
@@ -38,4 +44,8 @@ export {
   updateOrganizationIdentity,
   deleteOrganizationIdentity,
   isLfxMember,
+  syncMembersBatch,
+  getMembersForSync,
+  deleteIndexedEntities,
+  markEntitiesIndexed,
 }

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

@@ -0,0 +1,62 @@
+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)
+}
+
+export async function syncMembersBatch(
+  memberIds: string[],
+  withAggs: boolean,
+  chunkSize?: number,
+): Promise<{ docCount: number; memberCount: number }> {
+  try {
+    const service = new MemberSyncService(
+      svc.redis,
+      svc.postgres.writer,
+      new DbStore(svc.log, svc.questdbSQL),
+      svc.opensearch,
+      svc.log,
+    )
+
+    const CHUNK_SIZE = chunkSize || 10
+
+    svc.log.info(`Syncing members in chunks of ${CHUNK_SIZE} members!`)
+
+    const results = []
+    for (let i = 0; i < memberIds.length; i += CHUNK_SIZE) {
+      const chunk = memberIds.slice(i, i + CHUNK_SIZE)
+      const chunkResults = await Promise.all(
+        chunk.map((memberId) => service.syncMembers(memberId, { withAggs })),
+      )
+      results.push(...chunkResults)
+    }
+
+    return {
+      docCount: sumBy(results, (r) => r.documentsIndexed),
+      memberCount: sumBy(results, (r) => r.membersSynced),
+    }
+  } catch (err) {
+    throw new Error(err)
+  }
+}

services/apps/script_executor_worker/src/main.ts

@@ -10,7 +10,7 @@ const config: Config = {
     enabled: true,
   },
   questdb: {
-    enabled: false,
+    enabled: true,
   },
   redis: {
     enabled: true,
@@ -22,7 +22,7 @@ const options: Options = {
     enabled: true,
   },
   opensearch: {
-    enabled: false,
+    enabled: true,
   },
 }
 

services/apps/script_executor_worker/src/types.ts

@@ -26,3 +26,11 @@ export interface IFixOrgIdentitiesWithWrongUrlsArgs {
   tenantId: string
   testRun?: boolean
 }
+
+export interface ISyncMembersArgs {
+  batchSize?: number
+  chunkSize?: number
+  clean?: boolean
+  withAggs?: boolean
+  testRun?: boolean
+}

services/apps/script_executor_worker/src/workflows.ts

@@ -2,10 +2,12 @@ import { dissectMember } from './workflows/dissectMember'
 import { findAndMergeMembersWithSamePlatformIdentitiesDifferentCapitalization } from './workflows/findAndMergeMembersWithSamePlatformIdentitiesDifferentCapitalization'
 import { findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms } from './workflows/findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms'
 import { fixOrgIdentitiesWithWrongUrls } from './workflows/fixOrgIdentitiesWithWrongUrls'
+import { syncMembers } from './workflows/syncMembers'
 
 export {
   findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms,
   findAndMergeMembersWithSamePlatformIdentitiesDifferentCapitalization,
   dissectMember,
   fixOrgIdentitiesWithWrongUrls,
+  syncMembers,
 }

services/apps/script_executor_worker/src/workflows/syncMembers.ts

@@ -0,0 +1,61 @@
+import { proxyActivities } from '@temporalio/workflow'
+
+import { IndexedEntityType } from '@crowd/opensearch/src/repo/indexing.data'
+
+import * as activities from '../activities/sync/member'
+import { ISyncMembersArgs } from '../types'
+
+const activity = proxyActivities<typeof activities>({
+  startToCloseTimeout: '30 minute',
+  retry: { maximumAttempts: 3, backoffCoefficient: 3 },
+})
+
+export async function syncMembers(args: ISyncMembersArgs): Promise<void> {
+  const BATCH_SIZE = args.batchSize || 100
+  const WITH_AGGS = args.withAggs || true
+
+  console.log(`Starting members sync! (batchSize: ${BATCH_SIZE}, withAggs: ${WITH_AGGS})`)
+
+  if (args.clean) {
+    await activity.deleteIndexedEntities(IndexedEntityType.MEMBER)
+    console.log('Deleted indexed entities for members!')
+  }
+
+  let totalMembersSynced = 0
+  let totalDocumentsIndexed = 0
+
+  let memberIds = await activity.getMembersForSync(BATCH_SIZE)
+
+  while (memberIds.length > 0) {
+    const batchStartTime = new Date()
+    const { docCount, memberCount } = await activity.syncMembersBatch(
+      memberIds,
+      WITH_AGGS,
+      args.chunkSize,
+    )
+
+    totalMembersSynced += memberCount
+    totalDocumentsIndexed += docCount
+
+    const diffInSeconds = (new Date().getTime() - batchStartTime.getTime()) / 1000
+
+    console.log(
+      `Synced ${memberCount} members! Speed: ${Math.round(
+        memberCount / diffInSeconds,
+      )} members/second!`,
+    )
+
+    await activity.markEntitiesIndexed(IndexedEntityType.MEMBER, memberIds)
+
+    if (args.testRun) {
+      console.log('Test run completed - stopping after first batch!')
+      break
+    }
+
+    memberIds = await activity.getMembersForSync(BATCH_SIZE)
+  }
+
+  console.log(
+    `Synced total of ${totalMembersSynced} members with ${totalDocumentsIndexed} documents!`,
+  )
+}

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

@@ -27,23 +27,6 @@ router.post(
   }),
 )
 
-router.post(
-  '/sync/tenant/members',
-  asyncWrap(async (req: ApiRequest, res) => {
-    const memberSyncService = syncService(req)
-
-    const { tenantId } = req.body
-    try {
-      req.log.trace(`Calling memberSyncService.syncTenantMembers for tenant ${tenantId}`)
-      await memberSyncService.syncTenantMembers(tenantId)
-      res.sendStatus(200)
-    } catch (error) {
-      req.log.error(error)
-      res.status(500).send(error.message)
-    }
-  }),
-)
-
 router.post(
   '/sync/organization/members',
   asyncWrap(async (req: ApiRequest, res) => {

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

@@ -28,25 +28,6 @@ router.post(
   }),
 )
 
-router.post(
-  '/sync/tenant/organizations',
-  asyncWrap(async (req: ApiRequest, res) => {
-    const organizationSyncService = syncService(req)
-
-    const { tenantId } = req.body
-    try {
-      req.log.trace(
-        `Calling organizationSyncService.syncTenantOrganizations for tenant ${tenantId}`,
-      )
-      await organizationSyncService.syncTenantOrganizations(tenantId)
-      res.sendStatus(200)
-    } catch (error) {
-      req.log.error(error)
-      res.status(500).send(error.message)
-    }
-  }),
-)
-
 router.post(
   '/cleanup/tenant/organizations',
   asyncWrap(async (req: ApiRequest, res) => {

services/apps/search_sync_worker/src/bin/sync-all-members.ts

@@ -1,6 +1,4 @@
-import { timeout } from '@crowd/common'
 import { DbStore, getDbConnection } from '@crowd/data-access-layer/src/database'
-import { MemberRepository } from '@crowd/data-access-layer/src/old/apps/search_sync_worker/member.repo'
 import { getServiceLogger } from '@crowd/logging'
 import { MemberSyncService, OpenSearchService, getOpensearchClient } from '@crowd/opensearch'
 import { IndexedEntityType } from '@crowd/opensearch/src/repo/indexing.data'
@@ -14,8 +12,6 @@ const log = getServiceLogger()
 
 const processArguments = process.argv.slice(2)
 
-const MAX_CONCURRENT = 3
-
 setImmediate(async () => {
   const osClient = await getOpensearchClient(OPENSEARCH_CONFIG())
   const openSearchService = new OpenSearchService(log, osClient)
@@ -35,42 +31,18 @@ setImmediate(async () => {
     withAggs = false
   }
 
-  const repo = new MemberRepository(store, log)
-
-  const tenantIds = await repo.getTenantIds()
   const qdbConn = await getClientSQL()
   const qdbStore = new DbStore(log, qdbConn)
 
   const service = new MemberSyncService(redis, store, qdbStore, openSearchService, log)
 
-  let current = 0
-  for (let i = 0; i < tenantIds.length; i++) {
-    const tenantId = tenantIds[i]
-
-    while (current == MAX_CONCURRENT) {
-      await timeout(1000)
-    }
-
-    log.info(`Processing tenant ${i + 1}/${tenantIds.length}`)
-    current += 1
-    service
-      .syncTenantMembers(tenantId, 500, { withAggs })
-      .then(() => {
-        current--
-        log.info(`Processed tenant ${i + 1}/${tenantIds.length}`)
-      })
-      .catch((err) => {
-        current--
-        log.error(
-          err,
-          { tenantId },
-          `Error syncing members for tenant ${i + 1}/${tenantIds.length}!`,
-        )
-      })
-  }
+  log.info('Starting sync of all members')
 
-  while (current > 0) {
-    await timeout(1000)
+  try {
+    await service.syncAllMembers(500, { withAggs })
+    log.info('Successfully synced all members')
+  } catch (err) {
+    log.error(err, 'Error syncing members!')
   }
 
   process.exit(0)