Help us shape the future of Go! Take the 2026 Go Developer Survey

pipeline

package
v1.321.0 Latest Latest
Warning

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

 Go to latest Published: Oct 27, 2026 License: AGPL-3.0 Imports: 38 Imported by: 0

Documentation

Overview

SPDX-FileCopyrightText: 2025 Forkbomb BV

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-FileCopyrightText: 2025 Forkbomb BV

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-FileCopyrightText: 2026 Forkbomb BV

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-FileCopyrightText: 2025 Forkbomb BV

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-FileCopyrightText: 2025 Forkbomb BV

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-FileCopyrightText: 2025 Forkbomb BV

SPDX-License-Identifier: AGPL-3.0-or-later

Index

Constants

View Source 
const (
	GitHubPRCommentConfigKey                  = "github_pr_comment"
	GitHubPRCommentConfigRepositoryKey        = "repository"
	GitHubPRCommentConfigPullRequestNumberKey = "pull_request_number"
	GitHubPRCommentConfigCommitSHAKey         = "commit_sha"
	GitHubPRCommentConfigPipelineIDKey        = "pipeline_id"
	GitHubPRCommentConfigPipelineURLKey       = "pipeline_url"
	GitHubPRCommentConfigAppURLKey            = "app_url"
	GitHubPRCommentConfigSectionTitleKey      = "section_title"
)
View Source 
const (
	PipelineTaskQueue = "PipelineTaskQueue"

	PipelineMobileDevicesQuery = "GetPipelineMobileDevices"
)
View Source 
const (
	DefaultNameSpace                = "default"
	DefaultExecutionTimeout         = "24h"
	DefaultActivityScheduleTimeout  = "10m"
	DefaultActivityStartTimeout     = "5m"
	DefaultActivityHeartbeatTimeout = "30s"
	DefaultRetryMaxAttempts         = int32(5)
	DefaultRetryInitialInterval     = "5s"
	DefaultRetryMaxInterval         = "1m"
	DefaultRetryBackoff             = 2.0
)
View Source 
const CompletionNotificationConfigKey = "pipeline_completion_notification"

CompletionNotificationConfigKey marks queued runs that should send a web push notification when they reach a terminal state. The queue handler injects it for every run it starts, directly or through the semaphore.

Variables

This section is empty.

Functions

func ConformanceCheckCleanupHook 

func ConformanceCheckCleanupHook(
	ctx workflow.Context,
	wfDef *pipeline.WorkflowDefinition,
	_ *workflow.ActivityOptions,
	_ map[string]any,
	_ map[string]any,
	output *map[string]any,
) error

func ConformanceCheckSetupHook 

func ConformanceCheckSetupHook(
	ctx workflow.Context,
	wfDef *pipeline.WorkflowDefinition,
	config map[string]any,
	runData *map[string]any,
	_ *map[string]any,
	_ log.Logger,
) error

func DecodePayload 

func DecodePayload(s *pipeline.StepDefinition) (any, error)

func DeviceIDsWithGlobal 

func DeviceIDsWithGlobal(info PipelineDeviceInfo, globalDeviceID string) []string

func Execute 

func Execute(
	s *pipeline.StepDefinition,
	ctx workflow.Context,
	globalCfg map[string]any,
	dataCtx map[string]any,
	ao workflow.ActivityOptions,
) (any, error)

func ExecuteEventStepsOnError 

func ExecuteEventStepsOnError(
	ctx workflow.Context,
	eventSteps []*pipeline.OnErrorStepDefinition,
	stepInputs map[string]any,
	existingFailures []pipelineStepFailure,
	ao workflow.ActivityOptions,
	config map[string]any,
) []pipelineStepFailure

func ExecuteEventStepsOnSuccess 

func ExecuteEventStepsOnSuccess(
	ctx workflow.Context,
	eventSteps []*pipeline.OnSuccessStepDefinition,
	stepInputs map[string]any,
	existingFailures []pipelineStepFailure,
	ao workflow.ActivityOptions,
	config map[string]any,
) []pipelineStepFailure

func ExecuteOnError 

func ExecuteOnError(
	s *pipeline.OnErrorStepDefinition,
	ctx workflow.Context,
	globalCfg map[string]any,
	dataCtx map[string]any,
	ao workflow.ActivityOptions,
) (any, error)

func ExecuteOnSuccess 

func ExecuteOnSuccess(
	s *pipeline.OnSuccessStepDefinition,
	ctx workflow.Context,
	globalCfg map[string]any,
	dataCtx map[string]any,
	ao workflow.ActivityOptions,
) (any, error)

func ExecuteStep 

func ExecuteStep(
	id string,
	use string,
	with pipeline.StepInputs,
	activityOptions *pipeline.ActivityOptionsConfig,
	ctx workflow.Context,
	globalCfg map[string]any,
	dataCtx map[string]any,
	ao workflow.ActivityOptions,
) (any, error)

func ExtractPipelineOutput 

func ExtractPipelineOutput(dataCtx map[string]any) map[string]any

func GlobalDeviceIDFromConfig 

func GlobalDeviceIDFromConfig(config map[string]any) string

func MergePayload 

func MergePayload(dst, src *map[string]any) error

MergePayload merges all keys from src into dst recursively.

func MobileAutomationCleanupHook 

func MobileAutomationCleanupHook(
	ctx workflow.Context,
	_ *pipeline.WorkflowDefinition,
	ao *workflow.ActivityOptions,
	config map[string]any,
	runData map[string]any,
	output *map[string]any,
) error

func MobileAutomationSetupHook 

func MobileAutomationSetupHook(
	ctx workflow.Context,
	wfDef *pipeline.WorkflowDefinition,
	config map[string]any,
	runData *map[string]any,
	_ *map[string]any,
	_ log.Logger,
) error

func PipelineEvidenceSetupHook 

func PipelineEvidenceSetupHook(
	ctx workflow.Context,
	wfDef *pipelineinternal.WorkflowDefinition,
	_ map[string]any,
	runData *map[string]any,
	finalOutput *map[string]any,
	logger log.Logger,
) error

PipelineEvidenceSetupHook extracts issuer and verifier evidence for the pipeline's credential-offer and use-case-verification-deeplink steps in a single activity, before the steps run. The activity generates its own deeplinks: a verifier serves a request object only once, so evidence cannot reuse the deeplink the wallet receives.

func PipelineReportCleanupHook 

func PipelineReportCleanupHook(
	ctx workflow.Context,
	wfDef *pipelineinternal.WorkflowDefinition,
	ao *workflow.ActivityOptions,
	config map[string]any,
	runData map[string]any,
	finalOutput *map[string]any,
) error

func PrepareActivityOptions 

func PrepareActivityOptions(
	globalAO workflow.ActivityOptions,
	stepAO *pipeline.ActivityOptionsConfig,
) workflow.ActivityOptions

func PrepareWorkflowOptions 

func PrepareWorkflowOptions(rc pipeline.RuntimeConfig) pipeline.WorkflowOptions

func ResolveDeviceRecord 

func ResolveDeviceRecord(
	app core.App,
	deviceID string,
	deviceCache map[string]map[string]any,
) map[string]any

func ResolveDeviceRecords 

func ResolveDeviceRecords(
	app core.App,
	deviceIDs []string,
	deviceCache map[string]map[string]any,
) []map[string]any

func RunIdentifier 

func RunIdentifier(namespace, workflowID, runID string) string

RunIdentifier is the canonical pipeline run identifier used for runner workspaces and artifacts.

func SetConfigValue 

func SetConfigValue(config *map[string]any, key string, val any)

func SetPayloadValue 

func SetPayloadValue(payload *map[string]any, key string, val any) error

SetPayloadValue sets a key/value pair in the given payload map. If the key exists, it overwrites it; otherwise, it adds it.

func SetRunDataValue 

func SetRunDataValue(runData *map[string]any, key string, val any)

func ValidateDeviceIDYAML 

func ValidateDeviceIDYAML(yamlStr string) error

ValidateDeviceIDYAML enforces the mobile device_id configuration rule.

func ValidateFinallySteps 

func ValidateFinallySteps(finallyDef pipeline.FinallyDefinition) error

Types

type CleanupFunc 

type CleanupFunc func(
	ctx workflow.Context,
	wfDef *pipeline.WorkflowDefinition,
	ao *workflow.ActivityOptions,
	config map[string]any,
	runData map[string]any,
	output *map[string]any,
) error

type ConformanceTemplateActivity 

type ConformanceTemplateActivity struct {
	workflowengine.BaseActivity
}

ConformanceTemplateActivity reads a conformance check's config and StepCI templates outside workflow code.

func NewConformanceTemplateActivity 

func NewConformanceTemplateActivity() *ConformanceTemplateActivity

NewConformanceTemplateActivity returns the "Resolve conformance check template" activity.

func (*ConformanceTemplateActivity) Execute 

func (a *ConformanceTemplateActivity) Execute(
	_ context.Context,
	input workflowengine.ActivityInput,
) (workflowengine.ActivityResult, error)

Execute resolves the check's config template, suite extras and StepCI template.

func (*ConformanceTemplateActivity) Name 

func (a *ConformanceTemplateActivity) Name() string

Name returns the registered activity name.

type ConformanceTemplateActivityPayload 

type ConformanceTemplateActivityPayload struct {
	CheckID string `json:"check_id" validate:"required"`
	StepID  string `json:"step_id"`
}

ConformanceTemplateActivityPayload names the conformance check to resolve.

type DebugActivity 

type DebugActivity struct {
	workflowengine.BaseActivity
}

func NewDebugActivity 

func NewDebugActivity() *DebugActivity

func (*DebugActivity) Execute 

func (a *DebugActivity) Execute(
	ctx context.Context,
	input workflowengine.ActivityInput,
) (workflowengine.ActivityResult, error)

func (*DebugActivity) Name 

func (a *DebugActivity) Name() string

type DebugActivityPayload 

type DebugActivityPayload struct {
	Step    string         `json:"step"    yaml:"step"`
	Outputs map[string]any `json:"outputs" yaml:"outputs"`
	Inputs  map[string]any `json:"inputs"  yaml:"inputs"`
}

type PipelineDeviceInfo 

type PipelineDeviceInfo struct {
	DeviceIDs         []string
	NeedsGlobalDevice bool
}

func DeviceInfoFromDefinition 

func DeviceInfoFromDefinition(wfDef *pipeline.WorkflowDefinition) PipelineDeviceInfo

DeviceInfoFromDefinition collects mobile device IDs from a parsed workflow definition.

func ParsePipelineDeviceInfo 

func ParsePipelineDeviceInfo(yamlStr string) (PipelineDeviceInfo, error)

type PipelineWorkflow 

type PipelineWorkflow struct{}

func NewPipelineWorkflow 

func NewPipelineWorkflow() *PipelineWorkflow

func (PipelineWorkflow) GetOptions 

func (PipelineWorkflow) GetOptions() workflow.ActivityOptions

func (PipelineWorkflow) Name 

func (PipelineWorkflow) Name() string

func (*PipelineWorkflow) Start 

func (w *PipelineWorkflow) Start(
	inputYaml string,
	config map[string]any,
	memo map[string]any,
	pipelineIdentifier string,
) (workflowengine.WorkflowResult, error)

Start launches the pipeline workflow via Temporal client

func (*PipelineWorkflow) Workflow 

func (w *PipelineWorkflow) Workflow(
	ctx workflow.Context,
	input PipelineWorkflowInput,
) (result workflowengine.WorkflowResult, finalErr error)

Workflow executes all steps in the workflow definition sequentially

type PipelineWorkflowInput 

type PipelineWorkflowInput struct {
	WorkflowDefinition *pipeline.WorkflowDefinition `yaml:"workflow_definition" json:"workflow_definition"`
	WorkflowInput      workflowengine.WorkflowInput `yaml:"workflow_input"      json:"workflow_input"`

	Debug         bool           `yaml:"debug,omitempty"           json:"debug,omitempty"`
	ParentRunData map[string]any `yaml:"parent_run_data,omitempty" json:"parent_run_data,omitempty"`
	// ReturnOutputs lists the main step outputs the caller references, so the run returns
	// them in its result. Nil, as for a top-level run, returns no step outputs;
	// pipeline.AllStepOutputs returns all of them.
	ReturnOutputs []string `yaml:"return_outputs,omitempty" json:"return_outputs,omitempty"`
}

type SetupFunc 

type SetupFunc func(
	ctx workflow.Context,
	wfDef *pipeline.WorkflowDefinition,
	config map[string]any,
	runData *map[string]any,
	finalOutput *map[string]any,
	logger log.Logger,
) error

type StepDefinition 

type StepDefinition pipeline.StepDefinition

type WorkflowDefinition 

type WorkflowDefinition pipeline.WorkflowDefinition

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.