302 lines
9.2 KiB
Go
302 lines
9.2 KiB
Go
package flow
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"sort"
|
|
"sync"
|
|
)
|
|
|
|
// Phase identifies when a hook fires relative to its step's action.
|
|
type Phase string
|
|
|
|
const (
|
|
PhaseBefore Phase = "before"
|
|
PhaseAfter Phase = "after"
|
|
PhaseOnError Phase = "onError"
|
|
)
|
|
|
|
// State is the mutable bag shared across a single flow run. Actions and hooks
|
|
// read and write Vars to pass data down the chain (e.g. an authorization
|
|
// reference produced by one step and consumed by the next). ID is the domain
|
|
// entity the flow operates on (an order id, for the order flows).
|
|
type State struct {
|
|
ID uint64
|
|
Vars map[string]any
|
|
Logger *slog.Logger
|
|
}
|
|
|
|
// NewState returns an empty state for entity id.
|
|
func NewState(id uint64, logger *slog.Logger) *State {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &State{ID: id, Vars: map[string]any{}, Logger: logger}
|
|
}
|
|
|
|
// Action performs a step's work. params is the step's opaque JSON config.
|
|
type Action func(ctx context.Context, st *State, params json.RawMessage) error
|
|
|
|
// Predicate gates a step (the step runs only when it returns true).
|
|
type Predicate func(ctx context.Context, st *State) (bool, error)
|
|
|
|
// HookInfo describes the step and phase a hook is firing for.
|
|
type HookInfo struct {
|
|
Step string
|
|
Phase Phase
|
|
// Err is the action error when Phase == PhaseOnError, else nil.
|
|
Err error
|
|
}
|
|
|
|
// Hook reacts around a step. Hook errors never abort the flow — they are logged
|
|
// — so observability/notification hooks can't break the business transaction.
|
|
type Hook func(ctx context.Context, st *State, info HookInfo, params json.RawMessage) error
|
|
|
|
// Registry holds the named actions, predicates and hooks a flow can reference.
|
|
type Registry struct {
|
|
mu sync.RWMutex
|
|
actions map[string]Action
|
|
predicates map[string]Predicate
|
|
hooks map[string]Hook
|
|
}
|
|
|
|
// NewRegistry returns an empty registry.
|
|
func NewRegistry() *Registry {
|
|
return &Registry{
|
|
actions: map[string]Action{},
|
|
predicates: map[string]Predicate{},
|
|
hooks: map[string]Hook{},
|
|
}
|
|
}
|
|
|
|
// Action registers an action under name (overwrites an existing one).
|
|
func (r *Registry) Action(name string, fn Action) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.actions[name] = fn
|
|
}
|
|
|
|
// Predicate registers a predicate under name.
|
|
func (r *Registry) Predicate(name string, fn Predicate) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.predicates[name] = fn
|
|
}
|
|
|
|
// Hook registers a hook under name.
|
|
func (r *Registry) Hook(name string, fn Hook) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.hooks[name] = fn
|
|
}
|
|
|
|
func (r *Registry) action(name string) (Action, bool) {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
a, ok := r.actions[name]
|
|
return a, ok
|
|
}
|
|
|
|
func (r *Registry) predicate(name string) (Predicate, bool) {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
p, ok := r.predicates[name]
|
|
return p, ok
|
|
}
|
|
|
|
func (r *Registry) hook(name string) (Hook, bool) {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
h, ok := r.hooks[name]
|
|
return h, ok
|
|
}
|
|
|
|
// Capabilities is the set of registered names, surfaced so an editor UI can
|
|
// offer exactly the actions/predicates/hooks a flow may reference.
|
|
type Capabilities struct {
|
|
Actions []string `json:"actions"`
|
|
Predicates []string `json:"predicates"`
|
|
Hooks []string `json:"hooks"`
|
|
}
|
|
|
|
// Capabilities returns the sorted registered names.
|
|
func (r *Registry) Capabilities() Capabilities {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
return Capabilities{
|
|
Actions: sortedKeys(r.actions),
|
|
Predicates: sortedKeys(r.predicates),
|
|
Hooks: sortedKeys(r.hooks),
|
|
}
|
|
}
|
|
|
|
func sortedKeys[V any](m map[string]V) []string {
|
|
out := make([]string, 0, len(m))
|
|
for k := range m {
|
|
out = append(out, k)
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// StepResult records the outcome of one step.
|
|
type StepResult struct {
|
|
Name string `json:"name"`
|
|
Skipped bool `json:"skipped,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
// Result is the outcome of a whole flow run.
|
|
type Result struct {
|
|
Flow string `json:"flow"`
|
|
Steps []StepResult `json:"steps"`
|
|
Compensated []string `json:"compensated,omitempty"`
|
|
Failed bool `json:"failed"`
|
|
}
|
|
|
|
// Engine runs flow definitions against a registry.
|
|
type Engine struct {
|
|
reg *Registry
|
|
logger *slog.Logger
|
|
}
|
|
|
|
// NewEngine returns an engine bound to reg.
|
|
func NewEngine(reg *Registry, logger *slog.Logger) *Engine {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &Engine{reg: reg, logger: logger}
|
|
}
|
|
|
|
// Validate checks that every action, predicate and hook a definition references
|
|
// is registered. Call it at load time to fail fast on a bad flow config.
|
|
func (e *Engine) Validate(def *Definition) error {
|
|
for _, s := range def.Steps {
|
|
if _, ok := e.reg.action(s.Action); !ok {
|
|
return fmt.Errorf("flow %q step %q: unknown action %q", def.Name, s.Name, s.Action)
|
|
}
|
|
if s.When != "" {
|
|
if _, ok := e.reg.predicate(s.When); !ok {
|
|
return fmt.Errorf("flow %q step %q: unknown predicate %q", def.Name, s.Name, s.When)
|
|
}
|
|
}
|
|
if s.Compensate != nil {
|
|
if _, ok := e.reg.action(s.Compensate.Action); !ok {
|
|
return fmt.Errorf("flow %q step %q: unknown compensate action %q", def.Name, s.Name, s.Compensate.Action)
|
|
}
|
|
}
|
|
for _, hs := range [][]HookRef{s.Hooks.Before, s.Hooks.After, s.Hooks.OnError} {
|
|
for _, h := range hs {
|
|
if _, ok := e.reg.hook(h.Type); !ok {
|
|
return fmt.Errorf("flow %q step %q: unknown hook %q", def.Name, s.Name, h.Type)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Run executes the flow. On a step error (without ContinueOnError) it fires the
|
|
// step's on-error hooks, compensates every already-executed step that declared a
|
|
// compensating action (in reverse), and returns the error. Hook failures are
|
|
// logged, never fatal.
|
|
func (e *Engine) Run(ctx context.Context, def *Definition, st *State) (*Result, error) {
|
|
res := &Result{Flow: def.Name}
|
|
var executed []Step // steps that completed, for reverse compensation
|
|
|
|
for _, step := range def.Steps {
|
|
if step.When != "" {
|
|
pred, ok := e.reg.predicate(step.When)
|
|
if !ok {
|
|
return res, e.fail(res, step, fmt.Errorf("unknown predicate %q", step.When))
|
|
}
|
|
run, err := pred(ctx, st)
|
|
if err != nil {
|
|
return res, e.abort(ctx, res, &executed, step, err, st)
|
|
}
|
|
if !run {
|
|
res.Steps = append(res.Steps, StepResult{Name: step.Name, Skipped: true})
|
|
st.Logger.Info("flow step skipped", "flow", def.Name, "step", step.Name)
|
|
continue
|
|
}
|
|
}
|
|
|
|
action, ok := e.reg.action(step.Action)
|
|
if !ok {
|
|
return res, e.fail(res, step, fmt.Errorf("unknown action %q", step.Action))
|
|
}
|
|
|
|
e.fireHooks(ctx, st, step.Hooks.Before, HookInfo{Step: step.Name, Phase: PhaseBefore})
|
|
|
|
err := action(ctx, st, step.Params)
|
|
if err != nil {
|
|
e.fireHooks(ctx, st, step.Hooks.OnError, HookInfo{Step: step.Name, Phase: PhaseOnError, Err: err})
|
|
if step.ContinueOnError {
|
|
res.Steps = append(res.Steps, StepResult{Name: step.Name, Error: err.Error()})
|
|
st.Logger.Warn("flow step failed, continuing", "flow", def.Name, "step", step.Name, "err", err)
|
|
executed = append(executed, step)
|
|
continue
|
|
}
|
|
return res, e.abort(ctx, res, &executed, step, err, st)
|
|
}
|
|
|
|
e.fireHooks(ctx, st, step.Hooks.After, HookInfo{Step: step.Name, Phase: PhaseAfter})
|
|
res.Steps = append(res.Steps, StepResult{Name: step.Name})
|
|
executed = append(executed, step)
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
// fail records a step error without compensation (used for config errors found
|
|
// before the action ran, so nothing needs undoing for this step).
|
|
func (e *Engine) fail(res *Result, step Step, err error) error {
|
|
res.Failed = true
|
|
res.Steps = append(res.Steps, StepResult{Name: step.Name, Error: err.Error()})
|
|
return fmt.Errorf("flow %q step %q: %w", res.Flow, step.Name, err)
|
|
}
|
|
|
|
// abort records the failing step, then compensates executed steps in reverse.
|
|
func (e *Engine) abort(ctx context.Context, res *Result, executed *[]Step, step Step, cause error, st *State) error {
|
|
res.Failed = true
|
|
res.Steps = append(res.Steps, StepResult{Name: step.Name, Error: cause.Error()})
|
|
e.compensate(ctx, res, *executed, st)
|
|
return fmt.Errorf("flow %q step %q: %w", res.Flow, step.Name, cause)
|
|
}
|
|
|
|
// compensate runs the compensating action of each executed step in reverse.
|
|
func (e *Engine) compensate(ctx context.Context, res *Result, executed []Step, st *State) {
|
|
for i := len(executed) - 1; i >= 0; i-- {
|
|
step := executed[i]
|
|
if step.Compensate == nil {
|
|
continue
|
|
}
|
|
action, ok := e.reg.action(step.Compensate.Action)
|
|
if !ok {
|
|
st.Logger.Error("flow compensation skipped: unknown action",
|
|
"flow", res.Flow, "step", step.Name, "action", step.Compensate.Action)
|
|
continue
|
|
}
|
|
if err := action(ctx, st, step.Compensate.Params); err != nil {
|
|
st.Logger.Error("flow compensation failed", "flow", res.Flow, "step", step.Name, "err", err)
|
|
continue
|
|
}
|
|
res.Compensated = append(res.Compensated, step.Name)
|
|
}
|
|
}
|
|
|
|
// fireHooks runs every hook for a phase, logging (never propagating) failures.
|
|
func (e *Engine) fireHooks(ctx context.Context, st *State, refs []HookRef, info HookInfo) {
|
|
for _, ref := range refs {
|
|
h, ok := e.reg.hook(ref.Type)
|
|
if !ok {
|
|
st.Logger.Error("flow hook skipped: unknown type", "hook", ref.Type, "step", info.Step)
|
|
continue
|
|
}
|
|
if err := h(ctx, st, info, ref.Params); err != nil {
|
|
st.Logger.Warn("flow hook error", "hook", ref.Type, "step", info.Step, "phase", info.Phase, "err", err)
|
|
}
|
|
}
|
|
}
|