pipeline

package
v1.322.0 Latest Latest
Warning

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

 Go to latest Published: Nov 4, 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.