Bugfix/sync script optimization (#1390)

Author: Uros Marolt

backend/src/database/migrations/U1693124397__searchSyncedAt-index-fix.sql

@@ -0,0 +1,3 @@
+drop index if exists "ix_activities_searchSyncedAt";
+drop index if exists "ix_members_searchSyncedAt";
+create index if not exists "ix_organizations_tenantId_searchSyncedAt" on organizations ("tenantId", "searchSyncedAt");

backend/src/database/migrations/V1693124397__searchSyncedAt-index-fix.sql

@@ -0,0 +1,50 @@
+import { DB_CONFIG, SQS_CONFIG } from '@/conf'
+import IntegrationRunRepository from '@/repo/integrationRun.repo'
+import { singleOrDefault, timeout } from '@crowd/common'
+import { DbStore, getDbConnection } from '@crowd/database'
+import { INTEGRATION_SERVICES } from '@crowd/integrations'
+import { getServiceLogger } from '@crowd/logging'
+import { IntegrationRunWorkerEmitter, getSqsClient } from '@crowd/sqs'
+
+const log = getServiceLogger()
+
+setImmediate(async () => {
+  const sqsClient = getSqsClient(SQS_CONFIG())
+  const emitter = new IntegrationRunWorkerEmitter(sqsClient, log)
+  await emitter.init()
+
+  const dbConnection = getDbConnection(DB_CONFIG(), 1)
+  const store = new DbStore(log, dbConnection)
+
+  const repo = new IntegrationRunRepository(store, log)
+
+  const tenantIds = await repo.getTenantsWithIntegrations()
+
+  for (const tenantId of tenantIds) {
+    const integrations = await repo.getTenantIntegrations(tenantId)
+
+    for (const integration of integrations) {
+      if (singleOrDefault(INTEGRATION_SERVICES, (s) => s.type === integration.type)) {
+        log.info(
+          { integrationId: integration.id },
+          'Integration found in new framework - triggering onboarding!',
+        )
+        await emitter.triggerIntegrationRun(
+          integration.tenantId,
+          integration.type,
+          integration.id,
+          true,
+        )
+
+        log.info('Sleeping for 5 minutes between starts of integration onboarding!')
+        await timeout(5 * 60 * 1000)
+      }
+    }
+
+    log.info('Sleeping for 15 minutes between starts of tenant onboarding!')
+    await timeout(15 * 60 * 1000)
+  }
+
+  log.info('Done!')
+  process.exit(0)
+})

services/apps/integration_run_worker/src/bin/trigger-all-onboardings.ts

@@ -72,6 +72,34 @@ export default class IntegrationRunRepository extends RepositoryBase<Integration
     return result.id
   }
 
+  public async getTenantsWithIntegrations(): Promise<string[]> {
+    const results = await this.db().any(
+      `
+      select distinct "tenantId" from integrations where "deletedAt" is null;
+      `,
+    )
+
+    return results.map((r) => r.tenantId)
+  }
+
+  public async getTenantIntegrations(tenantId: string): Promise<IStartIntegrationRunData[]> {
+    const results = await this.db().any(
+      `
+      select id,
+             platform as type,
+             status as state,
+             "integrationIdentifier" as identifier,
+             "tenantId"
+      from integrations where "tenantId" = $(tenantId) and "deletedAt" is null
+    `,
+      {
+        tenantId,
+      },
+    )
+
+    return results
+  }
+
   public async getIntegrationData(integrationId: string): Promise<IStartIntegrationRunData | null> {
     const results = await this.db().oneOrNone(
       `

services/apps/integration_run_worker/src/repo/integrationRun.repo.ts

@@ -66,11 +66,6 @@ export default class IntegrationRunService extends LoggerBase {
       }
     }
 
-    if (count === 0) {
-      this.log.error('This run has no streams!')
-      return
-    }
-
     if (count === finishedCount) {
       const runInfo = await this.repo.getGenerateStreamData(runId)
 
@@ -208,6 +203,7 @@ export default class IntegrationRunService extends LoggerBase {
           this.log.error({ err }, 'Error while starting integration sync remote!')
         }
 
+        this.log.info('Marking run and integration as successfully processed!')
         await this.repo.markRunProcessed(runId)
         await this.repo.markIntegration(runId, 'done')
 

services/apps/integration_run_worker/src/service/integrationRunService.ts

@@ -147,16 +147,11 @@ export default class IntegrationStreamRepository extends RepositoryBase<Integrat
     this.checkUpdateRowCount(result.rowCount, 1)
   }
 
-  public async markStreamProcessed(streamId: string): Promise<void> {
+  public async deleteStream(streamId: string): Promise<void> {
     const result = await this.db().result(
-      `update integration.streams
-       set  state = $(state),
-            "processedAt" = now(),
-            "updatedAt" = now()
-       where id = $(streamId)`,
+      `delete from integration.streams where id = $(streamId)`,
       {
         streamId,
-        state: IntegrationStreamState.PROCESSED,
       },
     )
 

services/apps/integration_stream_worker/src/repo/integrationStream.repo.ts

@@ -335,14 +335,11 @@ export default class IntegrationStreamService extends LoggerBase {
       },
     }
 
-    this.log.debug('Marking webhook stream as in progress!')
-    await this.repo.markStreamInProgress(streamId)
-
     this.log.debug('Processing webhook stream!')
     try {
       await integrationService.processWebhookStream(context)
       this.log.debug('Finished processing webhook stream!')
-      await this.repo.markStreamProcessed(streamId)
+      await this.repo.deleteStream(streamId)
       await this.webhookRepo.markWebhookProcessed(webhookId)
     } catch (err) {
       this.log.error(err, 'Error while processing webhook stream!')
@@ -486,16 +483,11 @@ export default class IntegrationStreamService extends LoggerBase {
       },
     }
 
-    this.log.debug('Marking stream as in progress!')
-    await this.repo.markStreamInProgress(streamId)
-    // TODO we might need that later to check for stuck runs
-    // await this.repo.touchRun(streamInfo.runId)
-
     this.log.debug('Processing stream!')
     try {
       await integrationService.processStream(context)
       this.log.debug('Finished processing stream!')
-      await this.repo.markStreamProcessed(streamId)
+      await this.repo.deleteStream(streamId)
     } catch (err) {
       this.log.error(err, 'Error while processing stream!')
       await this.triggerStreamError(
@@ -506,8 +498,6 @@ export default class IntegrationStreamService extends LoggerBase {
         err,
       )
     } finally {
-      // TODO we might need that later to check for stuck runs
-      // await this.repo.touchRun(streamInfo.runId)
       await this.runWorkerEmitter.streamProcessed(
         streamInfo.tenantId,
         streamInfo.integrationType,

services/apps/integration_stream_worker/src/service/integrationStreamService.ts

@@ -9,7 +9,6 @@ const log = getServiceLogger()
 
 setImmediate(async () => {
   const openSearchService = new OpenSearchService(log)
-  await openSearchService.initialize()
 
   const dbConnection = getDbConnection(DB_CONFIG())
   const store = new DbStore(log, dbConnection)

services/apps/search_sync_worker/src/bin/cleanup-all-activities.ts

@@ -10,7 +10,6 @@ const log = getServiceLogger()
 
 setImmediate(async () => {
   const openSearchService = new OpenSearchService(log)
-  await openSearchService.initialize()
 
   const redis = await getRedisClient(REDIS_CONFIG(), true)
 

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

@@ -9,7 +9,6 @@ const log = getServiceLogger()
 
 setImmediate(async () => {
   const openSearchService = new OpenSearchService(log)
-  await openSearchService.initialize()
 
   const dbConnection = getDbConnection(DB_CONFIG())
   const store = new DbStore(log, dbConnection)