chore: split how we execute github nango sync workflow so it's one per integration (#3657)

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

Author: themarolt

backend/src/services/integrationService.ts

@@ -26,7 +26,7 @@ import {
   startNangoSync,
 } from '@crowd/nango'
 import { RedisCache } from '@crowd/redis'
-import { WorkflowIdReusePolicy } from '@crowd/temporal'
+import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal'
 import { CodePlatform, Edition, PlatformType } from '@crowd/types'
 
 import { IRepositoryOptions } from '@/database/repositories/IRepositoryOptions'
@@ -919,11 +919,12 @@ export default class IntegrationService {
       await this.options.temporal.workflow.start('syncGithubIntegration', {
         taskQueue: 'nango',
         workflowId: `github-nango-sync/${integration.id}`,
-        workflowIdReusePolicy: WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
+        workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+        workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
         retry: {
           maximumAttempts: 10,
         },
-        args: [{ integrationIds: [integration.id] }],
+        args: [{ integrationId: integration.id }],
       })
 
       return await this.findById(integration.id)

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

@@ -1,11 +1,16 @@
 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 { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { NangoIntegration, nangoIntegrationToPlatform } from '@crowd/nango'
-import { TEMPORAL_CONFIG, WorkflowIdReusePolicy, getTemporalClient } from '@crowd/temporal'
+import {
+  TEMPORAL_CONFIG,
+  WorkflowIdConflictPolicy,
+  WorkflowIdReusePolicy,
+  getTemporalClient,
+} from '@crowd/temporal'
 import { PlatformType } from '@crowd/types'
 
 import { IJobDefinition } from '../types'
@@ -15,7 +20,7 @@ const job: IJobDefinition = {
   cronTime: CronTime.every(
     Number(process.env.CROWD_GH_NANGO_SYNC_INTERVAL_MINUTES || IS_DEV_ENV ? 5 : 60),
   ).minutes(),
-  timeout: 10 * 60,
+  timeout: 4 * 60 * 60, // 4 hours
   process: async (ctx) => {
     ctx.log.info('Triggering nango API check as if a webhook was received!')
 
@@ -27,9 +32,13 @@ const job: IJobDefinition = {
       nangoIntegrationToPlatform(NangoIntegration.GITHUB),
     ])
 
-    const ids: string[] = []
+    const limiter = new ConcurrencyLimiter(5)
 
-    for (const int of integrations) {
+    // Collect all workflow start operations
+    const workflowStarts: Array<() => Promise<void>> = []
+
+    for (let i = 0; i < integrations.length; i++) {
+      const int = integrations[i]
       const { id, platform } = int
 
       if (platform !== PlatformType.GITHUB_NANGO) {
@@ -40,22 +49,51 @@ const job: IJobDefinition = {
         {
           integrationId: id,
         },
-        'Triggering nango github integration sync!',
+        `${i + 1}/${integrations.length} Triggering nango github integration sync!`,
       )
 
-      ids.push(id)
+      workflowStarts.push(async () => {
+        await temporal.workflow
+          .start('syncGithubIntegration', {
+            taskQueue: 'nango',
+            workflowId: `github-nango-sync/cron/${id}`,
+            workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+            workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
+            retry: {
+              maximumAttempts: 10,
+            },
+            args: [{ integrationId: id }],
+          })
+          .catch((err) =>
+            ctx.log.error(err, 'Error while triggering nango github integration sync!'),
+          )
+      })
     }
 
-    await temporal.workflow.start('syncGithubIntegration', {
-      taskQueue: 'nango',
-      workflowId: `github-nango-sync/cron/${new Date().toISOString()}`,
-      workflowIdReusePolicy:
-        WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY,
-      retry: {
-        maximumAttempts: 10,
-      },
-      args: [{ integrationIds: ids }],
+    ctx.log.info(
+      `Triggering nango github integration syncs 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 github integration syncs 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 github integration syncs in total`)
   },
 }
 

services/apps/nango_worker/src/activities.ts

@@ -1,10 +1,10 @@
 import {
   analyzeGithubIntegration,
+  canCreateGithubConnection,
   createGithubConnection,
   deleteConnection,
   logInfo,
   mapGithubRepo,
-  numberOfGithubConnectionsToCreate,
   processNangoWebhook,
   removeGithubConnection,
   setGithubConnection,
@@ -24,7 +24,7 @@ export {
   startNangoSync,
   mapGithubRepo,
   unmapGithubRepo,
-  numberOfGithubConnectionsToCreate,
+  canCreateGithubConnection,
   updateGitIntegrationWithRepo,
   syncGithubReposToInsights,
   logInfo,

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

@@ -50,23 +50,23 @@ async function getLastConnectTs(): Promise<Date | undefined> {
   return new Date(lastConnect)
 }
 
-export async function numberOfGithubConnectionsToCreate(): Promise<number> {
-  const max = Number(process.env.CROWD_MAX_GH_NANGO_CONNECTIONS_PER_HOUR || 1)
+export async function canCreateGithubConnection(): Promise<boolean> {
+  const minutes = Number(process.env.CROWD_MINUTES_BETWEEN_GH_NANGO_CONNECTION || 6)
 
-  svc.log.info(`[GITHUB] Max number of github connections to create: ${max}`)
+  svc.log.info(`[GITHUB] Min minutes between connection creation: ${minutes}`)
 
   if (IS_DEV_ENV || IS_STAGING_ENV) {
-    svc.log.info('[GITHUB] Number of github connections to create: 10')
-    return 10
+    svc.log.info('[GITHUB] DEV MODE - we can create a connection!')
+    return true
   }
 
   const lastConnectDate = await getLastConnectTs()
 
   svc.log.info(`[GITHUB] Last connect date: ${lastConnectDate.toISOString()}`)
 
   if (!lastConnectDate) {
-    svc.log.info(`[GITHUB] Number of github connections to create: ${max}`)
-    return max
+    svc.log.info('[GITHUB] no last connect date found - we can create a connection!')
+    return true
   }
 
   const now = new Date()
@@ -75,17 +75,21 @@ export async function numberOfGithubConnectionsToCreate(): Promise<number> {
   // time is milliseconds
   const diff = now.getTime() - lastConnectDate.getTime()
 
-  // how many hours
-  const hours = diff / (1000 * 60 * 60) // ms to seconds to minutes
-  svc.log.info(`[GITHUB] Diff: ${diff}, hours: ${hours}`)
+  // how many minutes from diff
+  const minutesSinceLastConnection = diff / (1000 * 60)
+  svc.log.info(`[GITHUB] Diff: ${diff}, minutes: ${minutesSinceLastConnection}`)
 
-  if (hours >= 1.0) {
-    svc.log.info(`[GITHUB] Number of github connections to create: ${max}`)
-    return max
+  if (minutesSinceLastConnection >= minutes) {
+    svc.log.info(
+      '[GITHUB] more time has passed since last connection - we can create a connection!',
+    )
+    return true
   }
 
-  svc.log.info('[GITHUB] Number of github connections to create: 0')
-  return 0
+  svc.log.info(
+    '[GITHUB] not enough time has passed since last connection - we cannot create a connection!',
+  )
+  return false
 }
 
 export async function processNangoWebhook(

services/apps/nango_worker/src/types.ts

@@ -7,7 +7,7 @@ export interface IProcessNangoWebhookArguments {
 }
 
 export interface ISyncGithubIntegrationArguments {
-  integrationIds: string[]
+  integrationId: string
 }
 
 export interface IGithubIntegrationSyncInstructions {

services/apps/nango_worker/src/workflows/syncGithubIntegration.ts

@@ -1,78 +1,68 @@
-import { proxyActivities, sleep } from '@temporalio/workflow'
+import { proxyActivities } from '@temporalio/workflow'
 
 import * as activities from '../activities/nangoActivities'
 import { ISyncGithubIntegrationArguments } from '../types'
 
-const REPO_ONBOARDING_INTERVAL_MINUTES = 6
-
 const activity = proxyActivities<typeof activities>({
   startToCloseTimeout: '2 hour',
   retry: { maximumAttempts: 20, backoffCoefficient: 2 },
 })
 
 export async function syncGithubIntegration(args: ISyncGithubIntegrationArguments): Promise<void> {
-  const limit = await activity.numberOfGithubConnectionsToCreate()
-  let created = 0
-
-  for (const integrationId of args.integrationIds) {
-    const result = await activity.analyzeGithubIntegration(integrationId)
+  const integrationId = args.integrationId
 
-    // delete connections that are no longer needed
-    for (const repo of result.reposToDelete) {
-      // delete nango connection
-      await activity.deleteConnection(integrationId, result.providerConfigKey, repo.connectionId)
+  const result = await activity.analyzeGithubIntegration(integrationId)
 
-      // delete connection from integrations.settings.nangoMapping object
-      await activity.removeGithubConnection(integrationId, repo.connectionId)
+  // delete connections that are no longer needed
+  for (const repo of result.reposToDelete) {
+    // delete nango connection
+    await activity.deleteConnection(integrationId, result.providerConfigKey, repo.connectionId)
 
-      // delete githubRepos mapping
-      await activity.unmapGithubRepo(integrationId, repo.repo)
-    }
+    // delete connection from integrations.settings.nangoMapping object
+    await activity.removeGithubConnection(integrationId, repo.connectionId)
 
-    // delete duplicate connections
-    for (const repo of result.duplicatesToDelete) {
-      // delete nango connection
-      await activity.deleteConnection(integrationId, result.providerConfigKey, repo.connectionId)
+    // delete githubRepos mapping
+    await activity.unmapGithubRepo(integrationId, repo.repo)
+  }
 
-      // delete connection from integrations.settings.nangoMapping object
-      await activity.removeGithubConnection(integrationId, repo.connectionId)
+  // delete duplicate connections
+  for (const repo of result.duplicatesToDelete) {
+    // delete nango connection
+    await activity.deleteConnection(integrationId, result.providerConfigKey, repo.connectionId)
 
-      // we don't unmap because this one was duplicated
-    }
+    // delete connection from integrations.settings.nangoMapping object
+    await activity.removeGithubConnection(integrationId, repo.connectionId)
 
-    // create connections for repos that are not already connected
-    for (const repo of result.reposToSync) {
-      if (created >= limit) {
-        await activity.logInfo(
-          `Max number of github connections reached! Skipping repo ${repo.owner}/${repo.repoName} from integration ${integrationId}!`,
-        )
-        continue
-      }
+    // we don't unmap because this one was duplicated
+  }
 
-      // create nango connection
-      const connectionId = await activity.createGithubConnection(integrationId, repo)
+  // create connections for repos that are not already connected
+  for (const repo of result.reposToSync) {
+    const canCreate = await activity.canCreateGithubConnection()
 
-      // add connection to integrations.settings.nangoMapping object
-      await activity.setGithubConnection(integrationId, repo, connectionId)
+    if (!canCreate) {
+      await activity.logInfo(
+        `Not enough time has passed since last connection! Skipping repo ${repo.owner}/${repo.repoName} from integration ${integrationId}!`,
+      )
+      continue
+    }
 
-      // add repo to githubRepos mapping if it's not already mapped
-      await activity.mapGithubRepo(integrationId, repo)
+    // create nango connection
+    const connectionId = await activity.createGithubConnection(integrationId, repo)
 
-      // add repo to git integration
-      await activity.updateGitIntegrationWithRepo(integrationId, repo)
+    // add connection to integrations.settings.nangoMapping object
+    await activity.setGithubConnection(integrationId, repo, connectionId)
 
-      // start nango sync
-      await activity.startNangoSync(integrationId, result.providerConfigKey, connectionId)
+    // add repo to githubRepos mapping if it's not already mapped
+    await activity.mapGithubRepo(integrationId, repo)
 
-      // sync repositories to segmentRepositories and insightsProjects after processing all repos
-      await activity.syncGithubReposToInsights(integrationId)
+    // add repo to git integration
+    await activity.updateGitIntegrationWithRepo(integrationId, repo)
 
-      created++
+    // start nango sync
+    await activity.startNangoSync(integrationId, result.providerConfigKey, connectionId)
 
-      if (created < limit) {
-        // fixed delay to spread onboarding evenly throughout the day
-        await sleep(REPO_ONBOARDING_INTERVAL_MINUTES * 60 * 1000)
-      }
-    }
+    // sync repositories to segmentRepositories and insightsProjects after processing all repos
+    await activity.syncGithubReposToInsights(integrationId)
   }
 }