func (s *Orchestrator) filterMatch(f *process.Process_Node_Filter, wf *process.Process, n *process.Process_Node, exec *execution.Execution, data *types.Struct) bool {
	log := s.logger.With(keyvals(wf, n, exec, nil, data)...)
	for _, condition := range f.Conditions {
		resolvedData, err := s.resolveRef(wf, exec, n.Key, data, condition.Ref)
		if err != nil {
			log.Error(err.Error())
			return false
		}
		match, err := condition.Match(resolvedData)
		if err != nil {
			log.Error(err.Error())
		}
		if !match {
			return false
		}
	}
	return true
}
filterMatch returns true if the data match the current list of filters.
package orchestrator
import (
	"context"
	"encoding/json"
	"fmt"
	"math/rand"
	sdk "github.com/cosmos/cosmos-sdk/types"
	"github.com/mesg-foundation/engine/cosmos"
	"github.com/mesg-foundation/engine/event"
	"github.com/mesg-foundation/engine/event/publisher"
	"github.com/mesg-foundation/engine/execution"
	"github.com/mesg-foundation/engine/ext/xrand"
	"github.com/mesg-foundation/engine/ext/xstrings"
	"github.com/mesg-foundation/engine/hash"
	"github.com/mesg-foundation/engine/process"
	"github.com/mesg-foundation/engine/protobuf/types"
	"github.com/mesg-foundation/engine/runner"
	executionmodule "github.com/mesg-foundation/engine/x/execution"
	processmodule "github.com/mesg-foundation/engine/x/process"
	runnermodule "github.com/mesg-foundation/engine/x/runner"
	tmlog "github.com/tendermint/tendermint/libs/log"
)
func init() {
	xrand.SeedInit()
}
type Orchestrator struct {
	rpc             *cosmos.RPC
	ep              *publisher.EventPublisher
	eventStream     *event.Listener
	executionStream chan *execution.Execution
	stopC           chan bool
	logger          tmlog.Logger
}
func New(rpc *cosmos.RPC, ep *publisher.EventPublisher, logger tmlog.Logger) *Orchestrator {
	return &Orchestrator{
		rpc:    rpc,
		ep:     ep,
		stopC:  make(chan bool),
		logger: logger.With("module", "orchestrator"),
	}
}
func (s *Orchestrator) Start() error {
	if s.eventStream != nil || s.executionStream != nil {
		return fmt.Errorf("process orchestrator already running")
	}
	s.eventStream = s.ep.GetStream(nil)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	if err := s.startExecutionStream(ctx); err != nil {
		return err
	}
	for {
		select {
		case event := <-s.eventStream.C:
			go s.execute(s.eventFilter(event), nil, event, event.Data)
		case execution := <-s.executionStream:
			go s.execute(s.resultFilter(execution), execution, nil, execution.Outputs)
			go s.execute(s.dependencyFilter(execution), execution, nil, execution.Outputs)
		case <-s.stopC:
			return nil
		}
	}
}
func (s *Orchestrator) Stop() {
	s.stopC <- true
	s.eventStream.Close()
}
func (s *Orchestrator) eventFilter(event *event.Event) func(wf *process.Process, node *process.Process_Node) (bool, error) {
	return func(wf *process.Process, node *process.Process_Node) (bool, error) {
		if e := node.GetEvent(); e != nil {
			return e.InstanceHash.Equal(event.InstanceHash) && e.EventKey == event.Key, nil
		}
		return false, nil
	}
}
func (s *Orchestrator) resultFilter(exec *execution.Execution) func(wf *process.Process, node *process.Process_Node) (bool, error) {
	return func(wf *process.Process, node *process.Process_Node) (bool, error) {
		if result := node.GetResult(); result != nil {
			return result.InstanceHash.Equal(exec.InstanceHash) && result.TaskKey == exec.TaskKey, nil
		}
		return false, nil
	}
}
func (s *Orchestrator) dependencyFilter(exec *execution.Execution) func(wf *process.Process, node *process.Process_Node) (bool, error) {
	return func(wf *process.Process, node *process.Process_Node) (bool, error) {
		if !exec.ProcessHash.Equal(wf.Hash) {
			return false, nil
		}
		parents := wf.ParentKeys(node.Key)
		if len(parents) == 0 {
			return false, nil
		}
		if len(parents) > 1 {
			return false, fmt.Errorf("multi parents not supported")
		}
		return parents[0] == exec.NodeKey, nil
	}
}
func (s *Orchestrator) findNodes(wf *process.Process, filter func(wf *process.Process, n *process.Process_Node) (bool, error)) []*process.Process_Node {
	return wf.FindNodes(func(n *process.Process_Node) bool {
		res, err := filter(wf, n)
		if err != nil {
			s.logger.With(keyvals(wf, n, nil, nil, nil)...).Error(err.Error())
		}
		return res
	})
}
func (s *Orchestrator) execute(filter func(wf *process.Process, node *process.Process_Node) (bool, error), exec *execution.Execution, event *event.Event, data *types.Struct) {
	var processes []*process.Process
	route := fmt.Sprintf("custom/%s/%s", processmodule.QuerierRoute, processmodule.QueryList)
	if err := s.rpc.QueryJSON(route, nil, &processes); err != nil {
		s.logger.Error(err.Error())
		return
	}
	for _, wf := range processes {
		for _, node := range s.findNodes(wf, filter) {
			if err := s.executeNode(wf, node, exec, event, data); err != nil {
				s.logger.With(keyvals(wf, node, exec, event, data)...).Error(err.Error())
			}
		}
	}
}
func (s *Orchestrator) executeNode(wf *process.Process, n *process.Process_Node, exec *execution.Execution, event *event.Event, data *types.Struct) error {
	log := s.logger.With(keyvals(wf, n, exec, event, data)...)
	switch x := n.Type.(type) {
	case *process.Process_Node_Task_:
		createdExecHash, err := s.processTask(n, x.Task, wf, exec, event, data)
		if err != nil {
			return err
		}
		log.With("createdExecHash", createdExecHash.String()).Info("execution created")
		return nil  
	case *process.Process_Node_Map_:
		var err error
		data, err = s.processMap(n.Key, x.Map.Outputs, wf, exec, data)
		if err != nil {
			return err
		}
		if result, err := json.Marshal(data); err == nil {
			log = log.With("output", string(result))
		}
	case *process.Process_Node_Filter_:
		if result, err := json.Marshal(x.Filter); err == nil {
			log = log.With("filter", string(result))
		}
		if !s.filterMatch(x.Filter, wf, n, exec, data) {
			log.Info("filter does not match data")
			return nil  
		}
	}
	for _, childrenID := range wf.ChildrenKeys(n.Key) {
		log = log.With("to", childrenID)
		children, err := wf.FindNode(childrenID)
		if err != nil {
			log.Error(err.Error())
			continue
		}
		log.Info("executed process transition")
		if err := s.executeNode(wf, children, exec, event, data); err != nil {
			s.logger.With(keyvals(wf, children, exec, event, data)...).Error(err.Error())
		}
	}
	return nil
}