engine

package
v0.38.0 Latest Latest
Warning

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

 Go to latest Published: Jul 11, 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")

	// ErrActiveVisitNotFound signals a broken visit-trail invariant: a path
	// that requires an executing node found no open visit for it.
	ErrActiveVisitNotFound = errors.New("no active node visit")

	// 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), then — inside the same transaction — advances the business projection and invokes registered host lifecycle hooks with the from/to pair. All instance paths (engine NodeActionComplete, pass-rule rejection, admin terminate, resubmit/withdraw, etc.) funnel through this helper so projection semantics and hooks stay consistent.

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 CancelActiveNodeVisits added in v0.35.0 

func CancelActiveNodeVisits(ctx context.Context, db orm.DB, instanceID string) error

CancelActiveNodeVisits concludes every active visit of the instance as canceled — withdraw and terminate cut the traversal short wherever it stands. A no-op when nothing is active (e.g. terminating an already-paused instance).

func ConcludeActiveNodeVisit added in v0.35.0 

func ConcludeActiveNodeVisit(ctx context.Context, db orm.DB, instanceID, nodeID string, status approval.NodeVisitStatus) error

ConcludeActiveNodeVisit stamps the outcome on the active visit of the given node. It is the conclusion primitive for paths that do not hold the visit in memory: pass-rule completion (HandleNodeCompletion), CC read-confirm advancement, and rollback. All of them act on an executing node, so exactly one row must match — anything else is a broken invariant.

func FindActiveNodeVisit added in v0.36.0 

func FindActiveNodeVisit(ctx context.Context, db orm.DB, instanceID, nodeID string) (*approval.NodeVisit, error)

FindActiveNodeVisit returns the node's open visit. The node must be executing when this is called, so a missing visit is a broken invariant and surfaces as an error rather than a silent fallback.

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. Counting is scoped to the node's open visit, so tasks left behind by an earlier traversal — e.g. an approval that survived a peer-initiated rollback — cannot contaminate the redo round's decision.

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 InstanceProjector added in v0.38.0 

type InstanceProjector interface {
	// Project records or applies the instance's latest complete business state.
	Project(ctx context.Context, db orm.DB, instance *approval.Instance) error
}

InstanceProjector advances the durable business projection after every instance status transition. It runs inside the caller's transaction.

type LifecycleHookRunner added in v0.24.0 

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

LifecycleHookRunner aggregates host-registered InstanceLifecycleHook implementations and invokes each of them; the order is unspecified (FX value groups carry no ordering), so hooks must not depend on one another. A non-nil error short-circuits the remaining hooks 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(projector InstanceProjector, hooks []approval.InstanceLifecycleHook) *LifecycleHookRunner

NewLifecycleHookRunner constructs a runner from the engine projector and FX group of host hooks.

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.

func (*LifecycleHookRunner) OnInstanceTransition added in v0.38.0 

func (r *LifecycleHookRunner) OnInstanceTransition(ctx context.Context, db orm.DB, instance *approval.Instance, from, to approval.InstanceStatus) error

OnInstanceTransition advances the engine-owned business projection, then invokes every registered hook's OnInstanceTransition — hooks always observe the business state as the projector left it.

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
	Visit         *approval.NodeVisit
	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.