Add partial 'For' implementation

Signed-off-by: Ricardo Zanini <[email protected]>

Author: ricardozanini

expr/expr.go

@@ -86,6 +86,8 @@ func TraverseAndEvaluate(node interface{}, input interface{}) (interface{}, erro
 	}
 }
 
+// TODO: add support to variables see https://github.com/itchyny/gojq/blob/main/option_variables_test.go
+
 // evaluateJQExpression evaluates a jq expression against a given JSON input
 func evaluateJQExpression(expression string, input interface{}) (interface{}, error) {
 	// Parse the sanitized jq expression

impl/task_runner.go

@@ -16,6 +16,8 @@ package impl
 
 import (
 	"fmt"
+	"reflect"
+	"strings"
 
 	"github.com/serverlessworkflow/sdk-go/v3/expr"
 	"github.com/serverlessworkflow/sdk-go/v3/model"
@@ -163,14 +165,86 @@ func NewForTaskRunner(taskName string, task *model.ForTask, taskSupport TaskSupp
 	}, nil
 }
 
+const (
+	forTaskDefaultEach = "$item"
+	forTaskDefaultAt   = "$index"
+)
+
 type ForTaskRunner struct {
 	Task     *model.ForTask
 	TaskName string
 	DoRunner *DoTaskRunner
 }
 
 func (f *ForTaskRunner) Run(input interface{}) (interface{}, error) {
-	return input, nil
+	f.sanitizeFor()
+	in, err := expr.TraverseAndEvaluate(f.Task.For.In, input)
+	if err != nil {
+		return nil, err
+	}
+
+	var forOutput interface{}
+	rv := reflect.ValueOf(in)
+	switch rv.Kind() {
+	case reflect.Slice, reflect.Array:
+		for i := 0; i < rv.Len(); i++ {
+			item := rv.Index(i).Interface()
+
+			if forOutput, err = f.processForItem(i, item, forOutput); err != nil {
+				return nil, err
+			}
+		}
+	case reflect.Invalid:
+		return input, nil
+	default:
+		if forOutput, err = f.processForItem(0, in, forOutput); err != nil {
+			return nil, err
+		}
+	}
+
+	return forOutput, nil
+}
+
+func (f *ForTaskRunner) processForItem(idx int, item interface{}, forOutput interface{}) (interface{}, error) {
+	forInput := map[string]interface{}{
+		f.Task.For.At:   idx,
+		f.Task.For.Each: item,
+	}
+	if forOutput != nil {
+		if outputMap, ok := forOutput.(map[string]interface{}); ok {
+			for key, value := range outputMap {
+				forInput[key] = value
+			}
+		} else {
+			return nil, fmt.Errorf("task %s item %s at index %d returned a non-json object, impossible to merge context", f.TaskName, f.Task.For.Each, idx)
+		}
+	}
+	var err error
+	forOutput, err = f.DoRunner.Run(forInput)
+	if err != nil {
+		return nil, err
+	}
+
+	return forOutput, nil
+}
+
+func (f *ForTaskRunner) sanitizeFor() {
+	f.Task.For.Each = strings.TrimSpace(f.Task.For.Each)
+	f.Task.For.At = strings.TrimSpace(f.Task.For.At)
+
+	if f.Task.For.Each == "" {
+		f.Task.For.Each = forTaskDefaultEach
+	}
+	if f.Task.For.At == "" {
+		f.Task.For.At = forTaskDefaultAt
+	}
+
+	if !strings.HasPrefix(f.Task.For.Each, "$") {
+		f.Task.For.Each = "$" + f.Task.For.Each
+	}
+	if !strings.HasPrefix(f.Task.For.At, "$") {
+		f.Task.For.At = "$" + f.Task.For.At
+	}
 }
 
 func (f *ForTaskRunner) GetTaskName() string {

impl/task_runner_test.go

@@ -309,5 +309,22 @@ func TestWorkflowRunner_Run_YAML_RaiseTasks_ControlFlow(t *testing.T) {
 				assert.Equal(t, "User is under the required age", model.AsError(err).Detail.String())
 			})
 	})
+}
+
+func TestForTaskRunner_Run(t *testing.T) {
+	t.Skip("Skipping until the For task is implemented - missing JQ variables implementation")
+	t.Run("Simple For with Colors", func(t *testing.T) {
+		workflowPath := "./testdata/for_colors.yaml"
+		input := map[string]interface{}{
+			"colors": []string{"red", "green", "blue"},
+		}
+		expectedOutput := map[string]interface{}{
+			"processed": map[string]interface{}{
+				"colors":  []string{"red", "green", "blue"},
+				"indexed": []float64{0, 1, 2},
+			},
+		}
+		runWorkflowTest(t, workflowPath, input, expectedOutput)
+	})
 
 }

impl/testdata/for_colors.yaml

@@ -0,0 +1,14 @@
+document:
+  dsl: '1.0.0'
+  namespace: default
+  name: for
+  version: '1.0.0'
+do:
+  - loopColors:
+      for:
+        each: color
+        in: '${ .colors }'
+      do:
+        - markProcessed:
+            set:
+              processed: '${ { colors: (.processed.colors + [ $color ]), indexes: (.processed.indexes + [ $index ])} }'

model/task.go

@@ -143,6 +143,8 @@ func unmarshalTask(key string, taskRaw json.RawMessage) (Task, error) {
 		return nil, fmt.Errorf("failed to parse task type for key '%s': %w", key, err)
 	}
 
+	// TODO: not the most elegant; can be improved in a smarter way
+
 	// Determine task type
 	var task Task
 	if callValue, hasCall := taskType["call"].(string); hasCall {
@@ -154,8 +156,11 @@ func unmarshalTask(key string, taskRaw json.RawMessage) (Task, error) {
 			// Default to CallFunction for unrecognized call values
 			task = &CallFunction{}
 		}
+	} else if _, hasFor := taskType["for"]; hasFor {
+		// Handle special case "for" that also has "do"
+		task = taskTypeRegistry["for"]()
 	} else {
-		// Handle non-call tasks (e.g., "do", "fork")
+		// Handle everything else (e.g., "do", "fork")
 		for typeKey := range taskType {
 			if constructor, exists := taskTypeRegistry[typeKey]; exists {
 				task = constructor()