workflows

package
v0.8.13 Latest Latest
Warning

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

 Go to latest Published: Aug 7, 2026 License: AGPL-3.0 Imports: 12 Imported by: 0

Documentation

Overview

Package workflows contains the storage layer for Epic 64 (triggers & workflows).

Pure data-access on the seven Epic 64 tables. Crypto (encrypt/decrypt of webhook HMAC secrets) is NOT done here — handlers encrypt before calling CreateWebhook; the webhook receiver decrypts after calling GetWebhook. This mirrors the pkg/secrets/mcp_store.go split exactly.

Index

Constants

This section is empty.

Variables

View Source 
var ErrConcurrentRun = errors.New("workflow: a run is already in flight for this workflow")

ErrConcurrentRun is returned by CreateWorkflowRun when the single-in-flight partial unique index (uq_workflow_run_single_inflight) rejects the insert because another queued/running run exists for the same workflow_id. Handlers map this to 409 "already running".

View Source 
var ErrDedupConflict = errors.New("workflow: duplicate webhook delivery")

ErrDedupConflict is returned by RecordWebhookDelivery when the (webhook_id, dedup_key) UNIQUE constraint rejects a duplicate delivery. Handlers map this to 200 "duplicate" (idempotent re-delivery).

View Source 
var ErrNotFound = errors.New("workflow: not found")

ErrNotFound is returned by Get/Update/Delete when the row does not exist (or the caller's owner scope does not match). Handlers map this to 404.

Functions

func RecordNodeDuration added in v0.8.12 

func RecordNodeDuration(nodeType string, durationSeconds float64)

RecordNodeDuration records per-node execution timing.

func RecordRunFinished added in v0.8.12 

func RecordRunFinished(status, errorCode, ownerType string, durationSeconds float64)

RecordRunFinished records a run's terminal status + duration.

func RecordRunStarted added in v0.8.12 

func RecordRunStarted()

RecordRunStarted increments the concurrent gauge.

func RecordSchedulerTick added in v0.8.12 

func RecordSchedulerTick(durationSeconds float64)

RecordSchedulerTick records scheduler tick timing.

func RecordTriggerFire added in v0.8.12 

func RecordTriggerFire(source, status string)

RecordTriggerFire records a trigger fire event.

func RecordWebhookDelivery added in v0.8.12 

func RecordWebhookDelivery(webhookID, status string)

RecordWebhookDelivery records a webhook delivery.

Types

type AgentNodeData 

type AgentNodeData struct {
	Agent                   string          `json:"agent,omitempty"`
	Prompt                  string          `json:"prompt"`
	System                  string          `json:"system,omitempty"`
	OutputSchema            json.RawMessage `json:"outputSchema,omitempty"`
	EnforceStructuredOutput bool            `json:"enforceStructuredOutput,omitempty"`
	Session                 string          `json:"session,omitempty"`
	SessionID               string          `json:"sessionId,omitempty"`
}

AgentNodeData is the typed shape of SpecNode.Data for agent nodes.

type ConditionCase 

type ConditionCase struct {
	ID         string `json:"id"`
	Expression string `json:"expression"`
}

ConditionCase is one branch of a condition node.

type ConditionNodeData 

type ConditionNodeData struct {
	Conditions []ConditionCase `json:"conditions"`
}

ConditionNodeData is the typed shape of SpecNode.Data for condition nodes.

type DefaultsBlock 

type DefaultsBlock struct {
	MaxAttempts *int   `json:"maxAttempts,omitempty"`
	Timeout     string `json:"timeout,omitempty"`
}

DefaultsBlock carries workflow-level defaults that are merged into each node's config (node-level wins). Only maxAttempts and timeout may be defaulted; behavioral fields must be per-node.

type HTTPNodeData 

type HTTPNodeData struct {
	Method  string            `json:"method,omitempty"`
	URL     string            `json:"url"`
	Headers map[string]string `json:"headers,omitempty"`
	Body    string            `json:"body,omitempty"`
	Timeout string            `json:"timeout,omitempty"`
}

HTTPNodeData is the typed shape of SpecNode.Data for http nodes.

type ScriptNodeData 

type ScriptNodeData struct {
	Language string `json:"language"`
	Handler  string `json:"handler"`
}

ScriptNodeData is the typed shape of SpecNode.Data for script nodes.

type Spec 

type Spec struct {
	Nodes []SpecNode `json:"nodes"`
	Edges []SpecEdge `json:"edges"`
}

Spec is the parsed and validated DAG stored in workflows.spec_json. It is the execution-ready form: defaults merged, node types validated, edges checked for cycles/dangling/condition-coverage.

func ParseSpec 

func ParseSpec(raw json.RawMessage) (*Spec, error)

ParseSpec parses raw JSON bytes into a Spec. Does NOT validate — call ValidateSpec for that. Returns an error only on JSON parse failure.

type SpecEdge 

type SpecEdge struct {
	Source       string `json:"source"`
	Target       string `json:"target"`
	SourceHandle string `json:"sourceHandle,omitempty"`
}

SpecEdge is a directed edge in the DAG. SourceHandle carries the condition-branch id (for condition nodes) or is empty (for all other types).

type SpecNode 

type SpecNode struct {
	ID          string          `json:"id"`
	Type        string          `json:"type"`
	Data        json.RawMessage `json:"data"`
	MaxAttempts int             `json:"maxAttempts,omitempty"`
	Timeout     string          `json:"timeout,omitempty"`
}

SpecNode is a single node in the workflow DAG.

type Store 

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

Store is the Postgres-backed data-access layer for Epic 64. Construct once at app boot via NewStore; share across handlers. Methods are safe for concurrent use (pgxpool serializes connections internally).

func NewStore 

func NewStore(pool *pgxpool.Pool) *Store

NewStore constructs a Store backed by the given pgxpool.

func (*Store) ClaimQueuedRuns 

func (s *Store) ClaimQueuedRuns(ctx context.Context, limit int) ([]*WorkflowRunRow, error)

ClaimQueuedRuns selects up to limit queued runs and atomically marks them running, returning the claimed rows. Uses FOR UPDATE SKIP LOCKED so multiple controller replicas (or multiple goroutines) can claim concurrently without contention. The reconciler polls this on each tick.

func (*Store) CountTriggersByOwner 

func (s *Store) CountTriggersByOwner(ctx context.Context, ownerType, ownerID string) (int, error)

CountTriggersByOwner returns the number of triggers owned by (ownerType, ownerID).

func (*Store) CountWorkflowsByOwner 

func (s *Store) CountWorkflowsByOwner(ctx context.Context, ownerType, ownerID string) (int, error)

CountWorkflowsByOwner returns the number of workflows owned by (ownerType, ownerID). Used for quota enforcement (workflows.maxPerUser / workflows.maxPerOrg).

func (*Store) CreateNodeRun 

func (s *Store) CreateNodeRun(ctx context.Context, row *WorkflowNodeRunRow) error

CreateNodeRun inserts a workflow_node_runs row.

func (*Store) CreateTrigger 

func (s *Store) CreateTrigger(ctx context.Context, row *TriggerRow) error

CreateTrigger inserts a row into triggers. The caller supplies a pre-generated UUID. For webhook triggers, an accompanying webhooks row is created via CreateWebhook in the same transaction by the handler (US-64.5).

func (*Store) CreateTriggerFire 

func (s *Store) CreateTriggerFire(ctx context.Context, row *TriggerFireRow) error

CreateTriggerFire inserts a trigger_fires row.

func (*Store) CreateWebhook 

func (s *Store) CreateWebhook(ctx context.Context, row *WebhookRow) error

CreateWebhook inserts a row into webhooks. The caller supplies a pre-encrypted SecretCipher (KEK for org scope, session-DEK for user scope — exact mcp_servers crypto envelope pattern). trigger_id has a 1:1 UNIQUE constraint.

func (*Store) CreateWorkflow 

func (s *Store) CreateWorkflow(ctx context.Context, row *WorkflowRow) error

CreateWorkflow inserts a row into workflows. The caller supplies a pre-generated UUID. spec_json must already be parsed + validated (US-64.4).

func (*Store) CreateWorkflowRun 

func (s *Store) CreateWorkflowRun(ctx context.Context, row *WorkflowRunRow) error

CreateWorkflowRun inserts a new run row. The single-in-flight partial unique index (uq_workflow_run_single_inflight) may reject the insert with a unique violation — mapped to ErrConcurrentRun so the handler can return 409 + Retry-After. Callers should use CreateWorkflowRunTx (with the fire row) for webhook-triggered runs; this method is for manual runs.

func (*Store) CreateWorkflowRunWithFire 

func (s *Store) CreateWorkflowRunWithFire(ctx context.Context, fire *TriggerFireRow, run *WorkflowRunRow) error

CreateWorkflowRunWithFire atomically inserts a trigger_fires row AND a workflow_runs row in a single transaction. If the run insert hits the single-in-flight unique violation, the whole tx rolls back (no orphan fire row claiming "fired" with no run). The caller then commits a separate trigger_fires row with status='skipped' + reason='already_running' and returns 409 + Retry-After.

This is the v1 correctness pattern for webhook fire-row + run-create atomicity (design Edge Case 4 / Webhook Security atomicity bullet).

func (*Store) DeleteTrigger 

func (s *Store) DeleteTrigger(ctx context.Context, ownerType, ownerID, triggerID string) error

DeleteTrigger deletes a trigger by ID scoped to (ownerType, ownerID). FK cascades handle webhooks + trigger_fires. Returns ErrNotFound.

func (*Store) DeleteWorkflow 

func (s *Store) DeleteWorkflow(ctx context.Context, ownerType, ownerID, workflowID string) error

DeleteWorkflow deletes a workflow by ID scoped to (ownerType, ownerID). FK cascades handle workflow_runs (ON DELETE CASCADE). Returns ErrNotFound if the workflow doesn't exist or the caller's scope doesn't match.

func (*Store) DisableTrigger 

func (s *Store) DisableTrigger(ctx context.Context, triggerID string) error

DisableTrigger sets enabled=false (called when the circuit breaker trips).

func (*Store) GetTrigger 

func (s *Store) GetTrigger(ctx context.Context, ownerType, ownerID, triggerID string) (*TriggerRow, error)

GetTrigger returns a single trigger by ID scoped to (ownerType, ownerID), or ErrNotFound.

func (*Store) GetTriggerByID added in v0.8.12 

func (s *Store) GetTriggerByID(ctx context.Context, triggerID string) (*TriggerRow, error)

GetTriggerByID returns a trigger by its UUID without owner scoping. Used by the webhook receiver where the trigger_id is authoritative (unguessable UUID).

func (*Store) GetWebhook 

func (s *Store) GetWebhook(ctx context.Context, webhookID string) (*WebhookRow, error)

GetWebhook returns a webhook by its own ID (the public webhook_id in the receiver URL path). Used by POST /api/v1/hooks/:webhook_id.

func (*Store) GetWebhookByTriggerID 

func (s *Store) GetWebhookByTriggerID(ctx context.Context, triggerID string) (*WebhookRow, error)

GetWebhookByTriggerID returns the webhook config for a trigger, or ErrNotFound. Used by the webhook receiver to fetch the HMAC secret + IP allowlist.

func (*Store) GetWorkflow 

func (s *Store) GetWorkflow(ctx context.Context, ownerType, ownerID, workflowID string) (*WorkflowRow, error)

GetWorkflow returns a single workflow by ID scoped to (ownerType, ownerID), or ErrNotFound if not found / wrong scope.

func (*Store) GetWorkflowRun 

func (s *Store) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRunRow, error)

GetWorkflowRun returns a single run by ID, or ErrNotFound. Not scoped — run IDs are unguessable UUIDs and the caller has already authorized via the workflow's owner scope.

func (*Store) HasInFlightRun 

func (s *Store) HasInFlightRun(ctx context.Context, workflowID string) (bool, error)

HasInFlightRun reports whether a non-terminal run exists for the workflow. Used for fast-path rejection at the API layer (the partial unique index is the authoritative gate; this is an early-reject optimization).

func (*Store) IncrementTriggerFailures 

func (s *Store) IncrementTriggerFailures(ctx context.Context, triggerID string) (int, error)

IncrementTriggerFailures increments consecutive_failures and returns the new value. If the new value crosses auto_disable_after, the caller sets enabled=false.

func (*Store) ListDueCronTriggers 

func (s *Store) ListDueCronTriggers(ctx context.Context, now time.Time, limit int) ([]*TriggerRow, error)

ListDueCronTriggers returns enabled cron triggers whose next_fire_at <= now, ordered by next_fire_at ASC. Used by the scheduler goroutine (US-64.9).

func (*Store) ListNodeRuns 

func (s *Store) ListNodeRuns(ctx context.Context, workflowRunID string) ([]*WorkflowNodeRunRow, error)

ListNodeRuns returns all node runs for a workflow run, ordered by started_at ASC.

func (*Store) ListTriggerFires 

func (s *Store) ListTriggerFires(ctx context.Context, triggerID string, limit, offset int) ([]*TriggerFireRow, error)

ListTriggerFires returns recent fires for a trigger, paginated by fired_at DESC.

func (*Store) ListTriggers 

func (s *Store) ListTriggers(ctx context.Context, ownerType, ownerID string) ([]*TriggerRow, error)

ListTriggers returns all triggers owned by (ownerType, ownerID), ordered by created_at ASC.

func (*Store) ListWorkflowRuns 

func (s *Store) ListWorkflowRuns(ctx context.Context, workflowID string, limit, offset int) ([]*WorkflowRunRow, error)

ListWorkflowRuns returns runs for a workflow, paginated by created_at DESC. limit must be > 0; offset is the number of rows to skip.

func (*Store) ListWorkflows 

func (s *Store) ListWorkflows(ctx context.Context, ownerType, ownerID string) ([]*WorkflowRow, error)

ListWorkflows returns all workflows owned by (ownerType, ownerID), ordered by created_at ASC. Never decrypts — display fields only.

func (*Store) RecordWebhookDelivery 

func (s *Store) RecordWebhookDelivery(ctx context.Context, webhookID, dedupKey string) error

RecordWebhookDelivery inserts a dedup row. Returns ErrDedupConflict if a row with the same (webhook_id, dedup_key) already exists — the caller returns 200 "duplicate" in that case. The unique constraint name is webhook_deliveries_webhook_dedup_uniq.

func (*Store) ResetTriggerFailures 

func (s *Store) ResetTriggerFailures(ctx context.Context, triggerID string) error

ResetTriggerFailures resets consecutive_failures to 0 (called on first success).

func (*Store) UpdateNodeRunStatus 

func (s *Store) UpdateNodeRunStatus(ctx context.Context, nodeRunID, status string, output json.RawMessage, branch *string, errorCode *string, errMsg json.RawMessage) error

UpdateNodeRunStatus sets status + terminal fields. On a terminal transition, finished_at is set. branch is set only on condition nodes (matched edge handle).

func (*Store) UpdateTrigger 

func (s *Store) UpdateTrigger(ctx context.Context, ownerType, ownerID, triggerID string, upd *TriggerUpdate) (*TriggerRow, error)

UpdateTrigger updates an existing trigger scoped to (ownerType, ownerID). source_type is NOT mutable (not in TriggerUpdate). Partial update via pointer fields: nil preserves existing values. Returns the updated row or ErrNotFound.

func (*Store) UpdateTriggerFireTimestamps 

func (s *Store) UpdateTriggerFireTimestamps(ctx context.Context, triggerID string, lastFiredAt time.Time, nextFireAt *time.Time) error

UpdateTriggerFireTimestamps sets last_fired_at + advances next_fire_at for cron triggers after a fire. nextFireAt may be nil to clear it.

func (*Store) UpdateWorkflow 

func (s *Store) UpdateWorkflow(ctx context.Context, ownerType, ownerID, workflowID string, upd *WorkflowUpdate) (*WorkflowRow, error)

UpdateWorkflow updates an existing workflow scoped to (ownerType, ownerID). Partial update via WorkflowUpdate pointer fields: nil preserves existing values. Returns the updated row or ErrNotFound.

func (*Store) UpdateWorkflowRunStatus 

func (s *Store) UpdateWorkflowRunStatus(ctx context.Context, runID, status string, errorCode *string, errMsg json.RawMessage, output json.RawMessage) error

UpdateWorkflowRunStatus sets status + terminal fields. On a terminal transition (succeeded/failed/canceled/timed_out), finished_at is set. error_code + error are nullable (null on success).

type TriggerFireRow 

type TriggerFireRow struct {
	ID            string
	TriggerID     string
	SourceType    string
	InputEnvelope json.RawMessage
	ActionType    string
	ActionResult  json.RawMessage
	Status        string
	FiredAt       time.Time
	CompletedAt   *time.Time
}

TriggerFireRow is the DB row shape for trigger_fires.

type TriggerRow 

type TriggerRow struct {
	ID                  string
	OwnerType           string
	OwnerID             string
	Name                string
	Description         string
	Enabled             bool
	SourceType          string
	SourceConfig        json.RawMessage
	TargetType          string
	TargetConfig        json.RawMessage
	ConsecutiveFailures int
	AutoDisableAfter    int
	LastFiredAt         *time.Time
	NextFireAt          *time.Time
	CreatedAt           time.Time
	UpdatedAt           time.Time
}

TriggerRow is the DB row shape for triggers.

type TriggerUpdate 

type TriggerUpdate struct {
	Name             *string
	Description      *string
	Enabled          *bool
	SourceConfig     json.RawMessage
	TargetType       *string
	TargetConfig     json.RawMessage
	AutoDisableAfter *int
}

TriggerUpdate carries only the fields a partial update may change. Pointer fields: nil means "keep existing". source_type is NOT in this struct — it's immutable after create (the source defines the trigger's identity).

type ValidationError 

type ValidationError struct {
	Code   string `json:"code"`
	NodeID string `json:"nodeId,omitempty"`
	Detail string `json:"detail,omitempty"`
}

ValidationError describes a single problem found during DAG validation.

func ValidateSpec 

func ValidateSpec(spec *Spec, predecessorSchemas map[string]json.RawMessage, defaults DefaultsBlock) []ValidationError

ValidateSpec validates a workflow DAG spec and applies defaults merging. It mutates the spec in place (defaults applied to nodes that omit maxAttempts/timeout) ONLY when validation passes without structural errors (dangling edges, duplicate IDs, cycles). Non-structural errors (unreachable nodes, missing branch edges) do NOT prevent defaults merging — the spec is still mutated. Callers should discard the spec if any errors are returned. predecessorSchemas maps node ID → that node's outputSchema (JSON Schema bytes), used for expr-lang type-checking condition expressions against the upstream node's output shape. nil/empty schemas skip type-checking for that predecessor.

Returns a slice of ValidationErrors (empty if valid).

func (ValidationError) Error 

func (e ValidationError) Error() string

type WebhookRow 

type WebhookRow struct {
	ID                string
	TriggerID         string
	SecretCipher      []byte
	KeyVersion        int
	AllowedIPs        []string
	IdempotencyMode   string
	IdempotencyHeader string
	CreatedAt         time.Time
}

WebhookRow is the DB row shape for webhooks. SecretCipher is the encrypted HMAC secret (crypto envelope); callers must decrypt before verifying signatures.

type WorkflowNodeRunRow 

type WorkflowNodeRunRow struct {
	ID            string
	WorkflowRunID string
	NodeID        string
	NodeType      string
	Status        string
	Attempt       int
	Input         json.RawMessage
	Output        json.RawMessage
	Branch        *string
	ErrorCode     *string
	Error         json.RawMessage
	StartedAt     time.Time
	FinishedAt    *time.Time
}

WorkflowNodeRunRow is the DB row shape for workflow_node_runs.

type WorkflowRow 

type WorkflowRow struct {
	ID                string
	OwnerType         string
	OwnerID           string
	Name              string
	Slug              string
	Description       string
	SpecYAML          string
	SpecJSON          json.RawMessage
	InputSchema       json.RawMessage
	TargetWorkspaceID *string
	Status            string
	Defaults          json.RawMessage
	CreatedAt         time.Time
	UpdatedAt         time.Time
}

WorkflowRow is the DB row shape for workflows.

type WorkflowRunRow 

type WorkflowRunRow struct {
	ID            string
	WorkflowID    string
	SpecSnapshot  json.RawMessage
	Input         json.RawMessage
	Output        json.RawMessage
	Status        string
	ErrorCode     *string
	Error         json.RawMessage
	TriggerID     *string
	TriggerFireID *string
	WorkspaceID   string
	StartedAt     *time.Time
	FinishedAt    *time.Time
	CreatedAt     time.Time
	UpdatedAt     time.Time
}

WorkflowRunRow is the DB row shape for workflow_runs.

type WorkflowUpdate 

type WorkflowUpdate struct {
	Name              *string
	Slug              *string
	Description       *string
	SpecYAML          *string
	SpecJSON          json.RawMessage
	InputSchema       json.RawMessage
	TargetWorkspaceID *string // nil = keep; non-nil "" = clear (set NULL); non-nil "uuid" = set
	Status            *string
	Defaults          json.RawMessage
}

WorkflowUpdate carries only the fields a partial update may change. Pointer fields: nil means "keep existing". This mirrors the API DTO pattern (UpdateWorkflowRequest) so the handler can pass fields through directly. The zero-value WorkflowUpdate preserves every field (no changes).

Source Files

View all Source files

Directories

Path Synopsis
Package exprlang compiles condition expressions for workflow DAG validation.
 Package exprlang compiles condition expressions for workflow DAG validation.
Package scriptwrap executes user-authored inline script handlers (Python, Node) inside the workspace sandbox.
 Package scriptwrap executes user-authored inline script handlers (Python, Node) inside the workspace sandbox.

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.