Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

workflow

package

Go to main page

Versions in this module

v0

v0.16.0

May 7, 2026

v0.16.0-rc2

May 4, 2026

v0.16.0-rc1

Apr 27, 2026

v0.15.0

Apr 15, 2026

Changes in this version

type StepTransitionRequest

+ AuthenticatedEmail string

+ AuthenticatedGroups []string

+ AuthenticatedIdentifiers []string

+ AuthenticatedUserID string

+ Signer *evidencesvc.SignerContext

v0.15.0-rc3

Apr 14, 2026

v0.15.0-rc2

Apr 9, 2026

v0.15.0-rc1

Apr 7, 2026

v0.14.1

Mar 26, 2026

v0.14.0

Mar 26, 2026

v0.13.0

Mar 20, 2026

v0.13.0-rc3

Mar 20, 2026

v0.13.0-rc2

Mar 19, 2026

v0.13.0-rc1

Mar 12, 2026

v0.12.0

Feb 23, 2026

Changes in this version

+ const JobTypeExecuteWorkflow

+ const JobTypeScheduleWorkflows

+ var ErrInvalidAssignee = errors.New("invalid assignee")

+ var ErrInvalidStepTransition = errors.New("invalid step transition")

+ var ErrReassignmentNotAllowed = errors.New("step status does not allow reassignment")

+ var ErrWorkflowExecutionAlreadyExists = errors.New("workflow execution already exists for instance and period")

+ func CalculateDueDate(scheduledTime time.Time, gracePeriodDays int) time.Time

+ func GeneratePeriodLabel(cadence string, t time.Time) string

+ func JobInsertOptionsForScheduler() *river.InsertOpts

+ func JobInsertOptionsForWorkflow() *river.InsertOpts

+ func ResolveGraceDays(instance *workflows.WorkflowInstance, defaultDays int) int

+ type Assignee struct

+ ID string

+ Type string

+ type AssignmentService struct

+ func NewAssignmentService(roleAssignmentService RoleAssignmentServiceInterface, ...) *AssignmentService

+ func (s *AssignmentService) BulkReassignByRole(ctx context.Context, workflowExecutionID uuid.UUID, roleName string, ...) (*BulkReassignResult, error)

+ func (s *AssignmentService) ReassignStep(ctx context.Context, stepExecutionID uuid.UUID, newAssignee Assignee, ...) error

+ func (s *AssignmentService) ResolveStepAssignees(ctx context.Context, instance *workflows.WorkflowInstance, ...) (map[uuid.UUID]Assignee, error)

+ type AssignmentServiceInterface interface

+ ResolveStepAssignees func(ctx context.Context, instance *workflows.WorkflowInstance, ...) (map[uuid.UUID]Assignee, error)

+ type BulkReassignResult struct

+ ExecutionID uuid.UUID

+ ReassignedCount int

+ ReassignedStepExecIDs []uuid.UUID

+ RoleName string

+ type DAGExecutor struct

+ func NewDAGExecutor(stepExecutionService StepExecutionServiceInterface, ...) *DAGExecutor

+ func (e *DAGExecutor) CheckAutomaticTriggers(ctx context.Context, stepExecutionID *uuid.UUID) error

+ func (e *DAGExecutor) InitializeWorkflow(ctx context.Context, workflowExecutionID *uuid.UUID) error

+ func (e *DAGExecutor) ProcessStepCompletion(ctx context.Context, stepExecutionID *uuid.UUID) error

+ func (e *DAGExecutor) ProcessStepFailure(ctx context.Context, stepExecutionID *uuid.UUID) error

+ func (e *DAGExecutor) SetEvidenceIntegration(evidenceIntegration *EvidenceIntegration)

+ type EvidenceIntegration struct

+ func NewEvidenceIntegration(db *gorm.DB, logger *zap.SugaredLogger) *EvidenceIntegration

+ func (e *EvidenceIntegration) AddExecutionCompletionEvidence(ctx context.Context, workflowExecutionID *uuid.UUID) error

+ func (e *EvidenceIntegration) AddExecutionFailureEvidence(ctx context.Context, workflowExecutionID *uuid.UUID) error

+ func (e *EvidenceIntegration) AddStepStartedEvidence(ctx context.Context, stepExecutionID *uuid.UUID) error

+ func (e *EvidenceIntegration) AddWorkflowExecutionEvidence(ctx context.Context, workflowExecutionID *uuid.UUID, status string) error

+ func (e *EvidenceIntegration) GetOrCreateExecutionStream(ctx context.Context, workflowExecutionID *uuid.UUID) (*relational.Evidence, error)

+ func (e *EvidenceIntegration) GetOrCreateInstanceStream(ctx context.Context, workflowInstanceID *uuid.UUID) (*relational.Evidence, error)

+ type EvidenceRequirement struct

+ Description string

+ Required bool

+ Type string

+ type EvidenceSubmission struct

+ Description string

+ EvidenceID *uuid.UUID

+ EvidenceType string

+ FileContent string

+ FileHash string

+ FilePath string

+ FileSize int64

+ MediaType string

+ Metadata string

+ Name string

+ type ExecuteWorkflowArgs struct

+ TriggeredBy string

+ TriggeredByID string

+ WorkflowExecutionID uuid.UUID

+ func (ExecuteWorkflowArgs) Kind() string

+ func (ExecuteWorkflowArgs) Timeout() time.Duration

+ type ExecutionMetrics struct

+ AverageStepDuration time.Duration

+ Duration time.Duration

+ ExecutionID uuid.UUID

+ LongestStepDuration time.Duration

+ TotalSteps int

+ type ExecutionStatus struct

+ BlockedSteps int

+ CancelledSteps int

+ CompletedAt *time.Time

+ CompletedSteps int

+ ExecutionID uuid.UUID

+ FailedAt *time.Time

+ FailedSteps int

+ FailureReason string

+ InProgressSteps int

+ OverdueSteps int

+ PendingSteps int

+ StartedAt *time.Time

+ Status string

+ TotalSteps int

+ type Manager struct

+ func NewManager(riverClient RiverClient, ...) *Manager

+ func (m *Manager) CancelExecution(ctx context.Context, executionID *uuid.UUID, reason string) (*workflows.WorkflowExecution, error)

+ func (m *Manager) GetExecutionMetrics(ctx context.Context, executionID *uuid.UUID) (*ExecutionMetrics, error)

+ func (m *Manager) GetExecutionStatus(ctx context.Context, executionID *uuid.UUID) (*ExecutionStatus, error)

+ func (m *Manager) ListExecutions(ctx context.Context, workflowInstanceID *uuid.UUID, limit, offset int) ([]*workflows.WorkflowExecution, error)

+ func (m *Manager) RetryExecution(ctx context.Context, executionID *uuid.UUID) (*workflows.WorkflowExecution, error)

+ func (m *Manager) StartWorkflowExecution(ctx context.Context, workflowInstanceID *uuid.UUID, opts StartWorkflowOptions) (*workflows.WorkflowExecution, error)

+ type NotificationEnqueuer interface

+ EnqueueWorkflowExecutionFailed func(ctx context.Context, execution *workflows.WorkflowExecution) error

+ EnqueueWorkflowTaskAssigned func(ctx context.Context, stepExecution *workflows.StepExecution) error

+ type OverdueService struct

+ func NewOverdueService(db *gorm.DB, workflowExecutionService *workflows.WorkflowExecutionService, ...) *OverdueService

+ func (s *OverdueService) CheckFailedExecutions(ctx context.Context) (int, error)

+ func (s *OverdueService) CheckOverdueExecutions(ctx context.Context) (int, error)

+ func (s *OverdueService) CheckOverdueSteps(ctx context.Context) (int, error)

+ type RiverClient interface

+ InsertMany func(ctx context.Context, params []river.InsertManyParams) ([]*rivertype.JobInsertResult, error)

+ type RoleAssignmentServiceInterface interface

+ FindAssigneeForRole func(instanceID *uuid.UUID, roleName string) (*workflows.RoleAssignment, error)

+ GetByWorkflowInstanceID func(instanceID *uuid.UUID) ([]workflows.RoleAssignment, error)

+ type ScheduleWorkflowsArgs struct

+ func (ScheduleWorkflowsArgs) Kind() string

+ type StartWorkflowOptions struct

+ DueDate *time.Time

+ PeriodLabel string

+ TriggeredBy string

+ TriggeredByID string

+ type StepExecutionAssignmentService interface

+ ReassignWithTx func(tx *gorm.DB, id *uuid.UUID, assignedToType, assignedToID string, ...) error

+ type StepExecutionServiceInterface interface

+ CanUnblock func(id *uuid.UUID) (bool, error)

+ Create func(stepExecution *workflows.StepExecution) error

+ Fail func(id *uuid.UUID, reason string) error

+ GetByID func(id *uuid.UUID) (*workflows.StepExecution, error)

+ GetByWorkflowExecutionID func(executionID *uuid.UUID) ([]workflows.StepExecution, error)

+ Unblock func(id *uuid.UUID) error

+ UpdateStatus func(ctx context.Context, id *uuid.UUID, status string) error

+ type StepStatus string

+ const StatusBlocked

+ const StatusCancelled

+ const StatusCompleted

+ const StatusFailed

+ const StatusInProgress

+ const StatusOverdue

+ const StatusPending

+ const StatusSkipped

+ func (s StepStatus) String() string

+ type StepStatusCounts struct

+ Blocked int

+ Cancelled int

+ Completed int

+ Failed int

+ InProgress int

+ Overdue int

+ Pending int

+ Skipped int

+ func CountStepStatuses(steps []workflows.StepExecution) StepStatusCounts

+ func (c StepStatusCounts) AllTerminal() bool

+ type StepTransitionRequest struct

+ Evidence []EvidenceSubmission

+ Notes string

+ Status string

+ UserID string

+ UserType string

+ type StepTransitionService struct

+ func NewStepTransitionService(stepExecutionService StepExecutionServiceInterface, ...) *StepTransitionService

+ func (s *StepTransitionService) CanUserTransitionStep(stepExecutionID *uuid.UUID, userID, userType string) (bool, error)

+ func (s *StepTransitionService) FailStep(ctx context.Context, stepExecutionID *uuid.UUID, reason string) error

+ func (s *StepTransitionService) GetEvidenceRequirements(stepExecutionID *uuid.UUID) ([]workflows.EvidenceRequirement, error)

+ func (s *StepTransitionService) GetStepExecutionService() *workflows.StepExecutionService

+ func (s *StepTransitionService) TransitionStepStatus(ctx context.Context, stepExecutionID *uuid.UUID, ...) error

+ type WorkflowDefinitionServiceInterface interface

+ GetByID func(id *uuid.UUID) (*workflows.WorkflowDefinition, error)

+ type WorkflowExecutionServiceInterface interface

+ Cancel func(id *uuid.UUID) error

+ Create func(execution *workflows.WorkflowExecution) error

+ Fail func(id *uuid.UUID, reason string) error

+ FailIfNotTerminal func(ctx context.Context, id *uuid.UUID, reason string) (bool, error)

+ GetByID func(id *uuid.UUID) (*workflows.WorkflowExecution, error)

+ GetByWorkflowInstanceID func(instanceID *uuid.UUID) ([]workflows.WorkflowExecution, error)

+ UpdateStatus func(ctx context.Context, id *uuid.UUID, status string) error

+ type WorkflowExecutionWorker struct

+ func NewWorkflowExecutionWorker(executor *DAGExecutor, evidenceIntegration *EvidenceIntegration, ...) *WorkflowExecutionWorker

+ func (w *WorkflowExecutionWorker) Work(ctx context.Context, job *river.Job[ExecuteWorkflowArgs]) error

+ type WorkflowInstanceServiceInterface interface

+ AdvanceSchedule func(ctx context.Context, id *uuid.UUID) error

+ GetByID func(id *uuid.UUID) (*workflows.WorkflowInstance, error)

+ GetDueInstances func(ctx context.Context) ([]workflows.WorkflowInstance, error)

+ UpdateSchedule func(ctx context.Context, id *uuid.UUID, nextSchedule time.Time) error

+ type WorkflowSchedulerWorker struct

+ func NewWorkflowSchedulerWorker(manager *Manager, workflowInstanceService WorkflowInstanceServiceInterface, ...) *WorkflowSchedulerWorker

+ func (w *WorkflowSchedulerWorker) Work(ctx context.Context, job *river.Job[ScheduleWorkflowsArgs]) error

+ type WorkflowStepDefinitionServiceInterface interface

+ GetByID func(id *uuid.UUID) (*workflows.WorkflowStepDefinition, error)

+ GetByWorkflowDefinitionID func(workflowDefID *uuid.UUID) ([]workflows.WorkflowStepDefinition, error)

+ GetDependencies func(stepID *uuid.UUID) ([]workflows.WorkflowStepDefinition, error)

+ GetDependentSteps func(stepID *uuid.UUID) ([]workflows.WorkflowStepDefinition, error)