fix: small ones

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/cron_service/src/jobs/incomingWebhooksCheck.job.ts

@@ -23,7 +23,12 @@ const job: IJobDefinition = {
     const admin = kafkaClient.admin()
     await admin.connect()
 
-    const counts = await getKafkaMessageCounts(ctx.log, admin, TOPIC, GROUP_ID)
+    let counts: { total: number; consumed: number; unconsumed: number }
+    try {
+      counts = await getKafkaMessageCounts(ctx.log, admin, TOPIC, GROUP_ID)
+    } finally {
+      await admin.disconnect()
+    }
 
     if (counts.unconsumed >= MAX_UNCONSUMED) {
       ctx.log.info(

services/apps/cron_service/src/jobs/integrationResultsCheck.job.ts

@@ -22,59 +22,63 @@ const job: IJobDefinition = {
     const admin = kafkaClient.admin()
     await admin.connect()
 
-    const counts = await getKafkaMessageCounts(ctx.log, admin, topic, groupId)
+    try {
+      const counts = await getKafkaMessageCounts(ctx.log, admin, topic, groupId)
 
-    // if we have less than 50k messages in the queue we can trigger 50k oldest results (we process between 100k and 300k results per hour on average)
-    if (counts.unconsumed < 50000) {
-      const dbConnection = await getDbConnection(WRITE_DB_CONFIG())
+      // if we have less than 50k messages in the queue we can trigger 50k oldest results (we process between 100k and 300k results per hour on average)
+      if (counts.unconsumed < 50000) {
+        const dbConnection = await getDbConnection(WRITE_DB_CONFIG())
 
-      // we check if we have more than unconsumed pending results so that we don't trigger just the ones in the queue :)
-      const count = (
-        await dbConnection.one(
-          `select count(*) as count from integration.results where state = '${IntegrationResultState.PENDING}'`,
-        )
-      ).count
+        // we check if we have more than unconsumed pending results so that we don't trigger just the ones in the queue :)
+        const count = (
+          await dbConnection.one(
+            `select count(*) as count from integration.results where state = '${IntegrationResultState.PENDING}'`,
+          )
+        ).count
 
-      if (count > counts.unconsumed) {
-        ctx.log.info(`We have ${count} pending results, triggering 100k oldest results!`)
+        if (count > counts.unconsumed) {
+          ctx.log.info(`We have ${count} pending results, triggering 100k oldest results!`)
 
-        const queueService = new KafkaQueueService(kafkaClient, ctx.log)
-        const dswEmitter = new DataSinkWorkerEmitter(queueService, ctx.log)
-        await dswEmitter.init()
+          const queueService = new KafkaQueueService(kafkaClient, ctx.log)
+          const dswEmitter = new DataSinkWorkerEmitter(queueService, ctx.log)
+          await dswEmitter.init()
 
-        const resultIds = (
-          await dbConnection.any(
-            `select id from integration.results where state = 'pending' order by "createdAt" desc limit 100000`,
-          )
-        ).map((r) => r.id)
-
-        let triggered = 0
-
-        for (const batch of partition(resultIds, 10)) {
-          const messages = batch.map((resultId) => {
-            return {
-              payload: {
-                type: DataSinkWorkerQueueMessageType.PROCESS_INTEGRATION_RESULT,
-                resultId,
-              },
-              groupId: generateUUIDv1(),
-              deduplicationId: resultId,
-            }
-          })
+          const resultIds = (
+            await dbConnection.any(
+              `select id from integration.results where state = 'pending' order by "createdAt" desc limit 100000`,
+            )
+          ).map((r) => r.id)
+
+          let triggered = 0
 
-          await dswEmitter.sendMessages(messages)
+          for (const batch of partition(resultIds, 10)) {
+            const messages = batch.map((resultId) => {
+              return {
+                payload: {
+                  type: DataSinkWorkerQueueMessageType.PROCESS_INTEGRATION_RESULT,
+                  resultId,
+                },
+                groupId: generateUUIDv1(),
+                deduplicationId: resultId,
+              }
+            })
 
-          triggered += batch.length
+            await dswEmitter.sendMessages(messages)
 
-          if (triggered % 1000 === 0) {
-            ctx.log.info(`Triggered ${triggered} results!`)
+            triggered += batch.length
+
+            if (triggered % 1000 === 0) {
+              ctx.log.info(`Triggered ${triggered} results!`)
+            }
           }
-        }
 
-        ctx.log.info(`Triggered ${triggered} results in total!`)
+          ctx.log.info(`Triggered ${triggered} results in total!`)
+        }
+      } else {
+        ctx.log.info(`We have ${counts.unconsumed} unconsumed messages in the queue, skipping!`)
       }
-    } else {
-      ctx.log.info(`We have ${counts.unconsumed} unconsumed messages in the queue, skipping!`)
+    } finally {
+      await admin.disconnect()
     }
   },
 }

services/apps/cron_service/src/jobs/integrationResultsReporting.job.ts

@@ -1,6 +1,6 @@
 import CronTime from 'cron-time-generator'
 
-import { IS_DEV_ENV } from '@crowd/common'
+import { IS_DEV_ENV, IS_PROD_ENV } from '@crowd/common'
 import { READ_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
 import {
   SlackChannel,
@@ -33,6 +33,7 @@ const job: IJobDefinition = {
   name: 'integration-results-reporting',
   cronTime: IS_DEV_ENV ? CronTime.everyMinute() : CronTime.everyDayAt(8, 30),
   timeout: 10 * 60, // 10 minutes
+  enabled: async () => IS_PROD_ENV,
   process: async (ctx) => {
     ctx.log.info('Running integration-results-reporting job...')