Merge pull request #174 from nyaruka/event_origin

Event validation

Author: rowanseymour

flows/engine/session.go

@@ -119,6 +119,11 @@ func (s *session) Start(trigger flows.Trigger, callerEvents []flows.Event) error
 		return fmt.Errorf("validation failed for flow[uuid=%s]: %v", trigger.Flow().UUID(), err)
 	}
 
+	// check caller events are valid
+	if err := s.validateCallerEvents(callerEvents); err != nil {
+		return err
+	}
+
 	if trigger.Environment() != nil {
 		s.env = trigger.Environment()
 	}
@@ -149,6 +154,11 @@ func (s *session) Resume(callerEvents []flows.Event) error {
 		return fmt.Errorf("validation failed for flow[uuid=%s]: %v", waitingRun.Flow().UUID(), err)
 	}
 
+	// check caller events are valid
+	if err := s.validateCallerEvents(callerEvents); err != nil {
+		return err
+	}
+
 	if err := s.tryToResume(waitingRun, callerEvents); err != nil {
 		// if we got an error, add it to the log and shut everything down
 		for _, run := range s.runs {
@@ -206,11 +216,7 @@ func (s *session) tryToResume(waitingRun flows.FlowRun, callerEvents []flows.Eve
 	s.status = flows.SessionStatusActive
 
 	// off to the races again...
-	if err = s.continueUntilWait(waitingRun, destination, step, []flows.Event{}); err != nil {
-		return err
-	}
-
-	return nil
+	return s.continueUntilWait(waitingRun, destination, step, []flows.Event{})
 }
 
 // finds the next destination in a run that may have been waiting or a parent paused for a child subflow
@@ -443,6 +449,18 @@ func (s *session) pickNodeExit(run flows.FlowRun, node flows.Node, step flows.St
 
 const noDestination = flows.NodeUUID("")
 
+func (s *session) validateCallerEvents(events []flows.Event) error {
+	for _, event := range events {
+		if event.AllowedOrigin()&flows.EventOriginCaller == 0 {
+			return fmt.Errorf("event[type=%s] can't be sent by callers", event.Type())
+		}
+		if err := event.Validate(s.assets); err != nil {
+			return fmt.Errorf("validation failed for event[type=%s]: %s", event.Type(), err)
+		}
+	}
+	return nil
+}
+
 //------------------------------------------------------------------------------------------
 // JSON Encoding / Decoding
 //------------------------------------------------------------------------------------------

flows/events/base.go

@@ -1,6 +1,7 @@
 package events
 
 import (
+	"github.com/nyaruka/goflow/flows"
 	"time"
 )
 
@@ -18,3 +19,25 @@ func (e *BaseEvent) SetCreatedOn(time time.Time) { e.CreatedOn_ = time }
 
 func (e *BaseEvent) FromCaller() bool              { return e.FromCaller_ }
 func (e *BaseEvent) SetFromCaller(fromCaller bool) { e.FromCaller_ = fromCaller }
+
+type callerOnlyEvent struct{}
+
+// AllowedOrigin determines where this event type can originate
+func (e *callerOnlyEvent) AllowedOrigin() flows.EventOrigin { return flows.EventOriginCaller }
+
+type engineOnlyEvent struct{}
+
+// AllowedOrigin determines where this event type can originate
+func (e *engineOnlyEvent) AllowedOrigin() flows.EventOrigin { return flows.EventOriginEngine }
+
+// Validate validates our event is valid and has all the assets it needs. We assume engine generated events are valid.
+func (e *engineOnlyEvent) Validate(assets flows.SessionAssets) error {
+	return nil
+}
+
+type callerOrEngineEvent struct{}
+
+// AllowedOrigin determines where this event type can originate
+func (e *callerOrEngineEvent) AllowedOrigin() flows.EventOrigin {
+	return flows.EventOriginCaller | flows.EventOriginEngine
+}

flows/events/broadcast_created.go

@@ -25,6 +25,8 @@ const TypeBroadcastCreated string = "broadcast_created"
 // @event broadcast_created
 type BroadcastCreatedEvent struct {
 	BaseEvent
+	engineOnlyEvent
+
 	Text         string                    `json:"text"`
 	Attachments  []flows.Attachment        `json:"attachments,omitempty"`
 	QuickReplies []string                  `json:"quick_replies,omitempty"`

flows/events/contact_changed.go

@@ -23,12 +23,19 @@ const TypeContactChanged string = "contact_changed"
 // @event contact_changed
 type ContactChangedEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	Contact json.RawMessage `json:"contact"`
 }
 
 // Type returns the type of this event
 func (e *ContactChangedEvent) Type() string { return TypeContactChanged }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ContactChangedEvent) Validate(assets flows.SessionAssets) error {
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *ContactChangedEvent) Apply(run flows.FlowRun) error {
 	contact, err := flows.ReadContact(run.Session(), e.Contact)

flows/events/contact_channel_changed.go

@@ -21,6 +21,8 @@ const TypeContactChannelChanged string = "contact_channel_changed"
 // @event contact_channel_changed
 type ContactChannelChangedEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	Channel *flows.ChannelReference `json:"channel" validate:"required"`
 }
 
@@ -35,6 +37,12 @@ func NewContactChannelChangedEvent(channel *flows.ChannelReference) *ContactChan
 // Type returns the type of this event
 func (e *ContactChannelChangedEvent) Type() string { return TypeContactChannelChanged }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ContactChannelChangedEvent) Validate(assets flows.SessionAssets) error {
+	_, err := assets.GetChannel(e.Channel.UUID)
+	return err
+}
+
 // Apply applies this event to the given run
 func (e *ContactChannelChangedEvent) Apply(run flows.FlowRun) error {
 	if run.Contact() == nil {

flows/events/contact_field_changed.go

@@ -23,6 +23,8 @@ const TypeContactFieldChanged string = "contact_field_changed"
 // @event contact_field_changed
 type ContactFieldChangedEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	Field *flows.FieldReference `json:"field" validate:"required"`
 	Value string                `json:"value" validate:"required"`
 }
@@ -39,6 +41,12 @@ func NewContactFieldChangedEvent(field *flows.FieldReference, value string) *Con
 // Type returns the type of this event
 func (e *ContactFieldChangedEvent) Type() string { return TypeContactFieldChanged }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ContactFieldChangedEvent) Validate(assets flows.SessionAssets) error {
+	_, err := assets.GetField(e.Field.Key)
+	return err
+}
+
 // Apply applies this event to the given run
 func (e *ContactFieldChangedEvent) Apply(run flows.FlowRun) error {
 	if run.Contact() == nil {

flows/events/contact_groups_added.go

@@ -22,6 +22,8 @@ const TypeContactGroupsAdded string = "contact_groups_added"
 // @event contact_groups_added
 type ContactGroupsAddedEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	Groups []*flows.GroupReference `json:"groups" validate:"required,min=1,dive"`
 }
 
@@ -36,6 +38,17 @@ func NewContactGroupsAddedEvent(groups []*flows.GroupReference) *ContactGroupsAd
 // Type returns the type of this event
 func (e *ContactGroupsAddedEvent) Type() string { return TypeContactGroupsAdded }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ContactGroupsAddedEvent) Validate(assets flows.SessionAssets) error {
+	for _, group := range e.Groups {
+		_, err := assets.GetGroup(group.UUID)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *ContactGroupsAddedEvent) Apply(run flows.FlowRun) error {
 	if run.Contact() == nil {

flows/events/contact_groups_removed.go

@@ -22,8 +22,10 @@ const TypeContactGroupsRemoved string = "contact_groups_removed"
 //
 // @event contact_groups_removed
 type ContactGroupsRemovedEvent struct {
-	Groups []*flows.GroupReference `json:"groups" validate:"required,min=1,dive"`
 	BaseEvent
+	callerOrEngineEvent
+
+	Groups []*flows.GroupReference `json:"groups" validate:"required,min=1,dive"`
 }
 
 // NewContactGroupsRemovedEvent returns a new remove from group event
@@ -37,6 +39,17 @@ func NewContactGroupsRemovedEvent(groups []*flows.GroupReference) *ContactGroups
 // Type returns the type of this event
 func (e *ContactGroupsRemovedEvent) Type() string { return TypeContactGroupsRemoved }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ContactGroupsRemovedEvent) Validate(assets flows.SessionAssets) error {
+	for _, group := range e.Groups {
+		_, err := assets.GetGroup(group.UUID)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *ContactGroupsRemovedEvent) Apply(run flows.FlowRun) error {
 	if run.Contact() == nil {

flows/events/contact_property_changed.go

@@ -24,6 +24,8 @@ const TypeContactPropertyChanged string = "contact_property_changed"
 // @event contact_property_changed
 type ContactPropertyChangedEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	Property string `json:"property" validate:"required,eq=name|eq=language"`
 	Value    string `json:"value"`
 }
@@ -40,6 +42,11 @@ func NewContactPropertyChangedEvent(property string, value string) *ContactPrope
 // Type returns the type of this event
 func (e *ContactPropertyChangedEvent) Type() string { return TypeContactPropertyChanged }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ContactPropertyChangedEvent) Validate(assets flows.SessionAssets) error {
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *ContactPropertyChangedEvent) Apply(run flows.FlowRun) error {
 	if run.Contact() == nil {

flows/events/contact_urn_added.go

@@ -23,6 +23,8 @@ const TypeContactURNAdded string = "contact_urn_added"
 // @event contact_urn_added
 type ContactURNAddedEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	URN urns.URN `json:"urn" validate:"urn"`
 }
 
@@ -34,6 +36,11 @@ func NewURNAddedEvent(urn urns.URN) *ContactURNAddedEvent {
 // Type returns the type of this event
 func (e *ContactURNAddedEvent) Type() string { return TypeContactURNAdded }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ContactURNAddedEvent) Validate(assets flows.SessionAssets) error {
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *ContactURNAddedEvent) Apply(run flows.FlowRun) error {
 	if run.Contact() == nil {

flows/events/email_created.go

@@ -20,6 +20,8 @@ const TypeEmailCreated string = "email_created"
 // @event email_created
 type EmailCreatedEvent struct {
 	BaseEvent
+	engineOnlyEvent
+
 	Addresses []string `json:"addresses" validate:"required,min=1"`
 	Subject   string   `json:"subject" validate:"required"`
 	Body      string   `json:"body"`
@@ -38,6 +40,11 @@ func NewEmailCreatedEvent(addresses []string, subject string, body string) *Emai
 // Type returns the type of this event
 func (e *EmailCreatedEvent) Type() string { return TypeEmailCreated }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *EmailCreatedEvent) Validate(assets flows.SessionAssets) error {
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *EmailCreatedEvent) Apply(run flows.FlowRun) error {
 	return nil

flows/events/environment_changed.go

@@ -28,12 +28,19 @@ const TypeEnvironmentChanged string = "environment_changed"
 // @event environment_changed
 type EnvironmentChangedEvent struct {
 	BaseEvent
+	callerOnlyEvent
+
 	Environment json.RawMessage `json:"environment"`
 }
 
 // Type returns the type of this event
 func (e *EnvironmentChangedEvent) Type() string { return TypeEnvironmentChanged }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *EnvironmentChangedEvent) Validate(assets flows.SessionAssets) error {
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *EnvironmentChangedEvent) Apply(run flows.FlowRun) error {
 	env, err := utils.ReadEnvironment(e.Environment)

flows/events/error.go

@@ -19,6 +19,8 @@ const TypeError string = "error"
 // @event error
 type ErrorEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	Text  string `json:"text" validate:"required"`
 	Fatal bool   `json:"fatal"`
 }
@@ -42,6 +44,11 @@ func NewFatalErrorEvent(err error) *ErrorEvent {
 // Type returns the type of this event
 func (e *ErrorEvent) Type() string { return TypeError }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *ErrorEvent) Validate(assets flows.SessionAssets) error {
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *ErrorEvent) Apply(run flows.FlowRun) error {
 	if e.Fatal {

flows/events/flow_triggered.go

@@ -21,6 +21,8 @@ const TypeFlowTriggered string = "flow_triggered"
 // @event flow_triggered
 type FlowTriggeredEvent struct {
 	BaseEvent
+	engineOnlyEvent
+
 	Flow          *flows.FlowReference `json:"flow" validate:"required"`
 	ParentRunUUID flows.RunUUID        `json:"parent_run_uuid" validate:"omitempty,uuid4"`
 }

flows/events/input_labels_added.go

@@ -19,6 +19,8 @@ const TypeInputLabelsAdded string = "input_labels_added"
 // @event input_labels_added
 type InputLabelsAddedEvent struct {
 	BaseEvent
+	callerOrEngineEvent
+
 	InputUUID flows.InputUUID         `json:"input_uuid" validate:"required,uuid4"`
 	Labels    []*flows.LabelReference `json:"labels" validate:"required,min=1,dive"`
 }
@@ -35,6 +37,17 @@ func NewInputLabelsAddedEvent(inputUUID flows.InputUUID, labels []*flows.LabelRe
 // Type returns the type of this event
 func (e *InputLabelsAddedEvent) Type() string { return TypeInputLabelsAdded }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *InputLabelsAddedEvent) Validate(assets flows.SessionAssets) error {
+	for _, label := range e.Labels {
+		_, err := assets.GetLabel(label.UUID)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *InputLabelsAddedEvent) Apply(run flows.FlowRun) error {
 	return nil

flows/events/msg_created.go

@@ -26,6 +26,8 @@ const TypeMsgCreated string = "msg_created"
 // @event msg_created
 type MsgCreatedEvent struct {
 	BaseEvent
+	engineOnlyEvent
+
 	Msg flows.MsgOut `json:"msg" validate:"required,dive"`
 }
 

flows/events/msg_received.go

@@ -28,6 +28,8 @@ const TypeMsgReceived string = "msg_received"
 // @event msg_received
 type MsgReceivedEvent struct {
 	BaseEvent
+	callerOnlyEvent
+
 	Msg flows.MsgIn `json:"msg" validate:"required,dive"`
 }
 
@@ -42,6 +44,15 @@ func NewMsgReceivedEvent(msg *flows.MsgIn) *MsgReceivedEvent {
 // Type returns the type of this event
 func (e *MsgReceivedEvent) Type() string { return TypeMsgReceived }
 
+// Validate validates our event is valid and has all the assets it needs
+func (e *MsgReceivedEvent) Validate(assets flows.SessionAssets) error {
+	if e.Msg.Channel() != nil {
+		_, err := assets.GetChannel(e.Msg.Channel().UUID)
+		return err
+	}
+	return nil
+}
+
 // Apply applies this event to the given run
 func (e *MsgReceivedEvent) Apply(run flows.FlowRun) error {
 	var channel flows.Channel