Add run-as validation for subset of triggers

automation/service/trigger.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"github.com/cortezaproject/corteza-server/automation/types"
+	cmpEvent "github.com/cortezaproject/corteza-server/compose/service/event"
 	"github.com/cortezaproject/corteza-server/pkg/actionlog"
 	"github.com/cortezaproject/corteza-server/pkg/auth"
 	"github.com/cortezaproject/corteza-server/pkg/errors"
@@ -17,6 +18,7 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/options"
 	"github.com/cortezaproject/corteza-server/pkg/wfexec"
 	"github.com/cortezaproject/corteza-server/store"
+	sysEvent "github.com/cortezaproject/corteza-server/system/service/event"
 	"go.uber.org/zap"
 )
 
@@ -460,10 +462,20 @@ func (svc *trigger) registerWorkflow(ctx context.Context, wf *types.Workflow, tt
 		return nil
 	}
 
+	if !wf.Enabled || len(wf.Issues) > 0 {
+		// do not even try to register disabled
+		// workflows or workflows with issues
+		return nil
+	}
+
 	if len(types.TriggerSet(tt).FilterByWorkflowID(wf.ID)) < len(tt) {
 		return fmt.Errorf("all triggers must reference the given workflow")
 	}
 
+	if wis := validateWorkflowTriggers(wf, tt...); len(wis) > 0 {
+		return wis
+	}
+
 	if wf.RunAs > 0 {
 		if runAs, err = DefaultUser.FindByAny(ctx, wf.RunAs); err != nil {
 			return fmt.Errorf("failed to load run-as user %d: %w", wf.RunAs, err)
@@ -629,3 +641,43 @@ func toLabeledTriggers(set []*types.Trigger) []label.LabeledResource {
 
 	return ll
 }
+
+// Checks if triggers are compatible with the workflow
+//
+// It ignores disabled triggers and does not care if triggers are in fact bond to the
+// given workflow
+func validateWorkflowTriggers(wf *types.Workflow, tt ...*types.Trigger) (wis types.WorkflowIssueSet) {
+	var (
+		// @todo find a better way how to flag events type that require
+		//       run-as param to be set
+		//       Possible solution: flag in definition that generates static
+		//       list w/o the need  of cross-component imports
+		requireRunAs = []eventbus.Event{
+			sysEvent.SinkOnRequest(nil, nil),
+			sysEvent.QueueOnMessage(nil),
+			sysEvent.SystemOnInterval(),
+			sysEvent.SystemOnTimestamp(),
+			cmpEvent.ComposeOnInterval(),
+			cmpEvent.ComposeOnTimestamp(),
+		}
+	)
+
+	for i, t := range tt {
+		if !t.Enabled {
+			continue
+		}
+
+		for _, ev := range requireRunAs {
+			if t.ResourceType == ev.ResourceType() && t.EventType == ev.EventType() {
+				if wf.RunAs == 0 {
+					wis = wis.Append(
+						errors.InvalidData("%s for %s requires run-as to be set", ev.ResourceType(), ev.EventType()),
+						map[string]int{"trigger": i},
+					)
+				}
+			}
+		}
+	}
+
+	return
+}

automation/service/workflow.go

@@ -210,7 +210,9 @@ func (svc *workflow) Create(ctx context.Context, new *types.Workflow) (wf *types
 			CreatedBy: cUser,
 		}
 
-		if _, wf.Issues = Convert(svc, wf); len(wf.Issues) == 0 {
+		if wf.Issues, err = svc.validateWorkflow(ctx, wf); err != nil {
+			return
+		} else if len(wf.Issues) == 0 {
 			if err = svc.triggers.registerWorkflows(ctx, wf); err != nil {
 				return err
 			}
@@ -249,7 +251,9 @@ func (svc *workflow) DeleteByID(ctx context.Context, workflowID uint64) error {
 			return workflowUnchanged, err
 		}
 
-		if _, res.Issues = Convert(svc, res); len(res.Issues) == 0 {
+		if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil {
+			return workflowUnchanged, err
+		} else if len(res.Issues) == 0 {
 			if err := svc.triggers.registerWorkflows(ctx, res); err != nil {
 				return workflowUnchanged, err
 			}
@@ -267,7 +271,9 @@ func (svc *workflow) UndeleteByID(ctx context.Context, workflowID uint64) error
 			return workflowUnchanged, err
 		}
 
-		if _, res.Issues = Convert(svc, res); len(res.Issues) == 0 {
+		if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil {
+			return workflowUnchanged, err
+		} else if len(res.Issues) == 0 {
 			if err := svc.triggers.registerWorkflows(ctx, res); err != nil {
 				return workflowUnchanged, err
 			}
@@ -313,7 +319,9 @@ func (svc *workflow) updater(ctx context.Context, workflowID uint64, action func
 			return err
 		}
 
-		if _, res.Issues = Convert(svc, res); len(res.Issues) == 0 {
+		if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil {
+			return
+		} else if len(res.Issues) == 0 {
 			if err = svc.triggers.registerWorkflows(ctx, res); err != nil {
 				return err
 			}
@@ -603,6 +611,25 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl
 	return results, stacktrace, svc.recordAction(ctx, wap, WorkflowActionExecute, err)
 }
 
+// validates workflow by trying to convert it to graph and checking assigned triggers
+func (svc *workflow) validateWorkflow(ctx context.Context, wf *types.Workflow) (types.WorkflowIssueSet, error) {
+	_, wis := Convert(svc, wf)
+
+	tt, _, err := store.SearchAutomationTriggers(ctx, svc.store, types.TriggerFilter{
+		WorkflowID: types.WorkflowSet{wf}.IDs(),
+		Deleted:    filter.StateInclusive,
+		Disabled:   filter.StateExcluded,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	wis = append(wis, validateWorkflowTriggers(wf, tt...)...)
+
+	return wis, nil
+}
+
 func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger, wf *types.Workflow, g *wfexec.Graph, _runAs intAuth.Identifiable) eventbus.HandlerFn {
 	return func(ctx context.Context, ev eventbus.Event) (err error) {
 		var (