order sagas

pkg/flow/engine.go

@@ -0,0 +1,301 @@
+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)
+		}
+	}
+}