Add raise task

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

Author: ricardozanini

expr/expr.go

@@ -61,7 +61,7 @@ func TraverseAndEvaluate(node interface{}, input interface{}) (interface{}, erro
 	case string:
 		// Check if the string is a runtime expression (e.g., ${ .some.path })
 		if IsStrictExpr(v) {
-			return EvaluateJQExpression(Sanitize(v), input)
+			return evaluateJQExpression(Sanitize(v), input)
 		}
 		return v, nil
 
@@ -71,8 +71,8 @@ func TraverseAndEvaluate(node interface{}, input interface{}) (interface{}, erro
 	}
 }
 
-// EvaluateJQExpression evaluates a jq expression against a given JSON input
-func EvaluateJQExpression(expression string, input interface{}) (interface{}, error) {
+// evaluateJQExpression evaluates a jq expression against a given JSON input
+func evaluateJQExpression(expression string, input interface{}) (interface{}, error) {
 	// Parse the sanitized jq expression
 	query, err := gojq.Parse(expression)
 	if err != nil {

impl/runner.go

@@ -51,7 +51,7 @@ func (wr *workflowRunnerImpl) Run(input interface{}) (output interface{}, err er
 	}()
 
 	// Process input
-	if input, err = wr.processInput(input); err != nil {
+	if input, err = wr.processWorkflowInput(input); err != nil {
 		return nil, err
 	}
 
@@ -64,16 +64,24 @@ func (wr *workflowRunnerImpl) Run(input interface{}) (output interface{}, err er
 	output = wr.RunnerCtx.GetOutput()
 
 	// Process output
-	if output, err = wr.processOutput(output); err != nil {
+	if output, err = wr.processWorkflowOutput(output); err != nil {
 		return nil, err
 	}
 
 	wr.RunnerCtx.SetStatus(CompletedStatus)
 	return output, nil
 }
 
-// processInput validates and transforms input if needed.
-func (wr *workflowRunnerImpl) processInput(input interface{}) (interface{}, error) {
+// wrapWorkflowError ensures workflow errors have a proper instance reference.
+func (wr *workflowRunnerImpl) wrapWorkflowError(err error, taskName string) error {
+	if knownErr := model.AsError(err); knownErr != nil {
+		return knownErr.WithInstanceRef(wr.Workflow, taskName)
+	}
+	return model.NewErrRuntime(err, taskName)
+}
+
+// processWorkflowInput validates and transforms input if needed.
+func (wr *workflowRunnerImpl) processWorkflowInput(input interface{}) (interface{}, error) {
 	if wr.Workflow.Input != nil {
 		var err error
 		if err = validateSchema(input, wr.Workflow.Input.Schema, "/"); err != nil {
@@ -99,39 +107,45 @@ func (wr *workflowRunnerImpl) executeTasks(tasks *model.TaskList) error {
 		return nil
 	}
 
-	// TODO: implement control flow: continue, end, then
-	for _, taskItem := range *tasks {
+	idx := 0
+	currentTask := (*tasks)[idx]
+
+	for currentTask != nil {
 		wr.RunnerCtx.SetInput(wr.RunnerCtx.GetOutput())
-		if shouldRun, err := wr.shouldRunTask(taskItem); err != nil {
+		if shouldRun, err := wr.shouldRunTask(currentTask); err != nil {
 			return err
 		} else if !shouldRun {
 			wr.RunnerCtx.SetOutput(wr.RunnerCtx.GetInput())
+			idx, currentTask = tasks.Next(idx)
 			continue
 		}
 
-		wr.RunnerCtx.SetTaskStatus(taskItem.Key, PendingStatus)
-		runner, err := NewTaskRunner(taskItem.Key, taskItem.Task)
+		wr.RunnerCtx.SetTaskStatus(currentTask.Key, PendingStatus)
+		runner, err := NewTaskRunner(currentTask.Key, currentTask.Task, wr)
 		if err != nil {
 			return err
 		}
 
-		wr.RunnerCtx.SetTaskStatus(taskItem.Key, RunningStatus)
+		wr.RunnerCtx.SetTaskStatus(currentTask.Key, RunningStatus)
 		var output interface{}
-		if output, err = wr.runTask(runner, taskItem.Task.GetBase()); err != nil {
-			wr.RunnerCtx.SetTaskStatus(taskItem.Key, FaultedStatus)
+		if output, err = wr.runTask(runner, currentTask.Task.GetBase()); err != nil {
+			wr.RunnerCtx.SetTaskStatus(currentTask.Key, FaultedStatus)
 			return err
 		}
+		// TODO: make sure that `output` is a map[string]interface{}, so compatible to JSON traversal.
 
-		wr.RunnerCtx.SetTaskStatus(taskItem.Key, CompletedStatus)
+		wr.RunnerCtx.SetTaskStatus(currentTask.Key, CompletedStatus)
 		wr.RunnerCtx.SetOutput(output)
+
+		idx, currentTask = tasks.Next(idx)
 	}
 
 	return nil
 }
 
 func (wr *workflowRunnerImpl) shouldRunTask(task *model.TaskItem) (bool, error) {
 	if task.GetBase().If != nil {
-		output, err := expr.EvaluateJQExpression(task.GetBase().If.String(), wr.RunnerCtx.GetInput())
+		output, err := expr.TraverseAndEvaluate(task.GetBase().If.String(), wr.RunnerCtx.GetInput())
 		if err != nil {
 			return false, model.NewErrExpression(err, task.Key)
 		}
@@ -142,8 +156,8 @@ func (wr *workflowRunnerImpl) shouldRunTask(task *model.TaskItem) (bool, error)
 	return true, nil
 }
 
-// processOutput applies output transformations.
-func (wr *workflowRunnerImpl) processOutput(output interface{}) (interface{}, error) {
+// processWorkflowOutput applies output transformations.
+func (wr *workflowRunnerImpl) processWorkflowOutput(output interface{}) (interface{}, error) {
 	if wr.Workflow.Output != nil {
 		var err error
 		if output, err = traverseAndEvaluate(wr.Workflow.Output.As, wr.RunnerCtx.GetOutput(), "/"); err != nil {
@@ -161,16 +175,40 @@ func (wr *workflowRunnerImpl) processOutput(output interface{}) (interface{}, er
 
 // ----------------- Task funcs ------------------- //
 
+// TODO: refactor to receive a resolver handler instead of the workflow runner
+
 // NewTaskRunner creates a TaskRunner instance based on the task type.
-func NewTaskRunner(taskName string, task model.Task) (TaskRunner, error) {
+func NewTaskRunner(taskName string, task model.Task, wr *workflowRunnerImpl) (TaskRunner, error) {
 	switch t := task.(type) {
 	case *model.SetTask:
 		return NewSetTaskRunner(taskName, t)
+	case *model.RaiseTask:
+		if err := wr.resolveErrorDefinition(t); err != nil {
+			return nil, err
+		}
+		return NewRaiseTaskRunner(taskName, t)
 	default:
 		return nil, fmt.Errorf("unsupported task type '%T' for task '%s'", t, taskName)
 	}
 }
 
+// TODO: can e refactored to a definition resolver callable from the context
+func (wr *workflowRunnerImpl) resolveErrorDefinition(t *model.RaiseTask) error {
+	if t.Raise.Error.Ref != nil {
+		notFoundErr := model.NewErrValidation(fmt.Errorf("%v error definition not found in 'uses'", t.Raise.Error.Ref), "")
+		if wr.Workflow.Use != nil && wr.Workflow.Use.Errors != nil {
+			definition, ok := wr.Workflow.Use.Errors[*t.Raise.Error.Ref]
+			if !ok {
+				return notFoundErr
+			}
+			t.Raise.Error.Definition = definition
+			return nil
+		}
+		return notFoundErr
+	}
+	return nil
+}
+
 // runTask executes an individual task.
 func (wr *workflowRunnerImpl) runTask(runner TaskRunner, task *model.TaskBase) (output interface{}, err error) {
 	taskInput := wr.RunnerCtx.GetInput()
@@ -234,14 +272,6 @@ func (wr *workflowRunnerImpl) validateAndEvaluateTaskOutput(task *model.TaskBase
 	return output, nil
 }
 
-// wrapWorkflowError ensures workflow errors have a proper instance reference.
-func (wr *workflowRunnerImpl) wrapWorkflowError(err error, taskName string) error {
-	if knownErr := model.AsError(err); knownErr != nil {
-		return knownErr.WithInstanceRef(wr.Workflow, taskName)
-	}
-	return model.NewErrRuntime(err, taskName)
-}
-
 func validateSchema(data interface{}, schema *model.Schema, taskName string) error {
 	if schema != nil {
 		if err := ValidateJSONSchema(data, schema); err != nil {

impl/task.go

@@ -7,6 +7,7 @@ import (
 )
 
 var _ TaskRunner = &SetTaskRunner{}
+var _ TaskRunner = &RaiseTaskRunner{}
 
 type TaskRunner interface {
 	Run(input interface{}) (interface{}, error)
@@ -50,3 +51,65 @@ func (s *SetTaskRunner) Run(input interface{}) (output interface{}, err error) {
 
 	return output, nil
 }
+
+func NewRaiseTaskRunner(taskName string, task *model.RaiseTask) (*RaiseTaskRunner, error) {
+	if task == nil || task.Raise.Error.Definition == nil {
+		return nil, model.NewErrValidation(fmt.Errorf("no raise configuration provided for RaiseTask %s", taskName), taskName)
+	}
+	return &RaiseTaskRunner{
+		Task:     task,
+		TaskName: taskName,
+	}, nil
+}
+
+type RaiseTaskRunner struct {
+	Task     *model.RaiseTask
+	TaskName string
+}
+
+var raiseErrFuncMapping = map[string]func(error, string) *model.Error{
+	model.ErrorTypeAuthentication: model.NewErrAuthentication,
+	model.ErrorTypeValidation:     model.NewErrValidation,
+	model.ErrorTypeCommunication:  model.NewErrCommunication,
+	model.ErrorTypeAuthorization:  model.NewErrAuthorization,
+	model.ErrorTypeConfiguration:  model.NewErrConfiguration,
+	model.ErrorTypeExpression:     model.NewErrExpression,
+	model.ErrorTypeRuntime:        model.NewErrRuntime,
+	model.ErrorTypeTimeout:        model.NewErrTimeout,
+}
+
+func (r *RaiseTaskRunner) Run(input interface{}) (output interface{}, err error) {
+	output = input
+	// TODO: make this an external func so we can call it after getting the reference? Or we can get the reference from the workflow definition
+	var detailResult interface{}
+	detailResult, err = traverseAndEvaluate(r.Task.Raise.Error.Definition.Detail.AsObjectOrRuntimeExpr(), input, r.TaskName)
+	if err != nil {
+		return nil, err
+	}
+
+	var titleResult interface{}
+	titleResult, err = traverseAndEvaluate(r.Task.Raise.Error.Definition.Title.AsObjectOrRuntimeExpr(), input, r.TaskName)
+	if err != nil {
+		return nil, err
+	}
+
+	instance := &model.JsonPointerOrRuntimeExpression{Value: r.TaskName}
+
+	var raiseErr *model.Error
+	if raiseErrF, ok := raiseErrFuncMapping[r.Task.Raise.Error.Definition.Type.String()]; ok {
+		raiseErr = raiseErrF(fmt.Errorf("%v", detailResult), instance.String())
+	} else {
+		raiseErr = r.Task.Raise.Error.Definition
+		raiseErr.Detail = model.NewStringOrRuntimeExpr(fmt.Sprintf("%v", detailResult))
+		raiseErr.Instance = instance
+	}
+
+	raiseErr.Title = model.NewStringOrRuntimeExpr(fmt.Sprintf("%v", titleResult))
+	err = raiseErr
+
+	return output, err
+}
+
+func (r *RaiseTaskRunner) GetTaskName() string {
+	return r.TaskName
+}

impl/task_raise_test.go

@@ -0,0 +1,142 @@
+package impl
+
+import (
+	"encoding/json"
+	"errors"
+	"github.com/serverlessworkflow/sdk-go/v3/model"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestRaiseTaskRunner_WithDefinedError(t *testing.T) {
+	input := map[string]interface{}{}
+
+	raiseTask := &model.RaiseTask{
+		Raise: model.RaiseTaskConfiguration{
+			Error: model.RaiseTaskError{
+				Definition: &model.Error{
+					Type:   model.NewUriTemplate(model.ErrorTypeValidation),
+					Status: 400,
+					Title:  model.NewStringOrRuntimeExpr("Validation Error"),
+					Detail: model.NewStringOrRuntimeExpr("Invalid input data"),
+				},
+			},
+		},
+	}
+
+	runner, err := NewRaiseTaskRunner("task_raise_defined", raiseTask)
+	assert.NoError(t, err)
+
+	output, err := runner.Run(input)
+	assert.Equal(t, output, input)
+	assert.Error(t, err)
+
+	expectedErr := model.NewErrValidation(errors.New("Invalid input data"), "task_raise_defined")
+
+	assert.Equal(t, expectedErr.Type.String(), err.(*model.Error).Type.String())
+	assert.Equal(t, expectedErr.Status, err.(*model.Error).Status)
+	assert.Equal(t, expectedErr.Title.String(), err.(*model.Error).Title.String())
+	assert.Equal(t, "Invalid input data", err.(*model.Error).Detail.String())
+	assert.Equal(t, expectedErr.Instance.String(), err.(*model.Error).Instance.String())
+}
+
+func TestRaiseTaskRunner_WithReferencedError(t *testing.T) {
+	var ref string
+	ref = "someErrorRef"
+	raiseTask := &model.RaiseTask{
+		Raise: model.RaiseTaskConfiguration{
+			Error: model.RaiseTaskError{
+				Ref: &ref,
+			},
+		},
+	}
+
+	runner, err := NewRaiseTaskRunner("task_raise_ref", raiseTask)
+	assert.Error(t, err)
+	assert.Nil(t, runner)
+}
+
+func TestRaiseTaskRunner_TimeoutErrorWithExpression(t *testing.T) {
+	input := map[string]interface{}{
+		"timeoutMessage": "Request took too long",
+	}
+
+	raiseTask := &model.RaiseTask{
+		Raise: model.RaiseTaskConfiguration{
+			Error: model.RaiseTaskError{
+				Definition: &model.Error{
+					Type:   model.NewUriTemplate(model.ErrorTypeTimeout),
+					Status: 408,
+					Title:  model.NewStringOrRuntimeExpr("Timeout Error"),
+					Detail: model.NewStringOrRuntimeExpr("${ .timeoutMessage }"),
+				},
+			},
+		},
+	}
+
+	runner, err := NewRaiseTaskRunner("task_raise_timeout_expr", raiseTask)
+	assert.NoError(t, err)
+
+	output, err := runner.Run(input)
+	assert.Equal(t, input, output)
+	assert.Error(t, err)
+
+	expectedErr := model.NewErrTimeout(errors.New("Request took too long"), "task_raise_timeout_expr")
+
+	assert.Equal(t, expectedErr.Type.String(), err.(*model.Error).Type.String())
+	assert.Equal(t, expectedErr.Status, err.(*model.Error).Status)
+	assert.Equal(t, expectedErr.Title.String(), err.(*model.Error).Title.String())
+	assert.Equal(t, "Request took too long", err.(*model.Error).Detail.String())
+	assert.Equal(t, expectedErr.Instance.String(), err.(*model.Error).Instance.String())
+}
+
+func TestRaiseTaskRunner_Serialization(t *testing.T) {
+	raiseTask := &model.RaiseTask{
+		Raise: model.RaiseTaskConfiguration{
+			Error: model.RaiseTaskError{
+				Definition: &model.Error{
+					Type:     model.NewUriTemplate(model.ErrorTypeRuntime),
+					Status:   500,
+					Title:    model.NewStringOrRuntimeExpr("Runtime Error"),
+					Detail:   model.NewStringOrRuntimeExpr("Unexpected failure"),
+					Instance: &model.JsonPointerOrRuntimeExpression{Value: "/task_runtime"},
+				},
+			},
+		},
+	}
+
+	data, err := json.Marshal(raiseTask)
+	assert.NoError(t, err)
+
+	var deserializedTask model.RaiseTask
+	err = json.Unmarshal(data, &deserializedTask)
+	assert.NoError(t, err)
+
+	assert.Equal(t, raiseTask.Raise.Error.Definition.Type.String(), deserializedTask.Raise.Error.Definition.Type.String())
+	assert.Equal(t, raiseTask.Raise.Error.Definition.Status, deserializedTask.Raise.Error.Definition.Status)
+	assert.Equal(t, raiseTask.Raise.Error.Definition.Title.String(), deserializedTask.Raise.Error.Definition.Title.String())
+	assert.Equal(t, raiseTask.Raise.Error.Definition.Detail.String(), deserializedTask.Raise.Error.Definition.Detail.String())
+	assert.Equal(t, raiseTask.Raise.Error.Definition.Instance.String(), deserializedTask.Raise.Error.Definition.Instance.String())
+}
+
+func TestRaiseTaskRunner_ReferenceSerialization(t *testing.T) {
+	var ref string
+	ref = "errorReference"
+	raiseTask := &model.RaiseTask{
+		Raise: model.RaiseTaskConfiguration{
+			Error: model.RaiseTaskError{
+				Ref: &ref,
+			},
+		},
+	}
+
+	data, err := json.Marshal(raiseTask)
+	assert.NoError(t, err)
+
+	var deserializedTask model.RaiseTask
+	err = json.Unmarshal(data, &deserializedTask)
+	assert.NoError(t, err)
+
+	assert.Equal(t, *raiseTask.Raise.Error.Ref, *deserializedTask.Raise.Error.Ref)
+	assert.Nil(t, deserializedTask.Raise.Error.Definition)
+}