engine

package
v0.29.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 15, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// Engine errors.
	ErrNoMatchingEdge    = errors.New("no matching outgoing edge for node")
	ErrProcessorNotFound = errors.New("node processor not found for node kind")
	ErrMaxNodeDepth      = errors.New("max node processing depth exceeded")

	// Approval node errors.
	ErrAssigneeServiceNotConfigured = errors.New("assignee service is not configured")

	// Condition node errors.
	ErrNoBranches       = errors.New("condition node has no branches")
	ErrNoMatchingBranch = errors.New("no matching branch and no default branch")

	// ErrInvalidTransition signals that a requested state transition is not
	// permitted by the state machine, or that a concurrent writer already
	// advanced the target row off the expected `from` status.
	ErrInvalidTransition = errors.New("invalid state transition")

	// CompiledFlow errors.
	ErrFlowMissingStartNode  = errors.New("flow has no start node")
	ErrFlowMissingTargetNode = errors.New("compiled flow is missing target node")
	ErrFlowNoNodes           = errors.New("compiled flow has no nodes for version")
)
View Source 
var InstanceStateMachine = buildInstanceStateMachine()

InstanceStateMachine defines valid instance state transitions.

View Source 
var Module = fx.Module(
	"vef:approval:engine",

	fx.Provide(

		shared.NewCCRecipientResolver,

		fx.Annotate(NewStartProcessor, fx.As(new(NodeProcessor)), fx.ResultTags(`group:"vef:approval:node_processors"`)),
		fx.Annotate(NewEndProcessor, fx.As(new(NodeProcessor)), fx.ResultTags(`group:"vef:approval:node_processors"`)),
		fx.Annotate(NewConditionProcessor, fx.As(new(NodeProcessor)), fx.ResultTags(`group:"vef:approval:node_processors"`)),
		fx.Annotate(NewApprovalProcessor, fx.As(new(NodeProcessor)), fx.ResultTags(`group:"vef:approval:node_processors"`)),
		fx.Annotate(NewHandleProcessor, fx.As(new(NodeProcessor)), fx.ResultTags(`group:"vef:approval:node_processors"`)),
		fx.Annotate(NewCCProcessor, fx.As(new(NodeProcessor)), fx.ResultTags(`group:"vef:approval:node_processors"`)),

		fx.Annotate(
			NewLifecycleHookRunner,
			fx.ParamTags(`group:"vef:approval:lifecycle_hooks"`),
		),

		newDefaultCompiledFlowCache,
		NewFlowCache,

		fx.Annotate(
			NewFlowEngine,
			fx.ParamTags(``, `group:"vef:approval:node_processors"`, ``, ``, ``, ``),
		),
	),
)

Module provides the flow engine and node processors.

View Source 
var TaskStateMachine = buildTaskStateMachine()

TaskStateMachine defines valid task state transitions.

Functions

func ApplyInstanceTransitionWithHooks added in v0.24.0 

func ApplyInstanceTransitionWithHooks(
	ctx context.Context,
	db orm.DB,
	instance *approval.Instance,
	to approval.InstanceStatus,
	hooks *LifecycleHookRunner,
	extraCols ...string,
) error

ApplyInstanceTransitionWithHooks is the single write-side primitive for instance status transitions. It validates the transition through InstanceStateMachine, applies it atomically via an optimistic-lock UPDATE (WHERE pk AND status=from), and — when the new status is final and hooks is non-nil — invokes the registered LifecycleHookRunner inside the same transaction. All instance-completion paths (engine NodeActionComplete, pass-rule rejection, admin terminate, resubmit/withdraw, etc.) funnel through this helper so hooks fire consistently.

extraCols lists columns the caller pre-populated on instance and wants persisted in the same UPDATE (e.g. "finished_at", "current_node_id", "form_data"). The status column is always included.

Returns ErrInvalidTransition when the transition is not declared on the state machine, or when zero rows match (concurrent writer already moved the row off `from`). The in-memory instance.Status is restored on failure. Pass hooks=nil to skip hook invocation (e.g. test fixtures).

func PublishEventsTx added in v0.24.0 

func PublishEventsTx(ctx context.Context, bus event.Bus, db orm.DB, events ...approval.DomainEvent) error

PublishEventsTx publishes domain events through the bus, enrolled in the caller transaction, projecting each payload's OccurredTime onto the envelope so downstream consumers see "when the thing happened" rather than "when we got around to publishing". Returns nil when bus is nil or events is empty.

This is the entry point used by sites outside the CQRS pipeline (timeout scanner, engine fallback when no EventCollector is bound, node service auto-CC). The CQRS pipeline itself uses EventCollector + EventPublishBehavior to batch the publish at the end of the handler. Both paths share behavior.PublishEventsTx so the publish loop lives in one place.

Types

type ApprovalProcessor 

type ApprovalProcessor struct {
	// contains filtered or unexported fields
}

ApprovalProcessor handles approval nodes.

func NewApprovalProcessor 

func NewApprovalProcessor(assigneeService approval.AssigneeService) *ApprovalProcessor

NewApprovalProcessor creates a new approval processor.

func (*ApprovalProcessor) NodeKind 

func (*ApprovalProcessor) NodeKind() approval.NodeKind

func (*ApprovalProcessor) Process 

func (p *ApprovalProcessor) Process(ctx context.Context, pc *ProcessContext) (*ProcessResult, error)

type CCProcessor 

type CCProcessor struct {
	// contains filtered or unexported fields
}

CCProcessor handles CC (carbon copy) notification nodes.

func NewCCProcessor 

func NewCCProcessor(ccResolver *shared.CCRecipientResolver) *CCProcessor

NewCCProcessor creates a CCProcessor.

func (*CCProcessor) NodeKind 

func (*CCProcessor) NodeKind() approval.NodeKind

func (*CCProcessor) Process 

func (p *CCProcessor) Process(ctx context.Context, pc *ProcessContext) (*ProcessResult, error)

type CompiledFlow added in v0.24.0 

type CompiledFlow struct {
	// FlowVersionID is the key under which this compilation was cached.
	FlowVersionID string
	// Nodes indexes FlowNode by ID.
	Nodes map[string]*approval.FlowNode
	// StartNode points to the unique start node of the flow (engine
	// traversal entry point).
	StartNode *approval.FlowNode
	// EdgesBySource groups outgoing edges by source node ID. Slice order
	// is undefined; callers needing branching look up via SourceHandle.
	EdgesBySource map[string][]*approval.FlowEdge
}

CompiledFlow is the in-memory, query-free representation of a published flow version: every FlowNode keyed by ID, and the outgoing edges grouped by source node so engine traversal becomes a series of map lookups.

FlowVersion rows are immutable once published (archive only flips status; nodes & edges never change), so a CompiledFlow can be cached for the lifetime of the version with no invalidation beyond explicit drops on version replacement.

func (*CompiledFlow) FindOutgoing added in v0.24.0 

func (c *CompiledFlow) FindOutgoing(sourceNodeID string, branchID *string) (*approval.FlowEdge, error)

FindOutgoing returns the first edge from sourceNodeID matching branchID (when branchID is nil, the unique unguarded out-edge). Returns ErrNoMatchingEdge when no edge matches and errAmbiguousEdges when more than one does (the engine relies on a total edge order per source).

type ConditionProcessor 

type ConditionProcessor struct{}

ConditionProcessor evaluates condition branches and selects the matching branch.

func NewConditionProcessor 

func NewConditionProcessor() *ConditionProcessor

NewConditionProcessor creates a ConditionProcessor.

func (*ConditionProcessor) NodeKind 

func (*ConditionProcessor) NodeKind() approval.NodeKind

func (*ConditionProcessor) Process 

func (*ConditionProcessor) Process(ctx context.Context, pc *ProcessContext) (*ProcessResult, error)

type EndProcessor 

type EndProcessor struct{}

EndProcessor handles end nodes by completing the flow as approved.

func NewEndProcessor 

func NewEndProcessor() *EndProcessor

NewEndProcessor creates an EndProcessor.

func (*EndProcessor) NodeKind 

func (*EndProcessor) NodeKind() approval.NodeKind

func (*EndProcessor) Process 

func (*EndProcessor) Process(context.Context, *ProcessContext) (*ProcessResult, error)

type FlowCache added in v0.24.0 

type FlowCache struct {
	// contains filtered or unexported fields
}

FlowCache provides cached access to CompiledFlow keyed by FlowVersionID. Hosts can swap the underlying cache.Cache (memory or Redis); the approval module ships memory by default.

func NewFlowCache added in v0.24.0 

func NewFlowCache(db orm.DB, c cache.Cache[*CompiledFlow]) *FlowCache

NewFlowCache constructs a FlowCache backed by the given cache.Cache. Pass cache.NewMemory[*CompiledFlow]() for a single-node deployment.

func (*FlowCache) Get added in v0.24.0 

func (c *FlowCache) Get(ctx context.Context, versionID string) (*CompiledFlow, error)

Get returns the compiled flow for versionID, hitting cache first and compiling from the database on miss. Concurrent misses for the same version coalesce into a single load via cache.GetOrLoad's singleflight.

func (*FlowCache) Invalidate added in v0.24.0 

func (c *FlowCache) Invalidate(ctx context.Context, versionID string) error

Invalidate evicts the cached compilation for a version. Called by publish_version when superseding a previously-published version (its status flips to archived, but in-flight instances keep their own FlowVersionID so no eviction is strictly required for them — the entry just becomes a one-off historical reference).

type FlowEngine 

type FlowEngine struct {
	// contains filtered or unexported fields
}

FlowEngine is the core engine for processing approval workflows.

func NewFlowEngine 

func NewFlowEngine(
	registry *strategy.StrategyRegistry,
	processors []NodeProcessor,
	bus event.Bus,
	userResolver approval.UserInfoResolver,
	hooks *LifecycleHookRunner,
	flowCache *FlowCache,
) *FlowEngine

NewFlowEngine creates a new flow engine. Duplicate NodeKind registrations panic on construction — silent overwrite is a deployment bug we want to surface at boot rather than mask at runtime.

flowCache is required: node/edge traversal resolves entirely through the compiled-flow cache (a series of map lookups over the immutable published version), so callers must supply one. Tests that exercise traversal build it with engine.NewFlowCache(db, cache.NewMemory[*CompiledFlow]()).

func (*FlowEngine) AdvanceToNextNode 

func (e *FlowEngine) AdvanceToNextNode(ctx context.Context, db orm.DB, instance *approval.Instance, fromNode *approval.FlowNode, branchID *string) error

AdvanceToNextNode finds the matching edge from the current node and advances to the next one. BranchID is used by condition nodes to select the edge matching the branch.

func (*FlowEngine) EvaluateNodeCompletion 

func (e *FlowEngine) EvaluateNodeCompletion(ctx context.Context, db orm.DB, instance *approval.Instance, node *approval.FlowNode) (approval.PassRuleResult, error)

EvaluateNodeCompletion evaluates whether a node is complete based on its tasks and pass rule.

func (*FlowEngine) EvaluatePassRuleWithTasks 

func (e *FlowEngine) EvaluatePassRuleWithTasks(node *approval.FlowNode, tasks []approval.Task) (approval.PassRuleResult, error)

EvaluatePassRuleWithTasks evaluates the pass rule for a node using the provided tasks. This is used for simulation (e.g., checking if removing an assignee would deadlock the node).

func (*FlowEngine) LifecycleHooks added in v0.24.0 

func (e *FlowEngine) LifecycleHooks() *LifecycleHookRunner

LifecycleHooks exposes the aggregated host-registered hooks so callers outside the engine (e.g. start_instance) can fire OnInstanceCreated at the right transactional moment.

func (*FlowEngine) ProcessNode 

func (e *FlowEngine) ProcessNode(ctx context.Context, db orm.DB, instance *approval.Instance, node *approval.FlowNode) error

ProcessNode dispatches a node to the appropriate processor.

func (*FlowEngine) StartProcess 

func (e *FlowEngine) StartProcess(ctx context.Context, db orm.DB, instance *approval.Instance) error

StartProcess starts a flow process by finding the start node and processing it.

type HandleProcessor 

type HandleProcessor struct {
	// contains filtered or unexported fields
}

HandleProcessor handles handle nodes (claim mode). Unlike approval nodes, handle nodes create all tasks with sortOrder=0, allowing any candidate to claim and complete the task.

func NewHandleProcessor 

func NewHandleProcessor(assigneeService approval.AssigneeService) *HandleProcessor

NewHandleProcessor creates a new handle processor.

func (*HandleProcessor) NodeKind 

func (*HandleProcessor) NodeKind() approval.NodeKind

func (*HandleProcessor) Process 

func (p *HandleProcessor) Process(ctx context.Context, pc *ProcessContext) (*ProcessResult, error)

type LifecycleHookRunner added in v0.24.0 

type LifecycleHookRunner struct {
	// contains filtered or unexported fields
}

LifecycleHookRunner aggregates host-registered InstanceLifecycleHook implementations and invokes them in registration order. A non-nil error short-circuits the chain so the caller can roll back the surrounding transaction.

The runner is intentionally minimal: hosts that want fan-out or async dispatch should subscribe to the corresponding domain events instead; hooks are reserved for cases that must run inside the business transaction.

func NewLifecycleHookRunner added in v0.24.0 

func NewLifecycleHookRunner(hooks []approval.InstanceLifecycleHook) *LifecycleHookRunner

NewLifecycleHookRunner constructs a runner from the FX group of hooks.

func (*LifecycleHookRunner) OnInstanceCompleted added in v0.24.0 

func (r *LifecycleHookRunner) OnInstanceCompleted(ctx context.Context, db orm.DB, instance *approval.Instance, finalStatus approval.InstanceStatus) error

OnInstanceCompleted invokes every registered hook's OnInstanceCompleted.

func (*LifecycleHookRunner) OnInstanceCreated added in v0.24.0 

func (r *LifecycleHookRunner) OnInstanceCreated(ctx context.Context, db orm.DB, instance *approval.Instance) error

OnInstanceCreated invokes every registered hook's OnInstanceCreated.

type NodeAction 

type NodeAction int

NodeAction represents the result action of node processing. 

const (
	NodeActionWait     NodeAction = iota // Wait for user action
	NodeActionContinue                   // Auto-advance to next node
	NodeActionComplete                   // Flow ends
)

type NodeProcessor 

type NodeProcessor interface {
	// NodeKind returns the node kind this processor handles (approval, handle, cc, condition, etc.).
	NodeKind() approval.NodeKind
	// Process executes the node logic and returns the action to take (wait, continue, or complete).
	Process(ctx context.Context, pc *ProcessContext) (*ProcessResult, error)
}

NodeProcessor processes a specific node kind.

type ProcessContext 

type ProcessContext struct {
	DB            orm.DB
	Instance      *approval.Instance
	Node          *approval.FlowNode
	FormData      approval.FormData
	ApplicantID   string
	ApplicantName string
	UserResolver  approval.UserInfoResolver
	Registry      *strategy.StrategyRegistry
}

ProcessContext provides context for node processing.

type ProcessResult 

type ProcessResult struct {
	Action      NodeAction
	FinalStatus *approval.InstanceStatus // Only set when Action == NodeActionComplete
	BranchID    *string                  // Only set when Action == NodeActionContinue (condition node)
	Events      []approval.DomainEvent   // Events to publish after processing
}

ProcessResult contains the outcome of node processing.

type StartProcessor 

type StartProcessor struct{}

StartProcessor handles start nodes by auto-advancing to the next node.

func NewStartProcessor 

func NewStartProcessor() *StartProcessor

NewStartProcessor creates a StartProcessor.

func (*StartProcessor) NodeKind 

func (*StartProcessor) NodeKind() approval.NodeKind

func (*StartProcessor) Process 

func (*StartProcessor) Process(context.Context, *ProcessContext) (*ProcessResult, error)

type State 

type State interface {
	comparable

	// String returns the string representation of this state.
	String() string
	// IsFinal returns true if this is a terminal state with no further transitions.
	IsFinal() bool
}

State represents a state that can be used in a state machine.

type StateMachine 

type StateMachine[S State] struct {
	// contains filtered or unexported fields
}

StateMachine manages state transitions.

func NewStateMachine 

func NewStateMachine[S State](name string) *StateMachine[S]

NewStateMachine creates a new state machine with the given name.

func (*StateMachine[S]) AddTransition 

func (sm *StateMachine[S]) AddTransition(from, to S) *StateMachine[S]

AddTransition registers a valid state transition.

func (*StateMachine[S]) CanTransition 

func (sm *StateMachine[S]) CanTransition(from, to S) bool

CanTransition checks if a transition from one state to another is valid.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.