feat(automation): implement completors, add completors to orchestrator

[ramfox]

4. Completors

tests

package automation

// TestCompletor struct
// TestHook struct

func TestCompletors(t *testing.T) {
    // create bus
    // handle events specific to the test completor
    // create orchestrator w/ TestCompletor
    // create workflow w/ hook & add to store
    // run o.handleTrigger() on workflow
    // expect test completor event to have fired
}

define Completor interface

package hook

type Completor interface {
    ConstructHook(m map[string]interface{}) (*workflow.Hook, error)
    Type() Type
    Start(ctx context.Context) error
    Stop() error
    Publish() error
}

add to orchestrator

// automation/orchestrator.go
package automation

type Options struct {
    ...
    completors []*hook.Completor
}

func OptCompletors(l []*hook.Completor) Option {
    return func(opt *Options) error {
        o.completors = c
        return nil
    }
}

var (
    ...
    // ErrNoHookType indicates that no "type" information was given in the hook option map
    ErrNoHookType = fmt.Errorf("")
    // ErrHookTypeNotFound indicates that the given hook "type" given in the hook option map
    // is not a type we support in the orchestrator
    ErrHookTypeNotFound = fmt.Errorf("")
)

type Orchestrator struct {
    ...
    completors []*hook.Completors
}

func (o *Orchestrator) CreateWorkflow(did dataset.ID, pid peer.ID, triggerOpts []map[string]interface{}, hookOpts []map[string]interface{}) (*workflow.State, error) {
    ...
    // iterate over hook Opts
    hooks := []*workflow.Hook
    for _, opt := range hookOpts {
        type, ok := opt["type"]
        if !ok { return nil, ErrNoHookType }
        completor, err := o.completorForHookType(type)
        if err != nil { return nil, err }    
        h, err := completor.ConstructHook(opt)
        hooks := append(hooks, h)
    }
    o.workflows.Create(did, pid, triggers, hooks)    
}

func (o *Orchestrator) handleTrigger() {
    // when event comes in
    // fetch workflow
    // create run id
    // call SaveFunc, passing in run id
    // look at output & derive exit code
    // iterate over completors in workflow
    // for any that match what we can currently do in the orchestrator (range o.completors, completor.Type() match)
    // emit event calling for that completor w/ all necessary completor info.
}

[ramfox]

Add HookSource interface to hook and ensure workflow.Store satisfies it ala this comment: https://github.com/qri-io/qri/pull/1797#issuecomment-850549868