fix: get correct parameters

Author: kevinkim-ogp

packages/backend/src/helpers/compute-parameters.ts

@@ -94,6 +94,7 @@ function findAndSubstituteVariables(
         // special handling for actions within the loop
         // where variables are either from the items in the list
         // or from actions within the loop itself
+
         if (isInsideLoop) {
           const columnName = get(data, keyPath)
 

packages/backend/src/services/action.ts

@@ -4,12 +4,14 @@ import HttpError from '@/errors/http'
 import PartialStepError from '@/errors/partial-error'
 import StepError from '@/errors/step'
 import computeParameters from '@/helpers/compute-parameters'
+import { DEFAULT_JOB_OPTIONS } from '@/helpers/default-job-configuration'
 import globalVariable from '@/helpers/global-variable'
 import logger from '@/helpers/logger'
 import Execution from '@/models/execution'
 import ExecutionStep from '@/models/execution-step'
 import Flow from '@/models/flow'
 import Step from '@/models/step'
+import { enqueueActionJob } from '@/queues/action'
 
 type ProcessActionOptions = {
   flowId: string
@@ -34,10 +36,21 @@ export const processAction = async (options: ProcessActionOptions) => {
   const flow = await Flow.query()
     .findById(flowId)
     .withGraphJoined('user')
+    .withGraphFetched('steps')
     .throwIfNotFound()
   const execution = await Execution.query()
     .findById(executionId)
     .throwIfNotFound()
+  const hasLoop = flow.steps.filter((step) => step.key === 'forEach')
+
+  let isInsideLoop = false
+  let loopIndex = -1
+  if (hasLoop.length > 0) {
+    loopIndex = hasLoop[0].position
+    if (step.position > hasLoop[0].position) {
+      isInsideLoop = true
+    }
+  }
 
   const $ = await globalVariable({
     flow,
@@ -60,6 +73,16 @@ export const processAction = async (options: ProcessActionOptions) => {
     $.step.parameters,
     priorExecutionSteps,
     actionCommand.preprocessVariable,
+    testRun,
+    {
+      isInsideLoop,
+      loopIndex,
+      stepPositions: flow.steps.reduce((acc, step) => {
+        acc[step.id] = step.position
+        return acc
+      }, {} as Record<string, number>),
+    },
+    metadata,
   )
 
   $.step.parameters = computedParameters
@@ -116,7 +139,9 @@ export const processAction = async (options: ProcessActionOptions) => {
       dataOut: $.actionOutput.data?.raw ?? null,
       errorDetails: $.actionOutput.error ?? null,
       appKey: $.app.key,
+      key: step.key,
       jobId,
+      metadata,
     })
 
   let nextStep = null
@@ -127,6 +152,48 @@ export const processAction = async (options: ProcessActionOptions) => {
         .findById(runResult.nextStep.stepId)
         .throwIfNotFound()
       break
+
+    case 'start-loop': {
+      nextStep = null // nothing to do, we enqueue the next step here
+      const dataOut = $.actionOutput.data?.raw ?? null
+      const iterations = Math.min(500, Number(dataOut.iterations ?? 0))
+
+      if (!iterations || iterations === 0) {
+        break
+      }
+
+      const firstStepInLoop = await step.getNextStep()
+
+      // does nothing for test run, do not enqueue any jobs
+      if (!testRun) {
+        if (Number.isInteger(iterations)) {
+          const enqueuePromises = []
+          for (let i = 0; i < Number(iterations); i++) {
+            const enqueuePromise = enqueueActionJob({
+              appKey: firstStepInLoop.appKey,
+              jobName: `${executionId}-${firstStepInLoop.id}-${i + 1}`,
+              jobData: {
+                flowId,
+                executionId,
+                stepId: firstStepInLoop.id,
+                metadata: {
+                  ...metadata,
+                  loopIteration: i + 1,
+                },
+              },
+              jobOptions: {
+                ...DEFAULT_JOB_OPTIONS,
+                delay: i * 100,
+              },
+            })
+            enqueuePromises.push(enqueuePromise)
+          }
+          await Promise.all(enqueuePromises)
+        }
+      }
+
+      break
+    }
     case 'stop-execution':
       // Nothing to do, nextStep is already null.
       break
@@ -141,7 +208,10 @@ export const processAction = async (options: ProcessActionOptions) => {
     executionStep,
     computedParameters,
     nextStep,
-    nextStepMetadata: runResult.nextStepMetadata,
+    nextStepMetadata: {
+      ...runResult.nextStepMetadata,
+      ...metadata,
+    },
     executionError,
   }
 }