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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419package workflow
import (
"context"
"fmt"
"log/slog"
"reflect"
"runtime/debug"
"slices"
"strings"
"sync"
"dario.cat/mergo"
"golang.org/x/sync/errgroup"
)
// Step is the basic unit of work in a workflow. It is an interface with a
// single method, Run, that takes a context and a generic request type T and
// returns a response of the same type T and an error.
type Step[T any] interface {
Run(context.Context, *T) (*T, error)
fmt.Stringer
}
// Name returns the name of a step.
func Name[T any](s Step[T]) string {
t := reflect.TypeOf(s)
if t.Kind() == reflect.Pointer {
t = t.Elem()
}
var z [0]T // zero alloc
return strings.Replace(t.Name(), reflect.TypeOf(z).Elem().PkgPath()+".", "", 1)
}
// Pipeline is a step that executes a series of other steps in sequential order.
// It can also have middleware that is applied to each step in the pipeline.
type Pipeline[T any] struct {
Steps []Step[T]
Middleware []Middleware[T]
}
// Run executes the pipeline.
func (p *Pipeline[T]) Run(ctx context.Context, req *T) (*T, error) {
resp := req
var err error
for i := range p.Steps {
for _, m := range slices.Backward(p.Middleware) {
p.Steps[i] = m(p.Steps[i])
}
resp, err = p.Steps[i].Run(ctx, req)
if err != nil {
return nil, err
}
req = resp
}
return resp, nil
}
func (p *Pipeline[T]) String() string {
if len(p.Steps) == 0 {
return Name(p)
}
var buf strings.Builder
buf.WriteString("\n")
buf.WriteString(Name(p))
for i, step := range p.Steps {
buf.WriteString("\n")
var prefix string
var childPrefix string
if i == len(p.Steps)-1 {
prefix = "└── "
childPrefix = " "
} else {
prefix = "├── "
childPrefix = "│ "
}
buf.WriteString(prefix)
s := step.String()
lines := strings.Split(s, "\n")
buf.WriteString(lines[0])
for _, line := range lines[1:] {
buf.WriteString("\n")
buf.WriteString(childPrefix)
buf.WriteString(line)
}
}
buf.WriteString("\n")
return buf.String()
}
// NewPipeline creates a new pipeline with the given middleware.
func NewPipeline[T any](mid ...Middleware[T]) *Pipeline[T] {
return &Pipeline[T]{
Middleware: mid,
Steps: make([]Step[T], 0),
}
}
// StepFunc is a function type for a unit of work in the workflow.
// It is an adapter to allow the use of ordinary functions as workflow steps.
type StepFunc[T any] func(context.Context, *T) (*T, error)
// Run executes the function that implements the Step interface.
func (f StepFunc[T]) Run(ctx context.Context, res *T) (*T, error) {
return f(ctx, res)
}
// String returns the name of the function.
func (f StepFunc[T]) String() string {
var z T
return fmt.Sprintf("StepFunc[%T]", z)
}
// Middleware
// MidFunc is an adapter to allow the use of ordinary functions as middleware.
type MidFunc[T any] struct {
Name string
Fn func(context.Context, *T) (*T, error)
Next Step[T]
}
// Run executes the function.
func (m *MidFunc[T]) Run(ctx context.Context, req *T) (*T, error) {
return m.Fn(ctx, req)
}
// String returns the name of the function.
func (m *MidFunc[T]) String() string {
s := m.Next.String()
return fmt.Sprintf("%s(%s)", m.Name, s)
}
// Middleware is a function that wraps a step to add functionality, such as
// logging or error handling.
type Middleware[T any] func(s Step[T]) Step[T]
// Selector
// Selector is a function that returns true or false based on the context and the request.
// Selector is a function type used to select steps conditionally in a workflow.
type Selector[T any] func(context.Context, *T) bool
// selector is a step that executes one of two other steps based on the result
// of a selector function.
type selector[T any] struct {
s Selector[T]
ifStep Step[T]
elseStep Step[T]
middleware []Middleware[T]
}
// String returns the name of the selector.
func (s selector[T]) String() string {
var buf strings.Builder
buf.WriteString(Name(&s))
// IF
buf.WriteString("\n")
buf.WriteString("├── IF: ")
if s.ifStep != nil {
ifStr := s.ifStep.String()
lines := strings.Split(ifStr, "\n")
buf.WriteString(lines[0])
for _, line := range lines[1:] {
buf.WriteString("\n")
buf.WriteString("│ ")
buf.WriteString(line)
}
} else {
buf.WriteString("none")
}
// ELSE
buf.WriteString("\n")
buf.WriteString("└── ELSE: ")
if s.elseStep != nil {
elseStr := s.elseStep.String()
lines := strings.Split(elseStr, "\n")
buf.WriteString(lines[0])
for _, line := range lines[1:] {
buf.WriteString("\n")
buf.WriteString(" ")
buf.WriteString(line)
}
} else {
buf.WriteString("none")
}
return buf.String()
}
// Select creates a new selector step.
func Select[T any](mid []Middleware[T], s Selector[T], ifStep, elseStep Step[T]) Step[T] {
return &selector[T]{
s: s,
ifStep: ifStep,
elseStep: elseStep,
middleware: mid,
}
}
// Run executes the selector.
func (s selector[T]) Run(ctx context.Context, r *T) (*T, error) {
var step Step[T]
if s.s(ctx, r) {
step = s.ifStep
} else {
step = s.elseStep
}
if step == nil {
return nil, fmt.Errorf("selector has no step for the selected condition")
}
for _, m := range slices.Backward(s.middleware) {
step = m(step)
}
return step.Run(ctx, r)
}
// Series
// series is a step that executes a list of other steps sequentially.
type series[T any] struct {
Stages []Step[T]
middleware []Middleware[T]
}
// String returns the name of the series.
func (s *series[T]) String() string {
if s == nil {
return "none"
}
if len(s.Stages) == 0 {
return Name(s)
}
var buf strings.Builder
buf.WriteString(Name(s))
for i, stage := range s.Stages {
buf.WriteString("\n")
var prefix string
var childPrefix string
if i == len(s.Stages)-1 {
prefix = "└── "
childPrefix = " "
} else {
prefix = "├── "
childPrefix = "│ "
}
buf.WriteString(prefix)
st := stage.String()
lines := strings.Split(st, "\n")
buf.WriteString(lines[0])
for _, line := range lines[1:] {
buf.WriteString("\n")
buf.WriteString(childPrefix)
buf.WriteString(line)
}
}
return buf.String()
}
// Sequential executes a series of steps in sequential order.
func Sequential[T any](mid []Middleware[T], steps ...Step[T]) *series[T] {
// Series creates a sequential pipeline of steps with optional middleware.
// Returns a private type *series[T].
return &series[T]{
Stages: steps,
middleware: mid,
}
}
// Run executes the series.
func (s *series[T]) Run(ctx context.Context, req *T) (*T, error) {
var err error
resp := req
for i := range s.Stages {
for _, m := range slices.Backward(s.middleware) {
s.Stages[i] = m(s.Stages[i])
}
resp, err = s.Stages[i].Run(ctx, req)
if err != nil {
return resp, err
}
req = resp
}
return resp, nil
}
// Parallel
// parallel is a step that executes a list of other steps in parallel.
type parallel[T any] struct {
merge MergeRequest[T]
Tasks []Step[T]
middleware []Middleware[T]
}
// String returns the name of the parallel step.
func (p *parallel[T]) String() string {
if p == nil {
return "none"
}
if len(p.Tasks) == 0 {
return Name(p)
}
var buf strings.Builder
buf.WriteString(Name(p))
for i, task := range p.Tasks {
buf.WriteString("\n")
var prefix string
var childPrefix string
if i == len(p.Tasks)-1 {
prefix = "└── "
childPrefix = " "
} else {
prefix = "├── "
childPrefix = "│ "
}
buf.WriteString(prefix)
st := task.String()
lines := strings.Split(st, "\n")
buf.WriteString(lines[0])
for _, line := range lines[1:] {
buf.WriteString("\n")
buf.WriteString(childPrefix)
buf.WriteString(line)
}
}
return buf.String()
}
// MergeRequest is a function that merges the results of multiple steps into a
// single result.
type MergeRequest[T any] func(context.Context, *T, ...*T) (*T, error)
// Parallel executes a list of steps in parallel.
// Once all the steps are done, the merge request [MergeRequest] will combine all the results into one struct T.
func Parallel[T any](mid []Middleware[T], merge MergeRequest[T], steps ...Step[T]) *parallel[T] {
return ¶llel[T]{
merge: merge,
Tasks: steps,
middleware: mid,
}
}
// Run executes the parallel step.
func (p *parallel[T]) Run(ctx context.Context, req *T) (*T, error) {
tasks := make([]Step[T], len(p.Tasks))
for i, s := range p.Tasks {
tasks[i] = s
for _, m := range slices.Backward(p.middleware) {
tasks[i] = m(tasks[i])
}
}
g, groupCtx := errgroup.WithContext(ctx)
resps := make([]*T, len(p.Tasks))
mu := sync.Mutex{}
for i := range tasks {
task := tasks[i] // Capture task
g.Go(func() error {
defer CapturePanic(groupCtx)
copyReq := SafeCopy(req)
resp, err := task.Run(groupCtx, copyReq)
if err != nil {
return err
}
mu.Lock()
resps[i] = resp
mu.Unlock()
return nil
})
}
if err := g.Wait(); err != nil {
return nil, err
}
return p.merge(ctx, req, resps...)
}
// MergeTransform is a merge request that merges the results of multiple steps
// into a single result using the mergo library.
func MergeTransform[T any](opts ...func(*mergo.Config)) MergeRequest[T] {
return func(ctx context.Context, res *T, responses ...*T) (*T, error) {
var err error
for _, r := range responses {
select {
case <-ctx.Done():
return nil, fmt.Errorf("aborting: %w", ctx.Err())
default:
err = mergo.Merge(res, r, opts...)
if err != nil {
return nil, err
}
}
}
return res, nil
}
}
// Merge is a merge request that merges the results of multiple steps into a
// single result using the mergo library.
func Merge[T any](ctx context.Context, req *T, responses ...*T) (*T, error) {
return MergeTransform[T]()(ctx, req, responses...)
}
// CapturePanic recovers from a panic and logs the error with stack trace.
func CapturePanic(ctx context.Context) {
if r := recover(); r != nil {
slog.Error("panic recovered",
"error", r,
"context_cancelled", ctx.Err() != nil,
"stack", string(debug.Stack()),
)
}
}