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

workflowengine

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: 24 Imported by: 0

Documentation

Overview

Package workflowengine is a package that provides a framework for defining and executing workflows. It includes interfaces and types for activities, activity input and output, and error handling.

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 (
	// HTTPHeaderContentType is the canonical HTTP Content-Type header key.
	HTTPHeaderContentType = "Content-Type"
	// MIMEApplicationJSON is the standard media type for JSON payloads.
	MIMEApplicationJSON = "application/json"
)
View Source 
const (
	MobileDeviceSemaphoreRunQueued   = mobiledevicesemaphore.MobileDeviceSemaphoreRunQueued
	MobileDeviceSemaphoreRunStarting = mobiledevicesemaphore.MobileDeviceSemaphoreRunStarting
	MobileDeviceSemaphoreRunRunning  = mobiledevicesemaphore.MobileDeviceSemaphoreRunRunning
	MobileDeviceSemaphoreRunFailed   = mobiledevicesemaphore.MobileDeviceSemaphoreRunFailed
	MobileDeviceSemaphoreRunCanceled = mobiledevicesemaphore.MobileDeviceSemaphoreRunCanceled
	MobileDeviceSemaphoreRunNotFound = mobiledevicesemaphore.MobileDeviceSemaphoreRunNotFound

	MobileDeviceSemaphoreDefaultNamespace = "default"
)
View Source 
const (
	// PipelineIdentifierSearchAttribute is the Temporal visibility field storing pipeline identifiers.
	PipelineIdentifierSearchAttribute = "PipelineIdentifier"
	// DeviceIdentifiersSearchAttribute is the Temporal visibility field storing device identifiers.
	DeviceIdentifiersSearchAttribute = "DeviceIdentifiers"
	ActionsSearchAttribute           = "ActionsID"
	VersionsSearchAttribute          = "VersionsID"
	CredentialsSearchAttribute       = "CredentialsID"
	UseCaseSearchAttribute           = "UseCaseID"
	ConformanceCheckSearchAttribute  = "ConformanceCheckID"
	CustomCheckSearchAttribute       = "CustomCheckID"
)
View Source 
const (
	TelemetryRootWorkflowIDKey    = "root_workflow_id"
	TelemetryRootRunIDKey         = "root_run_id"
	TelemetryParentWorkflowIDKey  = "parent_workflow_id"
	TelemetryParentRunIDKey       = "parent_run_id"
	TelemetryCurrentWorkflowIDKey = "current_workflow_id"
	TelemetryCurrentRunIDKey      = "current_run_id"
)
View Source 
const AppLogoConfigKey = "app_logo"

AppLogoConfigKey holds the URL of the application logo.

View Source 
const AppNameConfigKey = "app_name"

AppNameConfigKey holds the application name from PocketBase Settings.

View Source 
const AppURLConfigKey = "app_url"

AppURLConfigKey holds the public, user-facing base URL of the Credimi deployment (PocketBase Settings → App URL). Workflow code reads it only to build links shown to users (test runs, emails, PR comments, screenshots).

View Source 
const CollectPipelineStepFailuresConfigKey = "collect_pipeline_step_failures"
View Source 
const CredimiCapabilitiesMemoKey = "credimi_capabilities"

CredimiCapabilitiesMemoKey identifies Credimi-owned workflow capabilities in Temporal memo.

View Source 
const StepIDConfigKey = "step_id"

StepIDConfigKey carries the pipeline step ID in a step's activity or child workflow input config, so the run's history identifies which step recorded each result.

Variables

This section is empty.

Functions

func ActivityTelemetryConfig 

func ActivityTelemetryConfig(ctx workflow.Context, cfg map[string]any) map[string]string

func AppURL 

func AppURL(app core.App) string

AppURL returns the public application URL from PocketBase Settings.

func ApplyPipelineSearchAttributes 

func ApplyPipelineSearchAttributes(
	options *client.StartWorkflowOptions,
	pipelineIdentifier string,
	deviceIDs []string,
	entityIDs EntityIDs,
)

ApplyPipelineSearchAttributes ensures StartWorkflowOptions include pipeline visibility attributes.

func AsBool 

func AsBool(v any) bool

func AsMap 

func AsMap(v any) map[string]any

func AsSliceOfMaps 

func AsSliceOfMaps(val any) []map[string]any

func AsSliceOfStrings 

func AsSliceOfStrings(val any) []string

func AsString 

func AsString(v any) string

func BuildCalendarSpec 

func BuildCalendarSpec(mode ScheduleMode) []client.ScheduleCalendarSpec

func CustomSearchAttributeTypes 

func CustomSearchAttributeTypes() map[string]enumspb.IndexedValueType

CustomSearchAttributeTypes returns the custom search attributes every Credimi namespace must register, keyed by name.

func DecodeOutput 

func DecodeOutput[T any](output any) (T, error)

DecodeOutput decodes a JSON-shaped value, such as an activity output, into T through a JSON round-trip. Unlike DecodePayload it does not validate the result, so T may be any type, including slices.

func DecodePayload 

func DecodePayload[T any](input any) (T, error)

DecodePayload decodes a JSON payload into a given type. It returns the decoded object and an error if the decoding fails. This allows for the decoding of arbitrary JSON payloads into Go types.

func DecodeStringPayload 

func DecodeStringPayload(p *commonpb.Payload) string

DecodeStringPayload decodes a Temporal payload holding a string (such as a memo field) through the Credimi data converter. Payloads the converter cannot decode fall back to their raw data with surrounding quotes trimmed.

func ExtractOutputFromError 

func ExtractOutputFromError(err error) map[string]any

func FormatWorkflowFailureReason 

func FormatWorkflowFailureReason(failure WorkflowError) string

func IsServerOwnedConfigKey 

func IsServerOwnedConfigKey(key string) bool

IsServerOwnedConfigKey reports whether key may be set in workflow config only by the server. User pipeline YAML, step config, and rerun bodies must never set these keys.

func MergeTelemetryConfig 

func MergeTelemetryConfig(ctx workflow.Context, cfg map[string]any) map[string]any

func NewAppError 

func NewAppError(failure WorkflowError) error

func NewMissingConfigError 

func NewMissingConfigError(key string, metadata *WorkflowRunMetadata) error

func NewMissingOrInvalidPayloadError 

func NewMissingOrInvalidPayloadError(err error, runMetadata *WorkflowRunMetadata) error

func NewStepCIOutputError 

func NewStepCIOutputError(field string, output any, metadata *WorkflowRunMetadata) error

func NewWorkflowCancellationError 

func NewWorkflowCancellationError(metadata *WorkflowRunMetadata) error

func NewWorkflowError 

func NewWorkflowError(err error, metadata *WorkflowRunMetadata) error

func NormalizePipelineIdentifier 

func NormalizePipelineIdentifier(identifier string) string

NormalizePipelineIdentifier trims whitespace and leading/trailing slashes from identifiers.

func PipelineTypedSearchAttributes 

func PipelineTypedSearchAttributes(
	pipelineIdentifier string,
	deviceIDs []string,
	entityIDs EntityIDs,
) temporal.SearchAttributes

PipelineTypedSearchAttributes returns typed search attributes for scheduled workflow actions.

func RecordHeartbeat 

func RecordHeartbeat(ctx context.Context, details ...any)

RecordHeartbeat records an activity heartbeat. It is a no-op outside an activity context, so helpers shared with plain-context callers can use it.

func RunCommandWithCancellation 

func RunCommandWithCancellation(
	ctx context.Context,
	cmd *exec.Cmd,
	heartbeatInterval time.Duration,
) error

func RunPageURL 

func RunPageURL(config map[string]any, execution workflow.Execution) string

RunPageURL returns the Credimi page of a run, linked from workflow errors and pipeline notifications. It is empty when the config carries no app_url: workflows that build user-facing links check app_url themselves, the others do not need it.

func StartHeartbeat 

func StartHeartbeat(ctx context.Context, every time.Duration, details ...any) (stop func())

StartHeartbeat records a heartbeat with details every interval until the returned stop is called or ctx is done.

func StepOutputFromActivityResult 

func StepOutputFromActivityResult(kind OutputKind, result ActivityResult) any

StepOutputFromActivityResult converts an activity result into the value a pipeline stores as the step's outputs, according to the step's registered output kind.

func StringifyConfig 

func StringifyConfig(cfg map[string]any) map[string]string

StringifyConfig converts config values to strings for an ActivityInput, skipping nil values.

func WaitForUpdateResult 

func WaitForUpdateResult[T any](
	ctx context.Context,
	c client.Client,
	workflowID, runID, updateName string,
) (T, error)

WaitForUpdateResult sends updateName to the workflow and waits for the handler's result. The handler blocks until the data is ready, so ctx bounds the wait.

func WithAppConfig 

func WithAppConfig(app core.App, config map[string]any) map[string]any

WithAppConfig writes the server-owned app config keys into config, replacing any existing values, and returns it. A nil config is allocated.

func WorkflowFailureMessageFromHistory 

func WorkflowFailureMessageFromHistory(failure *failurepb.Failure) string

Types

type Activity 

type Activity interface {
	Name() string
	NewActivityError(failure ActivityError) error
	NewNonRetryableActivityError(failure ActivityError) error
	NewMissingOrInvalidPayloadError(err error) error
}

BaseActivity provides the common interface for all activities.

type ActivityError 

type ActivityError struct {
	Code         string         `json:"code"`
	Summary      string         `json:"summary"`
	Message      string         `json:"message,omitempty"`
	ActivityName string         `json:"activityName,omitempty"`
	Category     string         `json:"category,omitempty"`
	Details      map[string]any `json:"details,omitempty"`
}

type ActivityInput 

type ActivityInput struct {
	Payload any               `json:"payload,omitempty"`
	Config  map[string]string `json:"config,omitempty"`
	Secrets map[string]any    `json:"secrets,omitempty"`
}

ActivityInput represents the input to an activity, including payload and configuration. The payload is a map of string keys to any type of value.

type ActivityResult 

type ActivityResult struct {
	Output  any            `json:"output,omitempty"`
	Log     []string       `json:"log,omitempty"`
	Secrets map[string]any `json:"secrets,omitempty"`
}

ActivityResult represents the result of an activity execution, including output, errors, and logs. It is designed to be extensible, allowing for different types of output and error handling.

type BaseActivity 

type BaseActivity struct {
	Name string
}

BaseActivity provides a default implementation of the Activity interface.

func (*BaseActivity) NewActivityError 

func (a *BaseActivity) NewActivityError(
	failure ActivityError,
) error

func (*BaseActivity) NewCodedError 

func (a *BaseActivity) NewCodedError(code string, retryable bool, err error) error

NewCodedError builds an activity error for the errorcodes key code with err as its message; it is non-retryable unless retryable is true.

func (*BaseActivity) NewMissingOrInvalidPayloadError 

func (a *BaseActivity) NewMissingOrInvalidPayloadError(err error) error

func (*BaseActivity) NewNonRetryableActivityError 

func (a *BaseActivity) NewNonRetryableActivityError(
	failure ActivityError,
) error

func (*BaseActivity) NewNonRetryableActivityErrorWithCause 

func (a *BaseActivity) NewNonRetryableActivityErrorWithCause(
	failure ActivityError,
	cause error,
) error

NewNonRetryableActivityErrorWithCause is NewNonRetryableActivityError with a cause that errors.Is and errors.As can match.

type ConfigurableActivity 

type ConfigurableActivity interface {
	Activity
	Configure(input *ActivityInput) error
}

ConfigurableActivity defines an activity that can be configured.

type CredimiCapabilities 

type CredimiCapabilities struct {
	Logs bool `json:"logs"`
	QR   bool `json:"qr"`
}

CredimiCapabilities describes optional product capabilities of one workflow execution.

type EntityIDs 

type EntityIDs struct {
	Actions           []string `json:"actions,omitempty"`
	Versions          []string `json:"versions,omitempty"`
	Credentials       []string `json:"credentials,omitempty"`
	UseCases          []string `json:"use_cases,omitempty"`
	ConformanceChecks []string `json:"conformance_checks,omitempty"`
	CustomChecks      []string `json:"custom_checks,omitempty"`
}

type ExecutableActivity 

type ExecutableActivity interface {
	Activity
	Execute(ctx context.Context, input ActivityInput) (ActivityResult, error)
}

ExecutableActivity defines an activity that can be executed.

type OutputKind 

type OutputKind int

OutputKind represents the expected type of an activity output. 

const (
	OutputAny OutputKind = iota
	OutputString
	OutputMap
	OutputArrayOfString
	OutputArrayOfMap
	OutputBool
)

type ScheduleMode 

type ScheduleMode struct {
	Mode string `json:"mode"` // "daily", "weekly", "monthly"
	Day  *int   `json:"day"`  // 0-6 for weekly (0=Sunday), 0-30 for monthly
}

ScheduleMode represents the scheduling configuration

func ParseScheduleMode 

func ParseScheduleMode(calendars []client.ScheduleCalendarSpec) ScheduleMode

type Workflow 

type Workflow interface {
	Workflow(ctx workflow.Context, input WorkflowInput) (WorkflowResult, error)
	ExecuteWorkflow(ctx workflow.Context, input WorkflowInput) (WorkflowResult, error)
	Name() string
	GetOptions() workflow.ActivityOptions
}

Workflow defines the interface for a workflow, including its execution, name, and options.

type WorkflowError 

type WorkflowError struct {
	Code         string         `json:"code"`
	Summary      string         `json:"summary"`
	Message      string         `json:"message,omitempty"`
	ActivityName string         `json:"activityName,omitempty"`
	WorkflowName string         `json:"workflowName,omitempty"`
	WorkflowID   string         `json:"workflowId,omitempty"`
	RunID        string         `json:"runId,omitempty"`
	Namespace    string         `json:"namespace,omitempty"`
	TemporalUI   string         `json:"temporalUI,omitempty"`
	Details      map[string]any `json:"details,omitempty"`
}

func ParseWorkflowError 

func ParseWorkflowError(err error) WorkflowError

func ParseWorkflowFailure 

func ParseWorkflowFailure(failure *failurepb.Failure) WorkflowError

type WorkflowFn 

type WorkflowFn func(ctx workflow.Context, input WorkflowInput) (WorkflowResult, error)

WorkflowFn represents a function that executes a workflow.

func BuildWorkflow 

func BuildWorkflow(
	w Workflow,
) WorkflowFn

type WorkflowInput 

type WorkflowInput struct {
	Payload         any                       `json:"payload,omitempty"`
	Config          map[string]any            `json:"config,omitempty"`
	Secrets         map[string]any            `json:"secrets,omitempty"`
	ActivityOptions *workflow.ActivityOptions `json:"activityOptions,omitempty"`
	RunMetadata     *WorkflowRunMetadata      `json:"runMetadata,omitempty"`
}

WorkflowInput represents the input data required to start a workflow.

func WithCredimiCapabilities 

func WithCredimiCapabilities(
	input WorkflowInput,
	capabilities CredimiCapabilities,
) WorkflowInput

WithCredimiCapabilities returns an input with authoritative capabilities merged into its memo.

type WorkflowResult 

type WorkflowResult struct {
	WorkflowID    string `json:"workflowId,omitempty"`
	WorkflowRunID string `json:"workflowRunId,omitempty"`
	Author        string `json:"author,omitempty"`
	Message       string `json:"message,omitempty"`
	Errors        any    `json:"errors,omitempty"`
	Output        any    `json:"output,omitempty"`
	Log           any    `json:"log,omitempty"`
}

WorkflowResult represents the result of a workflow execution, including a message, errors, and a log.

func StartWorkflowWithOptions 

func StartWorkflowWithOptions(
	namespace string,
	options client.StartWorkflowOptions,
	name string,
	input WorkflowInput,
) (result WorkflowResult, err error)

func WaitForWorkflowResult 

func WaitForWorkflowResult(
	ctx context.Context,
	c client.Client,
	workflowID, runID string,
) (WorkflowResult, error)

Wait for final workflow result

type WorkflowRunInfo 

type WorkflowRunInfo struct {
	Name      string         `json:"name"`
	TaskQueue string         `json:"taskQueue"`
	Input     WorkflowInput  `json:"input,omitempty"`
	Memo      map[string]any `json:"memo,omitempty"`
}

type WorkflowRunMetadata 

type WorkflowRunMetadata struct {
	WorkflowName string `json:"workflowName,omitempty"`
	WorkflowID   string `json:"workflowId,omitempty"`
	RunID        string `json:"runId,omitempty"`
	Namespace    string `json:"namespace,omitempty"`
	TemporalUI   string `json:"temporalUI,omitempty"`
}

Source Files

View all Source files

Directories

Path Synopsis
Package activities is a package that provides activities for the workflow engine.
 Package activities is a package that provides activities for the workflow engine.
Package hooks provides functionality to manage and run Temporal workers for executing workflows and activities in a distributed system.
 Package hooks provides functionality to manage and run Temporal workers for executing workflows and activities in a distributed system.
SPDX-FileCopyrightText: 2025 Forkbomb BV
 SPDX-FileCopyrightText: 2025 Forkbomb BV
SPDX-FileCopyrightText: 2025 Forkbomb BV
 SPDX-FileCopyrightText: 2025 Forkbomb BV
Package pipelinehistory rebuilds a pipeline run's step outputs from its Temporal history, so activities can reference step outputs instead of receiving copies of them.
 Package pipelinehistory rebuilds a pipeline run's step outputs from its Temporal history, so activities can reference step outputs instead of receiving copies of them.
SPDX-FileCopyrightText: 2025 Forkbomb BV
 SPDX-FileCopyrightText: 2025 Forkbomb BV
Package semaphoreclient is the Temporal client side of the mobile-device semaphore workflow, shared by the HTTP handlers and the enqueue activity.
 Package semaphoreclient is the Temporal client side of the mobile-device semaphore workflow, shared by the HTTP handlers and the enqueue activity.
SPDX-FileCopyrightText: 2025 Forkbomb BV
 SPDX-FileCopyrightText: 2025 Forkbomb BV

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.