chore: optimize how we trigger nango related workflows (CM-795) (#3642)

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

Author: themarolt

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

@@ -47,6 +47,7 @@ const job: IJobDefinition = {
 
     const ghMissingNangoConnections: Map<string, string[]> = new Map()
     const ghNotConnectedToNangoYet: Map<string, number> = new Map()
+    const ghNoCursorsYet: Map<string, number> = new Map()
 
     let totalRepos = 0
 
@@ -82,6 +83,15 @@ const job: IJobDefinition = {
         // then get nango connection statuses for each connection
         if (int.settings.nangoMapping) {
           for (const connectionId of Object.keys(int.settings.nangoMapping)) {
+            // check if we have cursors already for this connection
+            if (!int.settings.cursors || !int.settings.cursors[connectionId]) {
+              if (ghNoCursorsYet.has(int.id)) {
+                ghNoCursorsYet.set(int.id, ghNoCursorsYet.get(int.id) + 1)
+              } else {
+                ghNoCursorsYet.set(int.id, 1)
+              }
+            }
+
             const nangoConnection = singleOrDefault(
               nangoConnections,
               (c) => c.connection_id === connectionId,
@@ -188,13 +198,20 @@ const job: IJobDefinition = {
 
           if (ghNotConnectedToNangoYet.size > 0) {
             let totalNotConnected = 0
-            for (const [integrationId, count] of ghNotConnectedToNangoYet.entries()) {
-              slackMessage += `- *${integrationId}* has ${count} repos not connected to nango yet\n`
+            for (const count of Array.from(ghNotConnectedToNangoYet.values())) {
               totalNotConnected += count
             }
 
             slackMessage += `We have in total ${totalRepos} repos and ${totalNotConnected} of them are not connected to nango yet!\n`
           }
+
+          if (ghNoCursorsYet.size > 0) {
+            let totalNoCursors = 0
+            for (const count of Array.from(ghNoCursorsYet.values())) {
+              totalNoCursors += count
+            }
+            slackMessage += `And ${totalNoCursors} of them have no cursors yet!\n`
+          }
         }
 
         if (failedConnections.length > 0) {

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

@@ -1,8 +1,8 @@
 import CronTime from 'cron-time-generator'
 
-import { IS_DEV_ENV } from '@crowd/common'
+import { ConcurrencyLimiter, IS_DEV_ENV } from '@crowd/common'
 import { READ_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
-import { fetchNangoIntegrationData } from '@crowd/data-access-layer/src/integrations'
+import { fetchNangoIntegrationDataForCheck } from '@crowd/data-access-layer/src/integrations'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import {
   ALL_NANGO_INTEGRATIONS,
@@ -28,10 +28,15 @@ const job: IJobDefinition = {
 
     const dbConnection = await getDbConnection(READ_DB_CONFIG(), 3, 0)
 
-    const integrationsToTrigger = await fetchNangoIntegrationData(pgpQx(dbConnection), [
+    const integrationsToTrigger = await fetchNangoIntegrationDataForCheck(pgpQx(dbConnection), [
       ...new Set(ALL_NANGO_INTEGRATIONS.map(nangoIntegrationToPlatform)),
     ])
 
+    const limiter = new ConcurrencyLimiter(5)
+
+    // Collect all workflow start operations
+    const workflowStarts: Array<() => Promise<void>> = []
+
     for (let i = 0; i < integrationsToTrigger.length; i++) {
       const int = integrationsToTrigger[i]
 
@@ -71,10 +76,51 @@ const job: IJobDefinition = {
               modifiedAfter: new Date().toISOString(),
             }
 
+            workflowStarts.push(async () => {
+              try {
+                await temporal.workflow.start('processNangoWebhook', {
+                  taskQueue: 'nango',
+                  workflowId: `nango-webhook/${platform}/${id}/${connectionId}/${model}/cron-triggered`,
+                  workflowIdReusePolicy:
+                    WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
+                  retry: {
+                    maximumAttempts: 10,
+                  },
+                  args: [payload],
+                })
+              } catch (error) {
+                if (error.name === 'WorkflowExecutionAlreadyStartedError') {
+                  ctx.log.debug(
+                    {
+                      integrationId: id,
+                      platform,
+                      model,
+                      connectionId,
+                    },
+                    'Workflow already running, skipping...',
+                  )
+                  return
+                }
+                throw error
+              }
+            })
+          }
+        } else {
+          const payload: INangoWebhookPayload = {
+            connectionId: id,
+            providerConfigKey: platform,
+            syncName: 'not important',
+            model,
+            responseResults: { added: 1, updated: 1, deleted: 1 },
+            syncType: 'INCREMENTAL',
+            modifiedAfter: new Date().toISOString(),
+          }
+
+          workflowStarts.push(async () => {
             try {
               await temporal.workflow.start('processNangoWebhook', {
                 taskQueue: 'nango',
-                workflowId: `nango-webhook/${platform}/${id}/${connectionId}/${model}/cron-triggered`,
+                workflowId: `nango-webhook/${platform}/${id}/${model}/cron-triggered`,
                 workflowIdReusePolicy:
                   WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
                 retry: {
@@ -89,53 +135,40 @@ const job: IJobDefinition = {
                     integrationId: id,
                     platform,
                     model,
-                    connectionId,
                   },
                   'Workflow already running, skipping...',
                 )
-                continue
+                return
               }
               throw error
             }
-          }
-        } else {
-          const payload: INangoWebhookPayload = {
-            connectionId: id,
-            providerConfigKey: platform,
-            syncName: 'not important',
-            model,
-            responseResults: { added: 1, updated: 1, deleted: 1 },
-            syncType: 'INCREMENTAL',
-            modifiedAfter: new Date().toISOString(),
-          }
-
-          try {
-            await temporal.workflow.start('processNangoWebhook', {
-              taskQueue: 'nango',
-              workflowId: `nango-webhook/${platform}/${id}/${model}/cron-triggered`,
-              workflowIdReusePolicy: WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
-              retry: {
-                maximumAttempts: 10,
-              },
-              args: [payload],
-            })
-          } catch (error) {
-            if (error.name === 'WorkflowExecutionAlreadyStartedError') {
-              ctx.log.debug(
-                {
-                  integrationId: id,
-                  platform,
-                  model,
-                },
-                'Workflow already running, skipping...',
-              )
-              continue
-            }
-            throw error
-          }
+          })
         }
       }
     }
+
+    ctx.log.info(`Triggering nango integration checks with ${workflowStarts.length} workflows!`)
+
+    // Track completed workflows
+    let completedWorkflows = 0
+
+    // Register callback to track completed workflows
+    limiter.setOnJobComplete(() => {
+      completedWorkflows++
+      if (completedWorkflows % 100 === 0) {
+        ctx.log.info(`Triggered ${completedWorkflows} nango integrations checks so far...`)
+      }
+    })
+
+    // Process all workflow starts with concurrency limit
+    for (const workflowStart of workflowStarts) {
+      await limiter.schedule(workflowStart)
+    }
+
+    // Wait for all remaining jobs to complete
+    await limiter.waitForFinish()
+
+    ctx.log.info(`Triggered ${completedWorkflows} nango integrations checks in total`)
   },
 }
 

services/apps/nango_worker/package.json

@@ -9,7 +9,8 @@
     "lint": "npx eslint --ext .ts src --max-warnings=0",
     "format": "npx prettier --write \"src/**/*.ts\"",
     "format-check": "npx prettier --check .",
-    "tsc-check": "tsc --noEmit"
+    "tsc-check": "tsc --noEmit",
+    "check-nango-mapping": "tsx src/bin/check-nango-mapping.ts"
   },
   "dependencies": {
     "@crowd/archetype-standard": "workspace:*",

services/apps/nango_worker/src/activities/nangoActivities.ts

@@ -121,13 +121,16 @@ export async function processNangoWebhook(
 
   const settings = integration.settings
   let cursor = args.nextPageCursor
+  let existingCursor = false
   if (
     !cursor &&
     settings.cursors &&
     settings.cursors[args.connectionId] &&
-    settings.cursors[args.connectionId][args.model]
+    settings.cursors[args.connectionId][args.model] &&
+    !['<no-cursor>', '<no-records>'].includes(settings.cursors[args.connectionId][args.model])
   ) {
     cursor = settings.cursors[args.connectionId][args.model]
+    existingCursor = true
   }
 
   await initNangoCloudClient()
@@ -170,14 +173,33 @@ export async function processNangoWebhook(
 
       return records.nextCursor
     } else {
-      await setNangoIntegrationCursor(
-        dbStoreQx(svc.postgres.writer),
-        integration.id,
-        args.connectionId,
-        args.model,
-        records.records[records.records.length - 1].metadata.cursor,
-      )
+      let cursor = '<no-cursor>'
+      const lastRecord = records.records[records.records.length - 1]
+      if (lastRecord.metadata?.cursor) {
+        cursor = lastRecord.metadata.cursor
+      }
+
+      // if we dont have a cursor but we have an existing one we keep existing one
+      // if we have a cursor from the last record we also set it
+      if ((cursor === '<no-cursor>' && !existingCursor) || (cursor && cursor !== '<no-cursor>')) {
+        await setNangoIntegrationCursor(
+          dbStoreQx(svc.postgres.writer),
+          integration.id,
+          args.connectionId,
+          args.model,
+          cursor,
+        )
+      }
     }
+  } else if (!existingCursor) {
+    // only update if we don't have an existing cursor
+    await setNangoIntegrationCursor(
+      dbStoreQx(svc.postgres.writer),
+      integration.id,
+      args.connectionId,
+      args.model,
+      '<no-records>',
+    )
   }
 }