-
Notifications
You must be signed in to change notification settings - Fork 6
/
trigger.go
89 lines (71 loc) · 2.31 KB
/
trigger.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
package workflow
import (
"context"
"errors"
"fmt"
"github.com/google/uuid"
)
func (w *Workflow[Type, Status]) Trigger(ctx context.Context, foreignID string, startingStatus Status, opts ...TriggerOption[Type, Status]) (runID string, err error) {
return trigger(ctx, w, w.recordStore.Latest, foreignID, startingStatus, opts...)
}
func trigger[Type any, Status StatusType](ctx context.Context, w *Workflow[Type, Status], lookup latestLookup, foreignID string, startingStatus Status, opts ...TriggerOption[Type, Status]) (runID string, err error) {
if !w.calledRun {
return "", fmt.Errorf("trigger failed: workflow is not running")
}
if !w.statusGraph.IsValid(int(startingStatus)) {
w.logger.maybeDebug(w.ctx, fmt.Sprintf("ensure %v is configured for workflow: %v", startingStatus, w.Name), map[string]string{})
return "", fmt.Errorf("trigger failed: status provided is not configured for workflow: %s", startingStatus)
}
var o triggerOpts[Type, Status]
for _, fn := range opts {
fn(&o)
}
var t Type
if o.initialValue != nil {
t = *o.initialValue
}
object, err := Marshal(&t)
if err != nil {
return "", err
}
lastRecord, err := lookup(ctx, w.Name, foreignID)
if errors.Is(err, ErrRecordNotFound) {
lastRecord = &Record{}
} else if err != nil {
return "", err
}
// Check that the last run has completed before triggering a new run.
if lastRecord.RunState.Valid() && !lastRecord.RunState.Finished() {
// Cannot trigger a new run for this foreignID if there is a workflow in progress.
return "", ErrWorkflowInProgress
}
uid, err := uuid.NewUUID()
if err != nil {
return "", err
}
runID = uid.String()
wr := &Record{
WorkflowName: w.Name,
ForeignID: foreignID,
RunID: runID,
RunState: RunStateInitiated,
Status: int(startingStatus),
Object: object,
CreatedAt: w.clock.Now(),
UpdatedAt: w.clock.Now(),
}
err = updateWireRecord(ctx, w.recordStore.Store, wr, RunStateUnknown)
if err != nil {
return "", err
}
return runID, nil
}
type triggerOpts[Type any, Status StatusType] struct {
initialValue *Type
}
type TriggerOption[Type any, Status StatusType] func(o *triggerOpts[Type, Status])
func WithInitialValue[Type any, Status StatusType](t *Type) TriggerOption[Type, Status] {
return func(o *triggerOpts[Type, Status]) {
o.initialValue = t
}
}