workflow

package
v0.98.0 Latest Latest
Warning

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

 Go to latest Published: Jul 21, 2026 License: GPL-3.0 Imports: 27 Imported by: 0

Documentation

Overview

Package workflow provides the workflow definition loader and runtime.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ApplyInputDefaults added in v0.98.0 

func ApplyInputDefaults(declared []types.WorkflowInputDef, input types.KV) types.KV

ApplyInputDefaults copies input and fills missing keys from declared defaults.

func DetermineRuntimeType 

func DetermineRuntimeType(t *types.Task) string

func ExportYAML added in v0.98.0 

func ExportYAML(wf *types.WorkflowMetadata) ([]byte, error)

ExportYAML marshals a workflow definition to YAML exchange format.

func InputsToMaps added in v0.98.0 

func InputsToMaps(inputs []types.WorkflowInputDef) []map[string]any

InputsToMaps converts WorkflowInputDef slice to JSON-friendly maps for storage.

func LoadFile 

func LoadFile(path string) (*types.WorkflowMetadata, error)

LoadFile reads and parses a workflow YAML file from disk.

func MetadataFromRows added in v0.98.0 

func MetadataFromRows(rows WorkflowRows) (*types.WorkflowMetadata, error)

MetadataFromRows converts normalized store rows into WorkflowMetadata.

func ParseYAML 

func ParseYAML(data []byte) (*types.WorkflowMetadata, error)

ParseYAML unmarshals workflow YAML, validates structure, input types, and input.* template refs.

func ReloadTriggers added in v0.98.0 

func ReloadTriggers(ctx context.Context) error

ReloadTriggers reloads cron and webhook triggers on the wired Service.

func RetryToMap added in v0.98.0 

func RetryToMap(r *types.RetryConfig) map[string]any

RetryToMap converts RetryConfig to a JSON map for storage.

func SetReloadService added in v0.98.0 

func SetReloadService(svc *Service)

SetReloadService wires the package-level Service used by ReloadTriggers.

func ValidateDAG 

func ValidateDAG(tasks []types.WorkflowTask) error

func ValidateInputs added in v0.98.0 

func ValidateInputs(declared []types.WorkflowInputDef, input types.KV) error

ValidateInputs checks required fields and type constraints against declared input defs.

func WorkflowTaskToTask 

func WorkflowTaskToTask(wt types.WorkflowTask) (*types.Task, error)

Types

type ActionInfo 

type ActionInfo struct {
	Type         string
	Details      string
	IsCapability bool
	CapType      string
	Operation    string
}

func ParseAction 

func ParseAction(action string) ActionInfo

type Catalog added in v0.98.0 

type Catalog interface {
	DefinitionStore
	ApplyDefinition(ctx context.Context, meta *types.WorkflowMetadata) (*gen.Workflow, error)
	ListDefinitions(ctx context.Context) ([]*gen.Workflow, error)
	DeleteDefinitionByName(ctx context.Context, name string) error
	ListRunsByName(ctx context.Context, name string) ([]*gen.WorkflowRun, error)
}

Catalog loads and mutates workflow definitions stored in the database.

type CheckpointData 

type CheckpointData struct {
	StepIndex      int               `json:"step_index"`
	CompletedTasks map[string]bool   `json:"completed_tasks"`
	StepResults    map[string]string `json:"step_results"`
	Input          types.KV          `json:"input"`
	HeartbeatAt    time.Time         `json:"heartbeat_at"`
}

CheckpointData is the intermediate state saved at each workflow step boundary.

type DefinitionStore added in v0.98.0 

type DefinitionStore interface {
	GetMetadata(ctx context.Context, name string) (*types.WorkflowMetadata, error)
}

DefinitionStore loads workflow definitions by name for resume and DB-backed runs.

type Runner 

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

func NewRunner 

func NewRunner() *Runner

NewRunner creates a Runner without persistence. Use NewRunnerWithStore to enable run records.

func NewRunnerWithStore 

func NewRunnerWithStore(store WorkflowRunStore, auditor audit.Auditor, wc *metrics.WorkflowCollector, workflowFile, triggerType string) *Runner

NewRunnerWithStore creates a Runner that persists run and step records to the given store. workflowFile and triggerType are recorded in the run for audit; empty workflowFile becomes "db".

func (*Runner) Close 

func (r *Runner) Close() error

Close releases all executor engine resources (Docker clients, SSH connections, capability runtimes).

func (*Runner) Execute 

func (r *Runner) Execute(ctx context.Context, wf types.WorkflowMetadata, input types.KV, file string) error

func (*Runner) ResumeWorkflow 

func (r *Runner) ResumeWorkflow(ctx context.Context, runID int64) error

ResumeWorkflow resumes a previously failed or incomplete workflow run from its checkpoint. Definitions are loaded from DefinitionStore by workflow name (file-based resume is removed).

func (*Runner) Run 

func (r *Runner) Run(ctx context.Context, t *types.Task) error

func (*Runner) WithDefinitionStore added in v0.98.0 

func (r *Runner) WithDefinitionStore(ds DefinitionStore) *Runner

WithDefinitionStore attaches a DefinitionStore used to load definitions by name on resume.

func (*Runner) WithExistingRun added in v0.98.0 

func (r *Runner) WithExistingRun(run *gen.WorkflowRun) *Runner

WithExistingRun skips CreateRun and continues execution against the given run record.

func (*Runner) WithWorkflowID added in v0.98.0 

func (r *Runner) WithWorkflowID(id int64) *Runner

WithWorkflowID sets the workflow definition ID recorded on new runs (0 if unknown).

type Service added in v0.98.0 

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

Service orchestrates DB-backed workflow apply/run and trigger registration.

func ActiveService added in v0.98.0 

func ActiveService() *Service

ActiveService returns the package-level Service, or nil if not wired.

func NewService added in v0.98.0 

func NewService(catalog Catalog, runs WorkflowRunStore, auditor audit.Auditor, wc *metrics.WorkflowCollector) *Service

NewService creates a workflow Service.

func (*Service) ApplyYAML added in v0.98.0 

func (s *Service) ApplyYAML(ctx context.Context, data []byte) (*gen.Workflow, error)

ApplyYAML parses YAML, upserts the definition, and reloads triggers.

func (*Service) Delete added in v0.98.0 

func (s *Service) Delete(ctx context.Context, name string) error

Delete removes a workflow definition and reloads triggers.

func (*Service) Export added in v0.98.0 

func (s *Service) Export(ctx context.Context, name string) ([]byte, error)

Export returns YAML for a stored workflow.

func (*Service) Get added in v0.98.0 

func (s *Service) Get(ctx context.Context, name string) (*types.WorkflowMetadata, error)

Get returns workflow metadata by name.

func (*Service) List added in v0.98.0 

func (s *Service) List(ctx context.Context) ([]*gen.Workflow, error)

List returns workflow definition rows.

func (*Service) ListRuns added in v0.98.0 

func (s *Service) ListRuns(ctx context.Context, name string) ([]*gen.WorkflowRun, error)

ListRuns returns recent runs for a workflow name.

func (*Service) LookupWebhook added in v0.98.0 

func (s *Service) LookupWebhook(path string) (*WebhookEndpoint, bool)

LookupWebhook returns the webhook endpoint for a path (without /webhook/workflow/ prefix).

func (*Service) ReloadTriggers added in v0.98.0 

func (s *Service) ReloadTriggers(ctx context.Context) error

ReloadTriggers rebuilds cron jobs and webhook configs from enabled workflow definitions.

func (*Service) StartRunAsync added in v0.98.0 

func (s *Service) StartRunAsync(ctx context.Context, name, triggerType string, input types.KV) (int64, error)

StartRunAsync validates inputs, creates a run record, and executes the workflow in a goroutine. It returns the new run ID immediately.

func (*Service) Stop added in v0.98.0 

func (s *Service) Stop()

Stop shuts down the cron scheduler.

func (*Service) WebhookConfigs added in v0.98.0 

func (s *Service) WebhookConfigs() map[string]*WebhookEndpoint

WebhookConfigs returns a copy of the current webhook path map.

type WebhookEndpoint added in v0.98.0 

type WebhookEndpoint struct {
	WorkflowName string
	Config       *pipeline.WebhookConfig
}

WebhookEndpoint describes a registered workflow webhook trigger.

type WorkflowRows added in v0.98.0 

type WorkflowRows struct {
	Workflow *gen.Workflow
	Tasks    []*gen.WorkflowTask
	Triggers []*gen.WorkflowTrigger
}

WorkflowRows is the store-level projection used to rebuild WorkflowMetadata.

type WorkflowRunStore 

type WorkflowRunStore interface {
	CreateRun(ctx context.Context, workflowID int64, workflowName, workflowFile, triggerType string, triggerInfo, inputParams map[string]any) (*gen.WorkflowRun, error)
	UpdateRunStatus(ctx context.Context, runID int64, status int, errMsg string) error
	CreateStepRun(ctx context.Context, runID int64, stepID, stepName, action, actionType string, params map[string]any, attempt int) (*gen.WorkflowStepRun, error)
	UpdateStepRun(ctx context.Context, stepRunID int64, status int, result map[string]any, errMsg string, attempt int) error
	SaveCheckpoint(ctx context.Context, runID int64, data any) error
	GetIncompleteRuns(ctx context.Context) ([]*gen.WorkflowRun, error)
	GetCheckpoint(ctx context.Context, runID int64, target any) error
	GetRun(ctx context.Context, runID int64) (*gen.WorkflowRun, error)
	UpdateRunHeartbeat(ctx context.Context, runID int64) error
}

WorkflowRunStore persists workflow runs, step runs, and checkpoint data.

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.