Optimize data workers

services/apps/data_sink_worker/src/jobs/processOldResults.ts

@@ -7,6 +7,8 @@ import { RedisClient, processWithLock } from '@crowd/redis'
 import { NodejsWorkerEmitter, SearchSyncWorkerEmitter } from '@crowd/sqs'
 import { Client as TemporalClient } from '@crowd/temporal'
 
+const MAX_CONCURRENT_PROMISES = 3
+
 export const processOldResultsJob = async (
   dbConn: DbConnection,
   redis: RedisClient,
@@ -30,13 +32,13 @@ export const processOldResultsJob = async (
 
   const loadNextBatch = async (): Promise<string[]> => {
     return await processWithLock(redis, 'process-old-results', 5 * 60, 3 * 60, async () => {
-      const resultIds = await repo.getOldResultsToProcess(5)
+      const resultIds = await repo.getOldResultsToProcess(MAX_CONCURRENT_PROMISES)
       await repo.touchUpdatedAt(resultIds)
       return resultIds
     })
   }
 
-  // load 5 oldest results and try process them
+  // load 3 oldest results and try process them
   let resultsToProcess = await loadNextBatch()
 
   let successCount = 0
@@ -45,20 +47,34 @@ export const processOldResultsJob = async (
   while (resultsToProcess.length > 0) {
     log.info(`Detected ${resultsToProcess.length} old results rows to process!`)
 
+    const promises = []
     for (const resultId of resultsToProcess) {
-      try {
-        const result = await service.processResult(resultId)
-        if (result) {
-          successCount++
-        } else {
-          errorCount++
-        }
-      } catch (err) {
-        log.error(err, 'Failed to process result!')
-        errorCount++
+      promises.push(
+        service
+          .processResult(resultId)
+          .then((result) => {
+            if (result) {
+              successCount++
+            } else {
+              errorCount++
+            }
+          })
+          .catch((err) => {
+            log.error(err, 'Failed to process result!')
+            errorCount++
+          }),
+      )
+
+      if (promises.length >= MAX_CONCURRENT_PROMISES) {
+        await Promise.all(promises)
+        promises.length = 0
       }
     }
 
+    if (promises.length > 0) {
+      await Promise.all(promises)
+    }
+
     log.info(`Processed ${successCount} old results successfully and ${errorCount} with errors.`)
 
     resultsToProcess = await loadNextBatch()

services/apps/data_sink_worker/src/main.ts

@@ -20,7 +20,7 @@ import { Client as TemporalClient, getTemporalClient } from '@crowd/temporal'
 const tracer = getServiceTracer()
 const log = getServiceLogger()
 
-const MAX_CONCURRENT_PROCESSING = 3
+const MAX_CONCURRENT_PROCESSING = 4
 const PROCESSING_INTERVAL_MINUTES = 5
 
 setImmediate(async () => {

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

@@ -41,7 +41,7 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
         this.log.trace({ messageType: message.type }, 'Processing message!')
 
         const service = new DataSinkService(
-          new DbStore(this.log, this.dbConn),
+          new DbStore(this.log, this.dbConn, undefined, false),
           this.nodejsWorkerEmitter,
           this.searchSyncWorkerEmitter,
           this.redisClient,

services/apps/data_sink_worker/src/repo/dataSink.repo.ts

@@ -59,13 +59,15 @@ export default class DataSinkRepository extends RepositoryBase<DataSinkRepositor
     try {
       const results = await this.db().any(
         `
-        select id
-        from integration.results
-        where state in ($(pendingState), $(processingState))
-          and "updatedAt" < now() - interval '1 hour'
-        order by case when "webhookId" is not null then 0 else 1 end,
-                "webhookId" asc,
-                "updatedAt" desc
+        select r.id
+        from integration.results r
+        inner join tenants t on t.id = r."tenantId"
+        where r.state in ($(pendingState), $(processingState))
+          and r."updatedAt" < now() - interval '1 hour'
+        order by case when t."plan" in ('Scale', 'Growth') then 0 else 1 end,
+                case when r."webhookId" is not null then 0 else 1 end,
+                r."webhookId" asc,
+                r."updatedAt" desc
         limit ${limit};
         `,
         {

services/apps/data_sink_worker/src/repo/organization.repo.ts

@@ -459,6 +459,7 @@ export class OrganizationRepository extends RepositoryBase<OrganizationRepositor
       },
     })
     const updatedAt = new Date()
+    const oneMinuteAgo = new Date(updatedAt.getTime() - 60 * 1000)
     const prepared = RepositoryBase.prepare(
       {
         ...data,
@@ -472,9 +473,9 @@ export class OrganizationRepository extends RepositoryBase<OrganizationRepositor
     )
 
     const query = this.dbInstance.helpers.update(prepared, dynamicColumnSet)
-    const condition = this.format('where id = $(id) and "updatedAt" < $(updatedAt)', {
+    const condition = this.format('where id = $(id) and "updatedAt" <= $(oneMinuteAgo)', {
       id,
-      updatedAt,
+      oneMinuteAgo,
     })
 
     await this.db().result(`${query} ${condition}`)

services/apps/data_sink_worker/src/service/activity.service.ts

@@ -17,14 +17,32 @@ import IntegrationRepository from '../repo/integration.repo'
 import GithubReposRepository from '../repo/githubRepos.repo'
 import MemberAffiliationService from './memberAffiliation.service'
 import { RedisClient } from '@crowd/redis'
-import { acquireLock, releaseLock } from '@crowd/redis'
+// import { acquireLock, releaseLock } from '@crowd/redis'
 import { Unleash, isFeatureEnabled } from '@crowd/feature-flags'
 import { Client as TemporalClient, WorkflowIdReusePolicy } from '@crowd/temporal'
 import { TEMPORAL_CONFIG } from '../conf'
 
 const MEMBER_LOCK_EXPIRE_AFTER = 10 * 60 // 10 minutes
 const MEMBER_LOCK_TIMEOUT_AFTER = 5 * 60 // 5 minutes
 
+const acquireLock = async (
+  redisClient: RedisClient,
+  lockKey: string,
+  lockReason: string,
+  lockExpireAfter: number,
+  lockTimeoutAfter: number,
+): Promise<void> => {
+  // empty
+}
+
+const releaseLock = async (
+  redisClient: RedisClient,
+  lockKey: string,
+  lockReason: string,
+): Promise<void> => {
+  // empty
+}
+
 export default class ActivityService extends LoggerBase {
   private readonly conversationService: ConversationService
 

services/apps/integration_data_worker/src/main.ts

@@ -10,7 +10,7 @@ import { processOldDataJob } from './jobs/processOldData'
 const tracer = getServiceTracer()
 const log = getServiceLogger()
 
-const MAX_CONCURRENT_PROCESSING = 2
+const MAX_CONCURRENT_PROCESSING = 3
 const PROCESSING_INTERVAL_MINUTES = 5
 
 setImmediate(async () => {

services/libs/database/src/dbStore.ts

@@ -16,6 +16,7 @@ export class DbStore extends LoggerBase {
     parentLog?: Logger,
     private readonly dbConnection?: DbConnection,
     private readonly dbTransaction?: DbTransaction,
+    private readonly withTransactions: boolean = true,
   ) {
     super(parentLog, { transactional: dbTransaction !== undefined })
 
@@ -51,6 +52,10 @@ export class DbStore extends LoggerBase {
   public transactionally<T>(inTransaction: (store: DbStore) => Promise<T>): Promise<T> {
     this.checkValid()
 
+    if (!this.withTransactions) {
+      return inTransaction(this)
+    }
+
     if (this.isTransaction()) {
       this.log.debug('Using an existing transaction!')
       return inTransaction(this)