Refactor, add generic step, add error&termination logic

automation/service/workflow.go

@@ -533,8 +533,11 @@ func (svc *workflow) workflowStepDefConv(g *wfexec.Graph, s *types.WorkflowStep,
 		case types.WorkflowStepKindFunction, types.WorkflowStepKindIterator:
 			return svc.convFunctionStep(g, s, out)
 
-		//case types.WorkflowStepKindMessage:
-		//	return svc.convMessageStep(s)
+		case types.WorkflowStepKindError:
+			return svc.convErrorStep(s, out)
+
+		case types.WorkflowStepKindTermination:
+			return svc.convTerminationStep(out)
 
 		case types.WorkflowStepKindPrompt:
 			return svc.convPromptStep(s)
@@ -734,6 +737,69 @@ func (svc *workflow) convFunctionStep(g *wfexec.Graph, s *types.WorkflowStep, ou
 	}
 }
 
+// creates error step
+//
+// Expects ZERO outgoing paths and
+func (svc *workflow) convErrorStep(s *types.WorkflowStep, out types.WorkflowPathSet) (wfexec.Step, error) {
+	const (
+		argName = "message"
+	)
+
+	if len(out) > 0 {
+		return nil, errors.Internal("error step must be last step in branch")
+	}
+
+	var (
+		args = types.ExprSet(s.Arguments)
+	)
+
+	if msgArg := args.GetByTarget(argName); msgArg == nil {
+		return nil, errors.Internal("error step must have %s argument", argName)
+	} else if msgArg.Type != (expr.String{}).Type() {
+		return nil, errors.Internal("%s argument on error step must be string, got type '%s'", argName, msgArg.Type)
+	} else if len(args) > 1 {
+		return nil, errors.Internal("too many arguments on error step")
+	}
+
+	if err := svc.parseExpressions(args...); err != nil {
+		return nil, err
+	}
+
+	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
+		var (
+			msg         string
+			result, err = args.Eval(ctx, r.Scope)
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		if result.Has(argName) {
+			str, _ := expr.NewString(expr.Must(result.Select(argName)))
+			msg = str.GetValue()
+		} else {
+			if aux, is := args.GetByTarget(argName).Value.(string); is {
+				msg = aux
+			} else {
+				msg = "ERROR"
+			}
+		}
+
+		return nil, errors.Automation(msg)
+	}), nil
+}
+
+// converts prompt definition to wfexec.Step
+func (svc *workflow) convTerminationStep(out types.WorkflowPathSet) (wfexec.Step, error) {
+	if len(out) > 0 {
+		return nil, errors.Internal("termination step must be last step in branch")
+	}
+
+	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
+		return wfexec.Termination(), nil
+	}), nil
+}
+
 // converts prompt definition to wfexec.Step
 func (svc *workflow) convPromptStep(s *types.WorkflowStep) (wfexec.Step, error) {
 	if err := svc.parseExpressions(s.Arguments...); err != nil {
@@ -821,6 +887,12 @@ func validateSteps(ss ...*types.WorkflowStep) error {
 		case types.WorkflowStepKindGateway:
 			checks = append(checks, noArgs, noResults)
 
+		case types.WorkflowStepKindError:
+			checks = append(checks, noResults)
+
+		case types.WorkflowStepKindTermination:
+			checks = append(checks, noArgs, noResults)
+
 		case types.WorkflowStepKindFunction, types.WorkflowStepKindIterator:
 
 		case types.WorkflowStepKindPrompt:

automation/types/debug.go

@@ -9,7 +9,7 @@ import (
 
 type (
 	debugStep struct {
-		identifiableStep
+		wfexec.StepIdentifier
 		log *zap.Logger
 	}
 )

automation/types/error.go

@@ -0,0 +1,23 @@
+package types
+
+import (
+	"context"
+	"fmt"
+	"github.com/cortezaproject/corteza-server/pkg/wfexec"
+)
+
+type (
+	errorStep struct {
+		wfexec.StepIdentifier
+		message string
+	}
+)
+
+func ErrorStep(msg string) *errorStep {
+	return &errorStep{message: msg}
+}
+
+// Executes prompt step
+func (e *errorStep) Exec(context.Context, *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
+	return nil, fmt.Errorf(e.message)
+}

automation/types/error_handler.go

@@ -7,7 +7,7 @@ import (
 
 type (
 	errorHandlerStep struct {
-		identifiableStep
+		wfexec.StepIdentifier
 		handler wfexec.Step
 	}
 )

automation/types/expr.go

@@ -44,7 +44,7 @@ type (
 
 	// WorkflowStepExpression is created from WorkflowStep with kind=expressions
 	expressionsStep struct {
-		identifiableStep
+		wfexec.StepIdentifier
 		Set ExprSet
 	}
 )

automation/types/function.go

@@ -33,14 +33,14 @@ type (
 	}
 
 	functionStep struct {
-		identifiableStep
+		wfexec.StepIdentifier
 		def       *Function
 		arguments ExprSet
 		results   ExprSet
 	}
 
 	iteratorStep struct {
-		identifiableStep
+		wfexec.StepIdentifier
 		def       *Function
 		arguments ExprSet
 		results   ExprSet

automation/types/step.go

@@ -1,8 +0,0 @@
-package types
-
-type (
-	identifiableStep struct{ id uint64 }
-)
-
-func (i *identifiableStep) ID() uint64      { return i.id }
-func (i *identifiableStep) SetID(id uint64) { i.id = id }

automation/types/workflow.go

@@ -120,7 +120,8 @@ const (
 	WorkflowStepKindGateway     WorkflowStepKind = "gateway"       // ref = join|fork|excl|incl
 	WorkflowStepKindFunction    WorkflowStepKind = "function"      // ref = <function ref>
 	WorkflowStepKindIterator    WorkflowStepKind = "iterator"      // ref = <iterator function ref>
-	WorkflowStepKindMessage     WorkflowStepKind = "message"       // ref = error|warning|info, ...
+	WorkflowStepKindError       WorkflowStepKind = "error"         // no ref
+	WorkflowStepKindTermination WorkflowStepKind = "termination"   // no ref
 	WorkflowStepKindPrompt      WorkflowStepKind = "prompt"        // ref = <client function>
 	WorkflowStepKindErrHandler  WorkflowStepKind = "error-handler" // no ref
 	WorkflowStepKindVisual      WorkflowStepKind = "visual"        // ref = <*>

pkg/wfexec/gateways.go

@@ -29,7 +29,7 @@ func NewGatewayPath(s Step, t pathTester) (gwp *GatewayPath, err error) {
 // joinGateway handles merging/joining of multiple paths into
 // a single path forward
 type joinGateway struct {
-	stepIdentifier
+	StepIdentifier
 	paths  Steps
 	scopes map[Step]*expr.Vars
 	l      sync.Mutex
@@ -75,7 +75,7 @@ func (gw *joinGateway) Exec(_ context.Context, r *ExecRequest) (ExecResponse, er
 
 // forkGateway handles forking to multiple paths
 type forkGateway struct {
-	stepIdentifier
+	StepIdentifier
 }
 
 // ForkGateway fn initializes fork gateway
@@ -93,7 +93,7 @@ func (gw forkGateway) Exec(context.Context, *ExecRequest) (ExecResponse, error)
 
 // inclGateway is an inclusive gateway that can return one or more paths
 type inclGateway struct {
-	stepIdentifier
+	StepIdentifier
 	paths []*GatewayPath
 }
 
@@ -134,7 +134,7 @@ func (gw inclGateway) Exec(ctx context.Context, r *ExecRequest) (ExecResponse, e
 
 // exclGateway is an exclusive gateway that can return exactly one path
 type exclGateway struct {
-	stepIdentifier
+	StepIdentifier
 	paths []*GatewayPath
 }
 

pkg/wfexec/graph.go

@@ -5,13 +5,6 @@ import (
 )
 
 type (
-	Steps []Step
-	Step  interface {
-		ID() uint64
-		SetID(uint64)
-		Exec(context.Context, *ExecRequest) (ExecResponse, error)
-	}
-
 	// list of Graph steps with relations
 	Graph struct {
 		steps    []Step
@@ -19,8 +12,6 @@ type (
 		parents  map[Step][]Step
 		index    map[uint64]Step
 	}
-
-	stepIdentifier struct{ id uint64 }
 )
 
 func NewGraph() *Graph {
@@ -34,9 +25,6 @@ func NewGraph() *Graph {
 	return wf
 }
 
-func (i *stepIdentifier) ID() uint64      { return i.id }
-func (i *stepIdentifier) SetID(id uint64) { i.id = id }
-
 func (g *Graph) AddStep(s Step, cc ...Step) {
 	g.steps = append(g.steps, s)
 

pkg/wfexec/graph_test.go

@@ -8,7 +8,7 @@ import (
 
 type (
 	wfTestStep struct {
-		stepIdentifier
+		StepIdentifier
 		name string
 	}
 )

pkg/wfexec/response.go

@@ -11,6 +11,8 @@ type (
 	errHandler struct {
 		handler Step
 	}
+
+	termination struct{}
 )
 
 func DelayExecution(until time.Time) *suspended {
@@ -25,6 +27,10 @@ func ErrorHandler(h Step) *errHandler {
 	return &errHandler{handler: h}
 }
 
+func Termination() *termination {
+	return &termination{}
+}
+
 type (
 	loopBreak    struct{}
 	loopContinue struct{}

pkg/wfexec/session.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/cortezaproject/corteza-server/pkg/auth"
+	"github.com/cortezaproject/corteza-server/pkg/errors"
 	"github.com/cortezaproject/corteza-server/pkg/expr"
 	"github.com/cortezaproject/corteza-server/pkg/id"
 	"github.com/cortezaproject/corteza-server/pkg/logger"
@@ -335,7 +336,7 @@ func (s *Session) worker(ctx context.Context) {
 			if st == nil {
 				// stop worker
 				s.log.Debug("worker done")
-				break
+				return
 			}
 
 			s.log.Debug("pulled state from queue", zap.Uint64("stateID", st.stateId))
@@ -482,7 +483,7 @@ func (s *Session) exec(ctx context.Context, st *State) {
 					zap.Error(st.err),
 				)
 
-				expr.Assign(scope, "error", expr.Must(expr.NewString(st.err.Error())))
+				_ = expr.Assign(scope, "error", expr.Must(expr.NewString(st.err.Error())))
 
 				// copy error handler & disable it on state to prevent inf. loop
 				// in case of another error in the error-handling branch
@@ -494,8 +495,13 @@ func (s *Session) exec(ctx context.Context, st *State) {
 
 				return
 			} else {
-				log.Error("step execution failed", zap.Error(st.err))
-				s.qErr <- fmt.Errorf("session %d step %d execution failed: %w", s.id, st.step.ID(), st.err)
+				if errors.IsAutomation(st.err) {
+					s.qErr <- st.err
+				} else {
+					log.Error("step execution failed", zap.Error(st.err))
+					s.qErr <- fmt.Errorf("session %d step %d execution failed: %w", s.id, st.step.ID(), st.err)
+				}
+
 				return
 			}
 		}
@@ -572,6 +578,13 @@ func (s *Session) exec(ctx context.Context, st *State) {
 			// it's used mainly for join gateway step that should be called multiple times (one for each parent path)
 			return
 
+		case *termination:
+			// terminate all activities, all suspended tasks and exit right away
+			log.Debug("termination", zap.Int("suspended", len(s.suspended)))
+			s.suspended = nil
+			s.qState <- FinalState(s, scope)
+			return
+
 		case *suspended:
 			// suspend execution because of delay or pending user input
 			// either way, it breaks execution loop for the current path
@@ -586,7 +599,6 @@ func (s *Session) exec(ctx context.Context, st *State) {
 				log.Debug("suspended, temporal", zap.Timep("at", result.resumeAt))
 			} else {
 				log.Debug("suspended, prompt")
-
 			}
 
 			result.state = st

pkg/wfexec/session_test.go

@@ -11,13 +11,13 @@ import (
 
 type (
 	sesTestStep struct {
-		stepIdentifier
+		StepIdentifier
 		name string
 		exec func(context.Context, *ExecRequest) (ExecResponse, error)
 	}
 
 	sesTestTemporal struct {
-		stepIdentifier
+		StepIdentifier
 		delay time.Duration
 		until time.Time
 	}

pkg/wfexec/step.go

@@ -0,0 +1,32 @@
+package wfexec
+
+import "context"
+
+type (
+	Steps []Step
+	Step  interface {
+		ID() uint64
+		SetID(uint64)
+		Exec(context.Context, *ExecRequest) (ExecResponse, error)
+	}
+
+	StepIdentifier struct{ id uint64 }
+
+	execFn func(context.Context, *ExecRequest) (ExecResponse, error)
+
+	genericStep struct {
+		StepIdentifier
+		fn execFn
+	}
+)
+
+func (i *StepIdentifier) ID() uint64      { return i.id }
+func (i *StepIdentifier) SetID(id uint64) { i.id = id }
+
+func NewGenericStep(fn execFn) *genericStep {
+	return &genericStep{fn: fn}
+}
+
+func (g *genericStep) Exec(ctx context.Context, r *ExecRequest) (ExecResponse, error) {
+	return g.fn(ctx, r)
+}