api

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Dec 12, 2025 License: MIT Imports: 9 Imported by: 0

Documentation

Overview

Package api contains the core building blocks used by the fluxo workflow engine. It provides the low-level primitives for defining workflows, building execution graphs, and observing engine behavior.

Most users interact with the higher-level fluxo package, which re-exports selected types and helpers from this package. The api package is intended for advanced use cases, custom integrations, or contributors extending the engine itself.

Concepts

The api package centers around a small set of concepts:

  • Workflow definitions
  • Steps and step functions
  • Control-flow composition
  • Observability

These primitives are assembled by the higher-level FlowBuilder API in the fluxo package, but can also be used directly where fine-grained control is needed.

Workflow Definitions

A workflow definition describes the structure of a workflow: its name, input/output types, and the graph of steps that will be executed.

Definitions are immutable once constructed and are registered with an engine before they can be started. The engine uses these definitions to compute execution plans and to reconstruct the next step to run when a workflow resumes.

Steps and Step Functions

A step represents a single unit of work in a workflow. Steps are backed by step functions, which encapsulate user code. The engine invokes step functions deterministically according to the workflow definition.

Step functions are expected to:

  • Be deterministic: same inputs yield the same observable behavior.
  • Be idempotent: they may be retried if a worker crashes or a task is rescheduled.
  • Modify workflow state through well-defined mechanisms provided by the engine (e.g. updating variables, scheduling timers, sending signals).

Control Flow

The api package defines the core control-flow nodes used to build workflows, including:

  • Sequential steps
  • Conditionals (if / switch style branching)
  • Parallel branches and parallel maps
  • Loops (while-style and counted loops)
  • Signal waits and timers

These constructs are exposed in a more ergonomic form from the fluxo package, but they are all built from common definitions and step primitives found here.

Typed helpers are available to work with strongly typed inputs and outputs without forcing the caller to manage serialization manually.

Observability

The api package defines the Observer interface, which is used by engines, workers, and runners to report lifecycle events and metrics.

Observers can be used to:

  • Log workflow and step transitions
  • Collect metrics (e.g. counts, latencies, error rates)
  • Integrate with external monitoring systems

The fluxo package exposes ready-made implementations such as logging and basic in-memory metrics, along with helpers to combine multiple observers.

Usage

Most applications should start from the fluxo package, using the FlowBuilder and Engine constructors provided there. The api package is useful when you need lower-level access, custom composition, or when contributing changes to the core engine.

See the fluxo package documentation and the examples directory for end-to-end usage.

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func ContextWithInstance 

func ContextWithInstance(ctx context.Context, inst *WorkflowInstance) context.Context

func IsWaitForAnyChildError 

func IsWaitForAnyChildError(err error) bool

IsWaitForAnyChildError returns true if the error is WaitForAnyChildError.

func IsWaitForSignalError 

func IsWaitForSignalError(err error) (string, bool)

IsWaitForSignalError returns (signalName, true) if Err indicates that the step wants to wait for a signal.

func NewWaitForSignalError 

func NewWaitForSignalError(name string) error

NewWaitForSignalError is primarily intended for use by helper step constructors (like WaitForSignalStep), but can be used by custom steps to integrate with the engine's signal semantics.

func WithEngine 

func WithEngine(ctx context.Context, e Engine) context.Context

WithEngine attaches an Engine to ctx so that StepFunc implementations can discover the engine (for example to start or inspect child workflows).

This is called internally by the engine implementation before invoking step functions.

Types

type BasicMetrics 

type BasicMetrics struct {
	NoopObserver
	// contains filtered or unexported fields
}

BasicMetrics collects simple counters and aggregate step durations. It implements Observer, and can be combined with LoggingObserver via NewCompositeObserver.

func (*BasicMetrics) OnStepCompleted 

func (m *BasicMetrics) OnStepCompleted(ctx context.Context, inst *WorkflowInstance, stepName string, idx int, err error, d time.Duration)

func (*BasicMetrics) OnWorkflowCompleted 

func (m *BasicMetrics) OnWorkflowCompleted(ctx context.Context, inst *WorkflowInstance)

func (*BasicMetrics) OnWorkflowFailed 

func (m *BasicMetrics) OnWorkflowFailed(ctx context.Context, inst *WorkflowInstance, err error)

func (*BasicMetrics) OnWorkflowStart 

func (m *BasicMetrics) OnWorkflowStart(ctx context.Context, inst *WorkflowInstance)

func (*BasicMetrics) Snapshot 

func (m *BasicMetrics) Snapshot() BasicMetricsSnapshot

Snapshot returns a snapshot of the current metrics.

type BasicMetricsSnapshot 

type BasicMetricsSnapshot struct {
	WorkflowsStarted   int64
	WorkflowsCompleted int64
	WorkflowsFailed    int64
	PendingWorkflows   int64

	StepsCompleted  int64
	AvgStepDuration time.Duration
}

BasicMetricsSnapshot is an immutable snapshot of BasicMetrics.

type ChildWorkflowSpec 

type ChildWorkflowSpec struct {
	// Name is the workflow definition name to start as a child.
	Name string
	// Input is the initial input for the child workflow.
	Input any
}

type CompositeObserver 

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

CompositeObserver fans out events to multiple observers.

func (*CompositeObserver) OnStepCompleted 

func (c *CompositeObserver) OnStepCompleted(ctx context.Context, inst *WorkflowInstance, stepName string, idx int, err error, d time.Duration)

func (*CompositeObserver) OnStepStart 

func (c *CompositeObserver) OnStepStart(ctx context.Context, inst *WorkflowInstance, stepName string, idx int)

func (*CompositeObserver) OnWorkflowCompleted 

func (c *CompositeObserver) OnWorkflowCompleted(ctx context.Context, inst *WorkflowInstance)

func (*CompositeObserver) OnWorkflowFailed 

func (c *CompositeObserver) OnWorkflowFailed(ctx context.Context, inst *WorkflowInstance, err error)

func (*CompositeObserver) OnWorkflowStart 

func (c *CompositeObserver) OnWorkflowStart(ctx context.Context, inst *WorkflowInstance)

type ConditionFunc 

type ConditionFunc func(input any) bool

ConditionFunc decides whether a branch should be taken based on the current input. It must be deterministic.

type Engine 

type Engine interface {
	// RegisterWorkflow registers a definition by name.
	RegisterWorkflow(def WorkflowDefinition) error

	// Run starts and runs the workflow to completion (synchronously).
	Run(ctx context.Context, name string, input any) (*WorkflowInstance, error)

	// GetInstance looks up a workflow instance by ID.
	// Returns an error if the instance is not found.
	GetInstance(ctx context.Context, id string) (*WorkflowInstance, error)

	// ListInstances returns workflow instances matching the given options.
	// If options are zero-valued, all instances are returned.
	ListInstances(ctx context.Context, opts InstanceListOptions) ([]*WorkflowInstance, error)

	// Resume restarts a previously failed workflow instance.
	// Semantics (first iteration):
	//   - Only FAILED instances can be resumed.
	//   - The instance is replayed from the beginning using its stored Input.
	//   - The same instance ID is reused; Status/Err/Output/CurrentStep are updated.
	Resume(ctx context.Context, id string) (*WorkflowInstance, error)

	// Signal delivers a named signal to a waiting workflow instance and
	// resumes it from the step that requested the signal.
	Signal(ctx context.Context, id string, name string, payload any) (*WorkflowInstance, error)

	// RecoverStuckInstances scans for in-flight workflow instances that are
	// still marked as StatusRunning (for example after a process crash) and
	// marks them as StatusFailed with a standard error message.
	//
	// It returns the number of instances it updated.
	//
	// This method is intended to be called on process startup *before*
	// starting workers or accepting new work, so that no instance is
	// legitimately running when it is executed.
	RecoverStuckInstances(ctx context.Context) (int, error)
}

Engine is the high-level engine API (iteration 1: synchronous).

func EngineFromContext 

func EngineFromContext(ctx context.Context) Engine

EngineFromContext retrieves the Engine previously attached with WithEngine. It returns nil if no engine is present in ctx.

type InstanceListOptions 

type InstanceListOptions struct {
	// WorkflowName, if non-empty, limits results to instances of the given workflow.
	WorkflowName string

	// Status, if non-empty, limits results to instances with the given status.
	Status Status
}

InstanceListOptions controls how instances are listed. Zero values mean "no filter" for that field.

type LoggingObserver 

type LoggingObserver struct {
	Logger *slog.Logger
}

LoggingObserver writes structured logs using log/slog.

func (*LoggingObserver) OnStepCompleted 

func (o *LoggingObserver) OnStepCompleted(ctx context.Context, inst *WorkflowInstance, stepName string, idx int, err error, d time.Duration)

func (*LoggingObserver) OnStepStart 

func (o *LoggingObserver) OnStepStart(ctx context.Context, inst *WorkflowInstance, stepName string, idx int)

func (*LoggingObserver) OnWorkflowCompleted 

func (o *LoggingObserver) OnWorkflowCompleted(ctx context.Context, inst *WorkflowInstance)

func (*LoggingObserver) OnWorkflowFailed 

func (o *LoggingObserver) OnWorkflowFailed(ctx context.Context, inst *WorkflowInstance, err error)

func (*LoggingObserver) OnWorkflowStart 

func (o *LoggingObserver) OnWorkflowStart(ctx context.Context, inst *WorkflowInstance)

type NoopObserver 

type NoopObserver struct{}

NoopObserver is an Observer that does nothing. It is used as the default when no observer is configured.

func (NoopObserver) OnStepCompleted 

func (NoopObserver) OnStepCompleted(ctx context.Context, inst *WorkflowInstance, stepName string, idx int, err error, d time.Duration)

func (NoopObserver) OnStepStart 

func (NoopObserver) OnStepStart(ctx context.Context, inst *WorkflowInstance, stepName string, idx int)

func (NoopObserver) OnWorkflowCompleted 

func (NoopObserver) OnWorkflowCompleted(ctx context.Context, inst *WorkflowInstance)

func (NoopObserver) OnWorkflowFailed 

func (NoopObserver) OnWorkflowFailed(ctx context.Context, inst *WorkflowInstance, err error)

func (NoopObserver) OnWorkflowStart 

func (NoopObserver) OnWorkflowStart(ctx context.Context, inst *WorkflowInstance)

type Observer 

type Observer interface {
	// OnWorkflowStart is called once when a workflow instance is first started
	// (Run), before the first step is executed.
	OnWorkflowStart(ctx context.Context, inst *WorkflowInstance)

	// OnWorkflowCompleted is called when a workflow instance successfully
	// reaches StatusCompleted.
	OnWorkflowCompleted(ctx context.Context, inst *WorkflowInstance)

	// OnWorkflowFailed is called when a workflow instance transitions to
	// StatusFailed.
	OnWorkflowFailed(ctx context.Context, inst *WorkflowInstance, err error)

	// OnStepStart is called before invoking a step function.
	// stepIndex is the 0-based index into WorkflowDefinition.Steps.
	OnStepStart(ctx context.Context, inst *WorkflowInstance, stepName string, stepIndex int)

	// OnStepCompleted is called after a step function returns, for both
	// successes and failures (err != nil).
	OnStepCompleted(ctx context.Context, inst *WorkflowInstance, stepName string, stepIndex int, err error, duration time.Duration)
}

Observer receives callbacks from the workflow engine for logging and metrics.

Implementations should be fast and non-blocking; heavy work should be done asynchronously so as not to delay workflow execution.

func NewCompositeObserver 

func NewCompositeObserver(obs ...Observer) Observer

NewCompositeObserver creates an Observer that forwards events to each non-nil observer in obs.

func NewLoggingObserver 

func NewLoggingObserver(logger *slog.Logger) Observer

NewLoggingObserver creates an Observer that logs workflow / step lifecycle events using the provided slog.Logger. If logger is nil, slog.Default() is used.

type ParallelResult 

type ParallelResult struct {
	Index int
	Value any
	Err   error
}

ParallelResult represents the out from a parallel step to next step.

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts int
	// Deprecated: prefer InitialBackoff + BackoffMultiplier. If
	// InitialBackoff is zero and BackoffMultiplier is zero, Backoff
	// is used as the initial delay between retries.
	Backoff time.Duration

	// InitialBackoff is the base delay before the first retry.
	InitialBackoff time.Duration

	// MaxBackoff caps the delay between retries. If zero, there is no cap.
	MaxBackoff time.Duration

	// BackoffMultiplier is the factor by which the delay grows after each
	// failed attempt. If <= 0, a default of 2.0 is used.
	//
	// Example:
	//   InitialBackoff = 100ms, BackoffMultiplier = 2.0
	//   attempt 1 -> 2: sleep 100ms
	//   attempt 2 -> 3: sleep 200ms
	//   attempt 3 -> 4: sleep 400ms (capped by MaxBackoff if set)
	BackoffMultiplier float64
}

RetryPolicy controls how a step is retried when it returns an error. MaxAttempts includes the first attempt. For example: 

MaxAttempts = 1 => no retries (just the initial call)
MaxAttempts = 3 => initial call + up to 2 retries

Backoff is the delay between failed attempts. It is not applied before the first attempt. If zero, retries happen immediately.

type SelectorFunc 

type SelectorFunc func(input any) string

SelectorFunc picks a branch key from the input. It must be deterministic.

type SignalPayload 

type SignalPayload struct {
	Name string
	Data any
}

SignalPayload is used by the engine to resume a workflow step that previously requested to wait for a signal.

type Status 

type Status string

Status represents the lifecycle state of a workflow instance. 

const (
	StatusPending   Status = "PENDING"
	StatusRunning   Status = "RUNNING"
	StatusCompleted Status = "COMPLETED"
	StatusFailed    Status = "FAILED"
	StatusWaiting   Status = "WAITING"
)

type StepDefinition 

type StepDefinition struct {
	Name  string
	Fn    StepFunc
	Retry *RetryPolicy
}

StepDefinition describes a named step.

type StepFunc 

type StepFunc func(ctx context.Context, input any) (any, error)

StepFunc is a single step in a workflow. Iteration 1: keep it simple with `any`, we can add generics later.

func IfStep 

func IfStep(cond ConditionFunc, thenStep StepFunc, elseStep StepFunc) StepFunc

IfStep returns a step that evaluates cond on the input and, depending on the ParallelResult, runs thenStep or elseStep.

The selected branch is executed as a nested step (i.e. the engine only sees this as a single step); retries/backoff for this step apply to the entire nested execution.

If elseStep is nil, the input is passed through unchanged on the false branch.

func LoopStep 

func LoopStep(times int, body StepFunc) StepFunc

LoopStep returns a StepFunc that executes body a fixed number of times. The loop is treated as a single engine step; retries/backoff (if any) apply to the entire loop execution.

func ParallelMapStep 

func ParallelMapStep(fn StepFunc) StepFunc

ParallelMapStep returns a StepFunc that expects the input to be a slice or array and runs fn in parallel for each element.

Input:

  • A slice or array Value (e.g. []T).

Behavior:

  • For each element input[i], fn(ctx, input[i]) is executed in its own goroutine.
  • The outputs are collected into a []any of the same length, preserving the original order.
  • If any call returns an error, the first error is returned.
  • If ctx is cancelled, ctx.Err() is returned.

This is a convenient fan-out/fan-in helper for data-parallel workloads.

func ParallelStep 

func ParallelStep(steps ...StepFunc) StepFunc

ParallelStep returns a StepFunc that runs multiple child steps in parallel and collects their outputs.

Semantics:

  • All non-nil child steps receive the same input Value.
  • The ParallelResult is a []any whose length equals len(steps), preserving order.
  • If any child step returns an error, the *first* error is returned and no further processing is done after all goroutines finish.
  • If ctx is cancelled, ctx.Err() is returned.

NOTE: This is an in-process parallelism helper. The underlying engine still sees this as a single step, so retries/backoff apply to the whole parallel group as a unit.

func SleepStep 

func SleepStep(d time.Duration) StepFunc

SleepStep returns a StepFunc that waits for the given duration before passing the input through unchanged.

It is context-aware: if the context is cancelled during the sleep, it returns ctx.Err and the workflow will fail at this step.

func SleepUntilStep 

func SleepUntilStep(deadline time.Time) StepFunc

SleepUntilStep returns a StepFunc that waits until the given deadline before passing the input through unchanged.

If the deadline is in the past or equal to now, it returns immediately. It is context-aware: if the context is cancelled while waiting, it returns ctx.Err and the workflow will fail at this step.

func StartChildrenStep 

func StartChildrenStep(specsFn func(input any) ([]ChildWorkflowSpec, error)) StepFunc

StartChildrenStep returns a step that starts one or more child workflows and returns their instance IDs as []string.

The specsFn callback is given the current input and must deterministically construct the list of child workflows to start.

func SwitchStep 

func SwitchStep(selector SelectorFunc, branches map[string]StepFunc, defaultStep StepFunc) StepFunc

SwitchStep returns a step that selects one of several branches based on selector(input). The branches map holds per-key steps; if no branch matches, defaultStep is used (if non-nil), otherwise the input is passed through unchanged.

As with IfStep, the chosen branch is executed as a nested step from the engine's perspective.

func TypedLoop 

func TypedLoop[I any](times int, body func(context.Context, I) (I, error)) StepFunc

TypedLoop returns a StepFunc that executes a strongly-typed body a fixed number of times. The loop is treated as a single engine step.

func TypedStep 

func TypedStep[I, O any](fn func(context.Context, I) (O, error)) StepFunc

TypedStep wraps a strongly-typed function into a generic StepFunc. It performs a type assertion on input at runtime and returns an error if the input is not of the expected type.

func TypedWhile 

func TypedWhile[I any](cond func(I) bool, body func(context.Context, I) (I, error)) StepFunc

TypedWhile returns a StepFunc that repeatedly executes a strongly-typed body while cond(input) is true. The loop is treated as a single engine step.

func WaitForAnyChildStep 

func WaitForAnyChildStep(
	getIDs func(input any) []string,
	pollInterval time.Duration,
) StepFunc

WaitForAnyChildStep waits until any one of the specified child workflow instances has reached a terminal state.

getIDs extracts the child instance IDs from the input (for example, from the []string returned by StartChildrenStep).

Semantics:

  • If getIDs(input) returns an empty slice, the step returns (nil, nil).
  • If any child is StatusFailed, the step returns a hard error describing the failure (not a WaitForAnyChildError).
  • If any child is StatusCompleted, the step returns that child ID as output. The "first" completed child is determined by the order of IDs returned by getIDs.
  • If no child is terminal yet (all are PENDING/RUNNING/etc.), the step returns a *WaitForAnyChildError with the IDs and PollAfter interval. The engine should treat this as a request to park the workflow and resume it later.

func WaitForAnySignalStep 

func WaitForAnySignalStep(names ...string) StepFunc

WaitForAnySignalStep returns a step that parks the workflow until a signal with any of the given names is delivered via Engine.Signal.

Behavior:

  • At first invocation (normal forward execution), it ignores its input and returns NewWaitForSignalError(names[0]) which causes the engine to mark the instance as WAITING.
  • When resumed via Engine.Signal, the engine passes a SignalPayload as input (Name, Data). If Name is in the allowed list, this step returns that SignalPayload as its output and the workflow continues.
  • If a signal with an unexpected name is delivered, the step will request to wait again via NewWaitForSignalError(names[0]).

This is useful for branching decisions such as approve/reject flows, where the next step can switch on the SignalPayload.Name to decide behavior.

func WaitForChildrenStep 

func WaitForChildrenStep(
	getIDs func(input any) []string,
	pollInterval time.Duration,
) StepFunc

WaitForChildrenStep returns a step that waits until all specified child workflow instances have completed.

getIDs extracts the child instance IDs from the input (for example, from the []string returned by StartChildrenStep).

Durable semantics:

  • Each invocation checks child states exactly once.
  • If some children are still running/pending, the step returns a WaitForChildrenError with a suggested PollAfter duration.
  • The engine must treat WaitForChildrenError like a "WAITING" signal: mark the instance as WAITING and schedule a resume after PollAfter.
  • When the workflow is resumed, this step is re-invoked with the same input, re-checks statuses, and eventually returns the children outputs once all are completed.

func WaitForSignalStep 

func WaitForSignalStep(name string) StepFunc

WaitForSignalStep returns a step that parks the workflow until a signal with the given name is delivered via Engine.Signal.

Semantics:

  • First time the step runs, it ignores its input and returns NewWaitForSignalError(name). The engine marks the instance as WAITING and stops execution.
  • When Engine.Signal is called, the engine resumes this step with input set to SignalPayload{Name: name, Data: payload}. In that case, the step returns payload (Data) as its output and the workflow continues with the next step.

func WhileStep 

func WhileStep(cond ConditionFunc, body StepFunc) StepFunc

WhileStep returns a StepFunc that repeatedly executes body while cond(input) is true. The entire loop is treated as a single engine step: retries/backoff (if any) apply to the whole loop execution.

type TimeoutPayload 

type TimeoutPayload struct {
	Reason string
}

TimeoutPayload is a special payload used for auto-expiry signals. Workflows can check for this type to detect that a wait timed out.

type WaitForAnyChildError 

type WaitForAnyChildError struct {
	ChildIDs  []string
	PollAfter time.Duration
}

WaitForAnyChildError signals that the workflow should pause until one of the given child workflows has completed.

func (*WaitForAnyChildError) Error 

func (e *WaitForAnyChildError) Error() string

type WaitForChildrenError 

type WaitForChildrenError struct {
	ChildIDs  []string
	PollAfter time.Duration
}

WaitForChildrenError is a sentinel error used by WaitForChildrenStep to tell the engine that the parent workflow should be parked (WAITING) and resumed later to re-check the child statuses.

The engine is responsible for scheduling a resume after PollAfter.

func (*WaitForChildrenError) Error 

func (e *WaitForChildrenError) Error() string

type WorkflowDefinition 

type WorkflowDefinition struct {
	Name  string
	Steps []StepDefinition
}

WorkflowDefinition describes a workflow as a sequence of steps.

Example

ExampleWorkflowDefinition shows how to build a workflow definition directly using the api package and register it on an Engine. 

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/petrijr/fluxo"
	"github.com/petrijr/fluxo/pkg/api"
)

func main() {
	ctx := context.Background()

	// Build a simple definition manually.
	def := api.WorkflowDefinition{
		Name: "AddPrefix",
		Steps: []api.StepDefinition{
			{
				Name: "addPrefix",
				Fn: func(ctx context.Context, input any) (any, error) {
					s, ok := input.(string)
					if !ok {
						return nil, fmt.Errorf("expected string input, got %T", input)
					}
					return "prefix:" + s, nil
				},
			},
		},
	}

	// Use a real engine implementation from the fluxo package.
	eng := fluxo.NewInMemoryEngine()

	if err := eng.RegisterWorkflow(def); err != nil {
		log.Fatal(err)
	}

	inst, err := eng.Run(ctx, def.Name, "value")
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("instance %s finished with status %s and output %v\n",
		inst.ID, inst.Status, inst.Output)
}

type WorkflowInstance 

type WorkflowInstance struct {
	ID     string
	Name   string
	Status Status
	Output any
	Err    error

	// Input is the original input provided to Run when this instance
	// was first started. It is used for deterministic replay on resume.
	Input any

	// CurrentStep tracks progress through the workflow steps.
	// Semantics:
	//   - Before any steps run: 0 (default)
	//   - While running step i: i
	//   - After successful completion: len(steps)
	//   - On failure: Index of the step that failed (or was cancelled).
	CurrentStep int

	// StepResults holds the output of successfully completed steps,
	// keyed by their index in the WorkflowDefinition.Steps slice.
	//
	// This is used for step-level idempotency: on replay/resume, the engine
	// can skip re-running steps that have already succeeded and reuse their
	// cached outputs as inputs to downstream steps.
	StepResults map[int]any
}

WorkflowInstance holds the ParallelResult of a run.

func InstanceFromContext 

func InstanceFromContext(ctx context.Context) *WorkflowInstance

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.