chore: refactor trigger workflows into parent - child workflows for better grouping in temporal ui

Signed-off-by: Uroš Marolt <[email protected]>

Author: themarolt

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

@@ -9,7 +9,6 @@ import {
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import {
   ALL_NANGO_INTEGRATIONS,
-  INangoWebhookPayload,
   NANGO_INTEGRATION_CONFIG,
   NangoIntegration,
   nangoIntegrationToPlatform,
@@ -40,6 +39,12 @@ const OLD_INTERVAL_MS = IS_DEV_ENV
   ? 15 * 60 * 1000 // 15 minutes
   : 6 * 60 * 60 * 1000 // 6 hours
 
+interface INangoConnectionToCheck {
+  connectionId: string
+  models: string[]
+  workflowIdPrefix: string
+}
+
 const job: IJobDefinition = {
   name: 'nango-trigger',
   cronTime: IS_DEV_ENV ? CronTime.every(5).minutes() : CronTime.everyHour(),
@@ -87,6 +92,9 @@ const job: IJobDefinition = {
       const connectionIds: string[] =
         platform === NangoIntegration.GITHUB ? Object.keys(settings.nangoMapping) : [id]
 
+      const models = Object.values(NANGO_INTEGRATION_CONFIG[platform].models) as string[]
+      const connections: INangoConnectionToCheck[] = []
+
       for (const connectionId of connectionIds) {
         const key = `${id}/${connectionId}`
         const lastCheckedAt = lastCheckedAtMap.get(key)
@@ -104,35 +112,30 @@ const job: IJobDefinition = {
           `${i + 1}/${allIntegrations.length} Triggering nango integration check for ${id} / ${connectionId} (${platform})`,
         )
 
-        for (const model of Object.values(NANGO_INTEGRATION_CONFIG[platform].models)) {
-          const payload: INangoWebhookPayload = {
-            connectionId,
-            providerConfigKey: platform,
-            syncName: 'not important',
-            model,
-            responseResults: { added: 1, updated: 1, deleted: 1 },
-            syncType: 'INCREMENTAL',
-            modifiedAfter: new Date().toISOString(),
-          }
+        let workflowIdPrefix = ''
+        if (platform === NangoIntegration.GITHUB) {
+          const mapping = settings.nangoMapping[connectionId]
+          workflowIdPrefix = `${mapping.owner}/${mapping.repoName}/${connectionId}`
+        }
+
+        connections.push({ connectionId, models, workflowIdPrefix })
+      }
 
-          const workflowId =
-            platform === NangoIntegration.GITHUB
-              ? `nango-webhook/${platform}/${id}/${connectionId}/${model}/cron-triggered`
-              : `nango-webhook/${platform}/${id}/${model}/cron-triggered`
-
-          workflowStarts.push(async () => {
-            await temporal.workflow.start('processNangoWebhook', {
-              taskQueue: 'nango',
-              workflowId,
-              workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
-              workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
-              retry: {
-                maximumAttempts: 10,
-              },
-              args: [payload],
-            })
+      if (connections.length > 0) {
+        const workflowId = `nango-trigger/${platform}/${id}/cron-triggered`
+
+        workflowStarts.push(async () => {
+          await temporal.workflow.start('triggerNangoIntegrationCheck', {
+            taskQueue: 'nango',
+            workflowId,
+            workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+            workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
+            retry: {
+              maximumAttempts: 10,
+            },
+            args: [{ integrationId: id, providerConfigKey: platform, connections }],
           })
-        }
+        })
       }
     }
 

services/apps/nango_worker/src/types.ts

@@ -52,3 +52,15 @@ export interface ISyncGithubRepoArgs {
 export interface ISyncGithubRepoResult {
   skipped: boolean
 }
+
+export interface INangoConnectionToCheck {
+  connectionId: string
+  models: string[]
+  workflowIdPrefix: string
+}
+
+export interface ITriggerNangoIntegrationCheckArguments {
+  integrationId: string
+  providerConfigKey: string
+  connections: INangoConnectionToCheck[]
+}

services/apps/nango_worker/src/workflows.ts

@@ -3,11 +3,13 @@ import { deleteGithubRepoConnection } from './workflows/deleteGithubRepoConnecti
 import { processNangoWebhook } from './workflows/processNangoWebhook'
 import { syncGithubIntegration } from './workflows/syncGithubIntegration'
 import { syncGithubRepo } from './workflows/syncGithubRepo'
+import { triggerNangoIntegrationCheck } from './workflows/triggerNangoIntegrationCheck'
 
 export {
   deleteDuplicateGithubConnection,
   deleteGithubRepoConnection,
   processNangoWebhook,
   syncGithubIntegration,
   syncGithubRepo,
+  triggerNangoIntegrationCheck,
 }

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

@@ -0,0 +1,38 @@
+import { ParentClosePolicy, WorkflowIdReusePolicy, startChild } from '@temporalio/workflow'
+
+import { ITriggerNangoIntegrationCheckArguments } from '../types'
+
+import { processNangoWebhook } from './processNangoWebhook'
+
+export async function triggerNangoIntegrationCheck(
+  args: ITriggerNangoIntegrationCheckArguments,
+): Promise<void> {
+  for (const connection of args.connections) {
+    for (const model of connection.models) {
+      const workflowId = connection.workflowIdPrefix
+        ? `nango-webhook/${args.providerConfigKey}/${args.integrationId}/${connection.workflowIdPrefix}/${model}/cron-triggered`
+        : `nango-webhook/${args.providerConfigKey}/${args.integrationId}/${model}/cron-triggered`
+
+      try {
+        await startChild(processNangoWebhook, {
+          workflowId,
+          parentClosePolicy: ParentClosePolicy.PARENT_CLOSE_POLICY_ABANDON,
+          workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+          args: [
+            {
+              connectionId: connection.connectionId,
+              providerConfigKey: args.providerConfigKey,
+              model,
+              syncType: 'INCREMENTAL' as const,
+            },
+          ],
+        })
+      } catch (err) {
+        if (err instanceof Error && err.name === 'WorkflowExecutionAlreadyStartedError') {
+          continue
+        }
+        throw err
+      }
+    }
+  }
+}