Fix #229 - Implement For task and refactor jq expr into context

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

Author: ricardozanini

README.md

@@ -126,7 +126,7 @@ The table below lists the current state of this implementation. This table is a
 | Task Call | ❌ |
 | Task Do | ✅ |
 | Task Emit | ❌ | 
-| Task For | ❌ |
+| Task For | ✅ |
 | Task Fork | ❌ | 
 | Task Listen | ❌ | 
 | Task Raise | ✅ |

go.mod

@@ -19,9 +19,11 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/itchyny/timefmt-go v0.1.6 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/relvacode/iso8601 v1.6.0 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect

go.sum

@@ -14,6 +14,8 @@ github.com/go-playground/validator/v10 v10.25.0/go.mod h1:GGzBIJMuE98Ic/kJsBXbz1
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/itchyny/gojq v0.12.17 h1:8av8eGduDb5+rvEdaOO+zQUjA04MS0m3Ps8HiD+fceg=
 github.com/itchyny/gojq v0.12.17/go.mod h1:WBrEMkgAfAGO1LUcGOckBl5O726KPp+OlkKug0I/FEY=
 github.com/itchyny/timefmt-go v0.1.6 h1:ia3s54iciXDdzWzwaVKXZPbiXzxxnv1SPGFfM/myJ5Q=
@@ -23,6 +25,8 @@ github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjS
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/relvacode/iso8601 v1.6.0 h1:eFXUhMJN3Gz8Rcq82f9DTMW0svjtAVuIEULglM7QHTU=
+github.com/relvacode/iso8601 v1.6.0/go.mod h1:FlNp+jz+TXpyRqgmM7tnzHHzBnz776kmAH2h3sZCn0I=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=

impl/ctx/context.go

@@ -16,9 +16,13 @@ package ctx
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"fmt"
+	"github.com/google/uuid"
 	"github.com/serverlessworkflow/sdk-go/v3/model"
 	"sync"
+	"time"
 )
 
 var ErrWorkflowContextNotFound = errors.New("workflow context not found")
@@ -29,34 +33,56 @@ type ctxKey string
 
 const (
 	runnerCtxKey ctxKey = "wfRunnerContext"
-	varsContext         = "$context"
-	varsInput           = "$input"
-	varsOutput          = "$output"
-	varsWorkflow        = "$workflow"
+
+	varsContext  = "$context"
+	varsInput    = "$input"
+	varsOutput   = "$output"
+	varsWorkflow = "$workflow"
+	varsRuntime  = "$runtime"
+	varsTask     = "$task"
+
+	// TODO: script during the release to update this value programmatically
+	runtimeVersion = "v3.1.0"
+	runtimeName    = "CNCF Serverless Workflow Specification Go SDK"
 )
 
 type WorkflowContext interface {
+	SetStartedAt(t time.Time)
 	SetStatus(status StatusPhase)
-	SetTaskStatus(task string, status StatusPhase)
+	SetRawInput(input interface{})
 	SetInstanceCtx(value interface{})
 	GetInstanceCtx() interface{}
 	SetInput(input interface{})
 	GetInput() interface{}
 	SetOutput(output interface{})
 	GetOutput() interface{}
 	GetOutputAsMap() map[string]interface{}
-	AsJQVars() map[string]interface{}
+	GetVars() map[string]interface{}
+	SetTaskStatus(task string, status StatusPhase)
+	SetTaskRawInput(input interface{})
+	SetTaskRawOutput(output interface{})
+	SetTaskDef(task model.Task) error
+	SetTaskStartedAt(startedAt time.Time)
+	SetTaskName(name string)
+	SetTaskReference(ref string)
+	GetTaskReference() string
+	ClearTaskContext()
+	SetLocalExprVars(vars map[string]interface{})
+	AddLocalExprVars(vars map[string]interface{})
+	RemoveLocalExprVars(keys ...string)
 }
 
 // workflowContext holds the necessary data for the workflow execution within the instance.
 type workflowContext struct {
-	mu               sync.Mutex
-	input            interface{}            // $input can hold any type
-	output           interface{}            // $output can hold any type
-	context          map[string]interface{} // Holds `$context` as the key
-	definition       map[string]interface{} // $workflow representation in the context
-	StatusPhase      []StatusPhaseLog
-	TasksStatusPhase map[string][]StatusPhaseLog
+	mu                 sync.Mutex
+	input              interface{}            // $input can hold any type
+	output             interface{}            // $output can hold any type
+	context            map[string]interface{} // Holds `$context` as the key
+	workflowDescriptor map[string]interface{} // $workflow representation in the context
+	taskDescriptor     map[string]interface{} // $task representation in the context
+	localExprVars      map[string]interface{} // Local expression variables defined in a given task or private context. E.g. a For task $item.
+	StatusPhase        []StatusPhaseLog
+	TasksStatusPhase   map[string][]StatusPhaseLog
 }
 
 func NewWorkflowContext(workflow *model.Workflow) (WorkflowContext, error) {
@@ -65,19 +91,110 @@ func NewWorkflowContext(workflow *model.Workflow) (WorkflowContext, error) {
 	if err != nil {
 		return nil, err
 	}
-
-	workflowCtx.definition = workflowDef
+	workflowCtx.taskDescriptor = map[string]interface{}{}
+	workflowCtx.workflowDescriptor = map[string]interface{}{
+		varsWorkflow: map[string]interface{}{
+			"id":         uuid.NewString(),
+			"definition": workflowDef,
+		},
+	}
 	workflowCtx.SetStatus(PendingStatus)
 
 	return workflowCtx, nil
 }
 
-func (ctx *workflowContext) AsJQVars() map[string]interface{} {
+// WithWorkflowContext adds the workflowContext to a parent context
+func WithWorkflowContext(parent context.Context, wfCtx WorkflowContext) context.Context {
+	return context.WithValue(parent, runnerCtxKey, wfCtx)
+}
+
+// GetWorkflowContext retrieves the workflowContext from a context
+func GetWorkflowContext(ctx context.Context) (WorkflowContext, error) {
+	wfCtx, ok := ctx.Value(runnerCtxKey).(*workflowContext)
+	if !ok {
+		return nil, ErrWorkflowContextNotFound
+	}
+	return wfCtx, nil
+}
+
+func (ctx *workflowContext) SetStartedAt(t time.Time) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	wf, ok := ctx.workflowDescriptor[varsWorkflow].(map[string]interface{})
+	if !ok {
+		wf = make(map[string]interface{})
+		ctx.workflowDescriptor[varsWorkflow] = wf
+	}
+
+	startedAt, ok := wf["startedAt"].(map[string]interface{})
+	if !ok {
+		startedAt = make(map[string]interface{})
+		wf["startedAt"] = startedAt
+	}
+
+	startedAt["iso8601"] = t.UTC().Format(time.RFC3339)
+}
+
+func (ctx *workflowContext) SetRawInput(input interface{}) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	// Ensure the outer "workflow" map
+	wf, ok := ctx.workflowDescriptor[varsWorkflow].(map[string]interface{})
+	if !ok {
+		wf = make(map[string]interface{})
+		ctx.workflowDescriptor[varsWorkflow] = wf
+	}
+
+	// Store the input
+	wf["input"] = input
+}
+
+func (ctx *workflowContext) AddLocalExprVars(vars map[string]interface{}) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+	if ctx.localExprVars == nil {
+		ctx.localExprVars = map[string]interface{}{}
+	}
+	for k, v := range vars {
+		ctx.localExprVars[k] = v
+	}
+}
+
+func (ctx *workflowContext) RemoveLocalExprVars(keys ...string) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	if ctx.localExprVars == nil {
+		return
+	}
+
+	for _, k := range keys {
+		delete(ctx.localExprVars, k)
+	}
+}
+
+func (ctx *workflowContext) SetLocalExprVars(vars map[string]interface{}) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+	ctx.localExprVars = vars
+}
+
+func (ctx *workflowContext) GetVars() map[string]interface{} {
 	vars := make(map[string]interface{})
 	vars[varsInput] = ctx.GetInput()
 	vars[varsOutput] = ctx.GetOutput()
 	vars[varsContext] = ctx.GetInstanceCtx()
-	vars[varsOutput] = ctx.definition
+	vars[varsTask] = ctx.taskDescriptor[varsTask]
+	vars[varsWorkflow] = ctx.workflowDescriptor[varsWorkflow]
+	vars[varsRuntime] = map[string]interface{}{
+		"name":    runtimeName,
+		"version": runtimeVersion,
+	}
+	for varName, varValue := range ctx.localExprVars {
+		vars[varName] = varValue
+	}
 	return vars
 }
 
@@ -90,15 +207,6 @@ func (ctx *workflowContext) SetStatus(status StatusPhase) {
 	ctx.StatusPhase = append(ctx.StatusPhase, NewStatusPhaseLog(status))
 }
 
-func (ctx *workflowContext) SetTaskStatus(task string, status StatusPhase) {
-	ctx.mu.Lock()
-	defer ctx.mu.Unlock()
-	if ctx.TasksStatusPhase == nil {
-		ctx.TasksStatusPhase = map[string][]StatusPhaseLog{}
-	}
-	ctx.TasksStatusPhase[task] = append(ctx.TasksStatusPhase[task], NewStatusPhaseLog(status))
-}
-
 // SetInstanceCtx safely sets the `$context` value
 func (ctx *workflowContext) SetInstanceCtx(value interface{}) {
 	ctx.mu.Lock()
@@ -179,16 +287,121 @@ func (ctx *workflowContext) GetOutputAsMap() map[string]interface{} {
 	}
 }
 
-// WithWorkflowContext adds the workflowContext to a parent context
-func WithWorkflowContext(parent context.Context, wfCtx WorkflowContext) context.Context {
-	return context.WithValue(parent, runnerCtxKey, wfCtx)
+func (ctx *workflowContext) SetTaskStatus(task string, status StatusPhase) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+	if ctx.TasksStatusPhase == nil {
+		ctx.TasksStatusPhase = map[string][]StatusPhaseLog{}
+	}
+	ctx.TasksStatusPhase[task] = append(ctx.TasksStatusPhase[task], NewStatusPhaseLog(status))
 }
 
-// GetWorkflowContext retrieves the workflowContext from a context
-func GetWorkflowContext(ctx context.Context) (WorkflowContext, error) {
-	wfCtx, ok := ctx.Value(runnerCtxKey).(*workflowContext)
+func (ctx *workflowContext) SetTaskRawInput(input interface{}) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	task, ok := ctx.taskDescriptor[varsTask].(map[string]interface{})
 	if !ok {
-		return nil, ErrWorkflowContextNotFound
+		task = make(map[string]interface{})
+		ctx.taskDescriptor[varsTask] = task
 	}
-	return wfCtx, nil
+
+	task["input"] = input
+}
+
+func (ctx *workflowContext) SetTaskRawOutput(output interface{}) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	task, ok := ctx.taskDescriptor[varsTask].(map[string]interface{})
+	if !ok {
+		task = make(map[string]interface{})
+		ctx.taskDescriptor[varsTask] = task
+	}
+
+	task["output"] = output
+}
+
+func (ctx *workflowContext) SetTaskDef(task model.Task) error {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	if task == nil {
+		return errors.New("SetTaskDef called with nil model.Task")
+	}
+
+	defBytes, err := json.Marshal(task)
+	if err != nil {
+		return fmt.Errorf("failed to marshal task: %w", err)
+	}
+
+	var defMap map[string]interface{}
+	if err := json.Unmarshal(defBytes, &defMap); err != nil {
+		return fmt.Errorf("failed to unmarshal task into map: %w", err)
+	}
+
+	taskMap, ok := ctx.taskDescriptor[varsTask].(map[string]interface{})
+	if !ok {
+		taskMap = make(map[string]interface{})
+		ctx.taskDescriptor[varsTask] = taskMap
+	}
+
+	taskMap["definition"] = defMap
+
+	return nil
+}
+
+func (ctx *workflowContext) SetTaskStartedAt(startedAt time.Time) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	task, ok := ctx.taskDescriptor[varsTask].(map[string]interface{})
+	if !ok {
+		task = make(map[string]interface{})
+		ctx.taskDescriptor[varsTask] = task
+	}
+
+	task["startedAt"] = startedAt.UTC().Format(time.RFC3339)
+}
+
+func (ctx *workflowContext) SetTaskName(name string) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	task, ok := ctx.taskDescriptor[varsTask].(map[string]interface{})
+	if !ok {
+		task = make(map[string]interface{})
+		ctx.taskDescriptor[varsTask] = task
+	}
+
+	task["name"] = name
+}
+
+func (ctx *workflowContext) SetTaskReference(ref string) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	task, ok := ctx.taskDescriptor[varsTask].(map[string]interface{})
+	if !ok {
+		task = make(map[string]interface{})
+		ctx.taskDescriptor[varsTask] = task
+	}
+
+	task["reference"] = ref
+}
+
+func (ctx *workflowContext) GetTaskReference() string {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+	task, ok := ctx.taskDescriptor[varsTask].(map[string]interface{})
+	if !ok {
+		return ""
+	}
+	return task["reference"].(string)
+}
+
+func (ctx *workflowContext) ClearTaskContext() {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+	ctx.taskDescriptor[varsTask] = make(map[string]interface{})
 }