automations

package
v1.801.466 Latest Latest
Warning

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

Go to latest
Published: Aug 5, 2026 License: Apache-2.0 Imports: 38 Imported by: 0

Documentation

Overview

Package automations is workflows that run themselves, on a schedule or a webhook.

An org authors flows — a trigger and a tree of connector actions — and this runs them durably at /v1/automations and keeps their run history.

It composes THREE existing seams rather than reinventing them:

  • apps/integrations — per-org connector credentials (KMS-sealed). Connectors reach a token ONLY through integrations.TokenFor, never KMS directly.
  • cloud.EmbeddedTasks — the ONE shared in-process durable engine. A flow runs as a durable workflow in the OWNER's namespace (engine.go).
  • apps/principal — the ONE tenant gate. Every data handler resolves the org from principal.Org; a client-forged X-Org-Id with no bearer is refused.

Surface (all under /v1/automations/*, all org-gated except the compose-root generic GET /v1/automations/health). A ✓ marks a TYPED op — one registry entry from which the REST route, the OpenAPI operation, the MCP tool, the CLI command and every generated SDK method follow; the four without one are routes and nothing else, and each names the wire fact that keeps it there at its handler:

✓ GET    /v1/automations/connectors             the connector catalogue (org-gated)
✓ GET    /v1/automations/pieces                 back-compat alias of /connectors
✓ POST   /v1/automations/connectors/:id/run     execute ONE connector action (apps/connectorruntime,
                                                sub-mounted below — typed there, and the reason this
                                                prefix serves NINETEEN routes while routes() registers 18)
✓ GET    /v1/automations/flows                  list flows
✓ POST   /v1/automations/flows                  create a flow (+ initial draft version)
✓ GET    /v1/automations/flows/:id              flow + latest version
✓ PATCH  /v1/automations/flows/:id              update flow metadata
✓ DELETE /v1/automations/flows/:id              delete a flow (+ versions + runs)
✓ GET    /v1/automations/flows/:id/versions     list versions
✓ POST   /v1/automations/flows/:id/versions     create a draft version
  POST   /v1/automations/flows/:id/operations   apply a FlowOperation — TWO Out shapes
✓ POST   /v1/automations/flows/:id/run          start a durable run
✓ POST   /v1/automations/flows/:id/enable       enable (POLLING → CreateSchedule)
✓ POST   /v1/automations/flows/:id/disable      disable (POLLING → DeleteSchedule)
✓ GET    /v1/automations/runs                   list runs
✓ GET    /v1/automations/runs/:id               run detail (refreshed from engine)
  POST   /v1/automations/runs/:id/resume        resume a paused run — arbitrary JSON in
  POST   /v1/automations/hooks/:source/:event   inbound event sink — raw-byte dedupe

Index

Constants

View Source
const (
	TriggerTypePiece = "PIECE_TRIGGER"
	TriggerTypeEmpty = "EMPTY"

	ActionTypePiece  = "PIECE"
	ActionTypeCode   = "CODE"
	ActionTypeBranch = "BRANCH"
	ActionTypeLoop   = "LOOP_ON_ITEMS"
	ActionTypeRouter = "ROUTER"
)

FlowTriggerType / FlowActionType mirror the discriminants of the step nodes.

View Source
const LatestFlowSchemaVersion = "21"

LatestFlowSchemaVersion is the shared schema stamp new versions carry, matching flow-version.ts LATEST_FLOW_SCHEMA_VERSION.

Variables

View Source
var (
	// ErrNoOrg — Deliver called without a server-verified org. The org is the sole tenant
	// key, so a missing/invalid one must start NOTHING (fail-closed).
	ErrNoOrg = errors.New("automations: Deliver requires a server-verified org")
	// ErrRateLimited — the org's durable per-window run-start budget is exhausted.
	ErrRateLimited = errors.New("automations: per-org run budget exceeded")
	// ErrBusy — the org's concurrent run-start slots are full (front-door burst bound).
	ErrBusy = errors.New("automations: too many concurrent run-starts for this org")
)
View Source
var ErrEngineNotReady = errors.New("automations: engine not ready")

ErrEngineNotReady is returned when cloud.EmbeddedTasks() is still nil (the engine is wired after MountAll). Handlers render it as 503 "automation engine not ready".

View Source
var (

	// ErrNotMounted is returned by InvokeTool before the subsystem has mounted.
	ErrNotMounted = errors.New("automations: not mounted")
)

Dispatch sentinels let each door map a failure onto its own error convention (the JSON-RPC door to -32601/-32005, the in-process door to a returned error) without duplicating the resolve/limit/run logic.

Functions

func Deliver

func Deliver(ctx context.Context, org string, ev TriggerEvent) (started int, err error)

Deliver fires every ENABLED flow in org whose WEBHOOK/APP_WEBHOOK trigger matches (ev.Source, ev.Name), starting each as a durable run with ev.Payload threaded in, and returns how many runs it started.

TENANT ISOLATION — org is the ONLY tenant key and MUST be a server-verified tenant (resolved from a signed principal, or a signature-verified webhook via integrations.OrgForExternalID) — NEVER a client-supplied field. MatchTriggers leads its index with org, so an event delivered for org A can only ever start org A's flows.

IDEMPOTENCY — with a DedupeKey the run id is deterministic in (org, flowID, Source, Name, DedupeKey), and CreateRunIfAbsent is the atomic gate: a re-delivered event finds the row already present and skips it, so a flow fires at most once per event. Without a DedupeKey every delivery is its own run.

FAIL-CLOSED — an empty/invalid org, or the engine not being ready, returns an error and starts nothing. A flow that was disabled or deleted since it subscribed is skipped (the index is a fast lookup; the flow's live status is authoritative).

func InvokeTool

func InvokeTool(ctx context.Context, org, name string, args map[string]any) (any, error)

InvokeTool runs a single MCP tool as principal `org`, in-process — the same dispatch, credential scope, per-org concurrency bound, metering, and audit as POST /v1/tools/call, minus the HTTP hop. It is the seam a sibling subsystem (the Business AI guide) uses to act through the per-principal MCP plane. `org` MUST be the caller's VALIDATED principal.Org: the dispatch pins every credential and effect to it, so an in-process caller can never exceed that org's authority. Returns ErrNotMounted before the subsystem has mounted.

func Mount

func Mount(app cloud.Router, deps cloud.Deps) error

Mount wires /v1/automations/* onto app per HIP-0106. Complex flavour: it keeps a package global (mounted) for Shutdown and the engine's run hooks, so it constructs the Service value directly.

func RecordRunEndActivity

func RecordRunEndActivity(ctx context.Context, in RunEndInput) error

RecordRunEndActivity records a run's terminal status so listRuns reflects it without a getRun refresh. Best-effort; mounted==nil is a no-op.

func RecordRunStartActivity

func RecordRunStartActivity(ctx context.Context, in RunStartInput) error

RecordRunStartActivity persists the run row + meters + audits EXACTLY once (idempotent by run id via the store's metered-flag claim). The single owner of run bookkeeping for every entrypoint. mounted==nil (an engine-only test without Mount) is a no-op — nothing to record against.

func Shutdown

func Shutdown(_ context.Context) error

Shutdown closes the store. Idempotent — safe when nothing is mounted.

func ToolExists

func ToolExists(name string) bool

ToolExists reports whether name resolves to a registered "<connector>_<action>" tool — the cheap pre-check a caller uses to give an honest error before dispatch.

Types

type Action

type Action struct {
	Name        string
	DisplayName string
	Description string
	Props       []PropSpec
	Run         func(ctx context.Context, rc RunContext) (any, error)
}

Action is one invocable capability of a connector. Props declares its inputs (drives the catalogue + the MCP tool input schema). Run is the side-effecting body; it MUST handle every error explicitly and MUST NOT leak another org's data.

type AddActionRequest

type AddActionRequest struct {
	ParentStep string      `json:"parentStep"`
	Action     *FlowAction `json:"action"`
}

AddActionRequest inserts Action after the step named ParentStep (empty ⇒ append to the end of the chain, or set as the trigger's first action).

type Catalog

type Catalog struct {
	ConnectorCount int                 `json:"connectorCount"`
	Connectors     []ConnectorMetadata `json:"connectors"`
}

Catalog is the browse catalogue served at GET /v1/automations/connectors. It is the go:embed'd catalog/catalog.json, seeded with the Tier-A connectors and later overwritten with the full 700+ connector set at this EXACT schema.

type ChangeNameRequest

type ChangeNameRequest struct {
	DisplayName string `json:"displayName"`
}

ChangeNameRequest renames the version.

type ChangeStatusRequest

type ChangeStatusRequest struct {
	Status FlowStatus `json:"status"`
}

ChangeStatusRequest flips ENABLED/DISABLED.

type Connector

type Connector struct {
	Name        string
	DisplayName string
	AuthType    string // "none" | "bot_token" | "oauth2" — for the catalogue card
	AuthReq     bool
	Actions     map[string]*Action
	Triggers    map[string]*Trigger
}

Connector is one connectable capability provider. Name == the integrations provider id where credentials apply.

type ConnectorAction

type ConnectorAction struct {
	Name        string     `json:"name"`
	DisplayName string     `json:"displayName"`
	Description string     `json:"description"`
	Props       []PropSpec `json:"props"`
}

ConnectorAction / ConnectorTrigger are the catalogue's action/trigger descriptors.

type ConnectorAuth

type ConnectorAuth struct {
	Type     string `json:"type"`
	Required bool   `json:"required"`
}

ConnectorAuth is the catalogue's auth descriptor: which credential a connector needs ("none" for core, "oauth2"/"bot_token" for a connected provider) and whether it is required.

type ConnectorMetadata

type ConnectorMetadata struct {
	Name        string             `json:"name"`
	DisplayName string             `json:"displayName"`
	Description string             `json:"description"`
	LogoURL     string             `json:"logoUrl"`
	Version     string             `json:"version"`
	Categories  []string           `json:"categories"`
	Auth        ConnectorAuth      `json:"auth"`
	Actions     []ConnectorAction  `json:"actions"`
	Triggers    []ConnectorTrigger `json:"triggers"`
}

ConnectorMetadata is one catalogue entry. The catalog WIRE schema models a connector's actions/triggers as arrays, so this Go shape uses arrays to match the wire exactly, as the contract requires.

type ConnectorTrigger

type ConnectorTrigger struct {
	Name        string     `json:"name"`
	DisplayName string     `json:"displayName"`
	Description string     `json:"description"`
	Strategy    string     `json:"strategy"`
	Props       []PropSpec `json:"props"`
}

type DeleteActionRequest

type DeleteActionRequest struct {
	Names []string `json:"names"`
}

DeleteActionRequest removes the named steps and relinks the chain.

type Flow

type Flow struct {
	ID                 string          `json:"id"`
	Org                string          `json:"projectId"` // projectId == org (server-derived)
	ExternalID         string          `json:"externalId"`
	FolderID           string          `json:"folderId"`
	Status             FlowStatus      `json:"status"`
	PublishedVersionID string          `json:"publishedVersionId"`
	Metadata           json.RawMessage `json:"metadata,omitempty"`
	Created            int64           `json:"created"`
	Updated            int64           `json:"updated"`
}

Flow is an org-scoped automation. projectId IS the org (the TS `projectId` field), always server-derived from principal.Org and NEVER trusted from a request body — a caller can never author a flow into another tenant.

type FlowAction

type FlowAction struct {
	Name        string       `json:"name"`
	Type        string       `json:"type"` // PIECE | CODE | ROUTER | LOOP_ON_ITEMS
	DisplayName string       `json:"displayName"`
	Valid       bool         `json:"valid"`
	Skip        bool         `json:"skip,omitempty"`
	Settings    StepSettings `json:"settings"`
	NextAction  *FlowAction  `json:"nextAction,omitempty"`
}

FlowAction is an action node (actions/action.ts). The engine walks the linear NextAction chain; the tree discriminants (ROUTER/LOOP) are modeled for contract fidelity but branch/loop execution is out of Phase-1 scope (see operations.go).

type FlowOperation

type FlowOperation struct {
	Type    FlowOperationType `json:"type"`
	Request json.RawMessage   `json:"request"`
}

FlowOperation is the discriminated-union envelope the builder POSTs to /v1/automations/flows/:id/operations: a type + an opaque request the apply switch decodes per type.

type FlowOperationType

type FlowOperationType string

FlowOperationType mirrors operations/index.ts FlowOperationType — the builder's edit-op discriminants. The engine applies the linear-chain subset (see operations.go); tree-restructuring ops are honestly rejected, never faked.

const (
	OpAddAction     FlowOperationType = "ADD_ACTION"
	OpUpdateAction  FlowOperationType = "UPDATE_ACTION"
	OpDeleteAction  FlowOperationType = "DELETE_ACTION"
	OpMoveAction    FlowOperationType = "MOVE_ACTION"
	OpUpdateTrigger FlowOperationType = "UPDATE_TRIGGER"
	OpChangeName    FlowOperationType = "CHANGE_NAME"
	OpChangeStatus  FlowOperationType = "CHANGE_STATUS"
)

type FlowRun

type FlowRun struct {
	ID            string        `json:"id"`
	Org           string        `json:"-"`
	FlowID        string        `json:"flowId"`
	FlowVersionID string        `json:"flowVersionId"`
	WorkflowID    string        `json:"-"`
	Status        FlowRunStatus `json:"status"`
	StartTime     int64         `json:"startTime"`
	FinishTime    int64         `json:"finishTime"`
	Created       int64         `json:"created"`
	Updated       int64         `json:"updated"`
}

FlowRun is an execution record (flow-run/flow-run.ts). WorkflowID (the tasks engine handle) equals ID and is internal — resume/describe address the engine through it, scoped to the org's namespace.

type FlowRunInput

type FlowRunInput struct {
	Owner         string     `json:"owner"`
	FlowID        string     `json:"flowId"`
	FlowVersionID string     `json:"flowVersionId"`
	RunID         string     `json:"runId"`
	Steps         []FlowStep `json:"steps"`
	// Trigger is the inbound event payload that fired the flow — nil for a manual /run
	// or a bare cron tick, set by Deliver for a WEBHOOK/APP_WEBHOOK event. The workflow
	// seeds it as outputs["trigger"], so an action can reference the event that fired the
	// flow with {{trigger.field}}.
	Trigger map[string]any `json:"trigger,omitempty"`
	// Depth is the run's causation depth — how many in-platform trigger→action hops led
	// here (0 = an external origin). It rides in from TriggerEvent.Depth and lets an
	// in-platform cycle terminate at maxCausationDepth instead of amplifying.
	Depth int `json:"depth,omitempty"`
}

FlowRunInput is the durable workflow's typed input. Owner is the VALIDATED org set at flow-start from principal.Org — the SOLE credential scope and the cross-tenant isolation boundary; it is NEVER read from a request body. Steps is the flattened trigger→action chain (side-effecting steps in order).

type FlowRunResult

type FlowRunResult struct {
	RunID   string         `json:"runId"`
	Status  FlowRunStatus  `json:"status"`
	Steps   int            `json:"steps"`
	Outputs map[string]any `json:"outputs"`
}

FlowRunResult is the workflow's terminal result: the final status plus every step's output, keyed by step name (the threaded outputs).

func FlowRunWorkflow

func FlowRunWorkflow(ctx workflow.Context, in FlowRunInput) (FlowRunResult, error)

FlowRunWorkflow walks the flow's flattened step chain. Each side-effecting step runs as a retried activity (ExecuteStepActivity); a core.wait_for_approval step is a durable PAUSE that blocks on the resume signal. Prior step outputs are threaded so later steps can reference them ({{step.field}}). Deterministic: it executes steps strictly in order, Get-ing each before dispatching the next.

It also owns run bookkeeping (MED-1): the per-EXECUTION run id is the workflow id (workflow.GetInfo) — manual runs set it to the run id; a scheduled cron mints a fresh one per tick — so a run-start activity persists+meters+audits the run EXACTLY once, and a run-end activity records the terminal status.

type FlowRunStatus

type FlowRunStatus string

FlowRunStatus mirrors execution/flow-execution.ts FlowRunStatus (the subset the engine emits; the full enum has memory/log/quota terminal states we map onto FAILED).

const (
	RunRunning   FlowRunStatus = "RUNNING"
	RunSucceeded FlowRunStatus = "SUCCEEDED"
	RunFailed    FlowRunStatus = "FAILED"
	RunPaused    FlowRunStatus = "PAUSED"
	RunQueued    FlowRunStatus = "QUEUED"
	RunCanceled  FlowRunStatus = "CANCELED"
	RunTimeout   FlowRunStatus = "TIMEOUT"
)

type FlowStatus

type FlowStatus string

FlowStatus mirrors flows/flow.ts FlowStatus.

const (
	FlowEnabled  FlowStatus = "ENABLED"
	FlowDisabled FlowStatus = "DISABLED"
)

type FlowStep

type FlowStep struct {
	Name       string         `json:"name"`
	PieceName  string         `json:"pieceName"`
	ActionName string         `json:"actionName"`
	Input      map[string]any `json:"input"`
}

FlowStep is one resolved, executable step in the flattened chain.

type FlowTrigger

type FlowTrigger struct {
	Name        string          `json:"name"`
	Type        string          `json:"type"` // PIECE_TRIGGER | EMPTY
	DisplayName string          `json:"displayName"`
	Valid       bool            `json:"valid"`
	Strategy    TriggerStrategy `json:"strategy,omitempty"`
	Settings    StepSettings    `json:"settings"`
	NextAction  *FlowAction     `json:"nextAction,omitempty"`
}

FlowTrigger is the root of the step tree (triggers/trigger.ts). Strategy drives enable/disable (POLLING → CreateSchedule); NextAction chains into the action list.

type FlowVersion

type FlowVersion struct {
	ID            string           `json:"id"`
	Org           string           `json:"-"`
	FlowID        string           `json:"flowId"`
	DisplayName   string           `json:"displayName"`
	Trigger       *FlowTrigger     `json:"trigger"`
	Valid         bool             `json:"valid"`
	State         FlowVersionState `json:"state"`
	SchemaVersion string           `json:"schemaVersion"`
	Created       int64            `json:"created"`
	Updated       int64            `json:"updated"`
}

FlowVersion is one editable revision of a flow: a display name plus the root trigger of the step tree. Org is the isolation key (never serialized).

type FlowVersionState

type FlowVersionState string

FlowVersionState mirrors flows/flow-version.ts FlowVersionState.

const (
	VersionDraft  FlowVersionState = "DRAFT"
	VersionLocked FlowVersionState = "LOCKED"
)

type MoveActionRequest

type MoveActionRequest struct {
	Name          string `json:"name"`
	NewParentStep string `json:"newParentStep"`
}

MoveActionRequest moves Name to sit immediately after NewParentStep.

type PropSpec

type PropSpec struct {
	Name        string `json:"name"`
	DisplayName string `json:"displayName,omitempty"`
	Type        string `json:"type"` // string|number|boolean|object|array
	Required    bool   `json:"required,omitempty"`
	Description string `json:"description,omitempty"`
}

PropSpec describes one input property of an action/trigger. It is the ONE prop shape shared by the connector framework (connector.go), the catalogue, and the Tool input-schema derivation (tool.go) — one definition, every consumer.

type RunContext

type RunContext struct {
	Org         string
	Input       map[string]any
	PrevOutputs map[string]any
	Token       func(secretName string) ([]byte, error)
}

RunContext is what an action's Run receives. It is the connector's ENTIRE view of the world: the org (for logging/attribution only — never a place to widen scope), the resolved input, the prior steps' outputs (threaded), and Token — the ONLY door to a credential. Token is bound at dispatch time to the VALIDATED org (StepInput.Owner) and the connector's own provider id, so a connector can reach no other tenant's and no other provider's secret.

type RunEndInput

type RunEndInput struct {
	RunID  string `json:"runId"`
	Owner  string `json:"owner"`
	Status string `json:"status"`
}

RunStartInput / RunEndInput are the durable run-bookkeeping activity payloads. The durable path is the SINGLE owner of run bookkeeping (MED-1): whichever entrypoint drives a workflow — manual /run, MCP, or a scheduled cron tick — records the run exactly once here, so metering/audit never double-count and every execution lands a FlowRun row visible to listRuns/getRun.

type RunStartInput

type RunStartInput struct {
	RunID         string `json:"runId"`
	Owner         string `json:"owner"`
	FlowID        string `json:"flowId"`
	FlowVersionID string `json:"flowVersionId"`
}

RunStartInput / RunEndInput are the durable run-bookkeeping activity payloads. The durable path is the SINGLE owner of run bookkeeping (MED-1): whichever entrypoint drives a workflow — manual /run, MCP, or a scheduled cron tick — records the run exactly once here, so metering/audit never double-count and every execution lands a FlowRun row visible to listRuns/getRun.

type StepInput

type StepInput struct {
	Owner       string         `json:"owner"`
	RunID       string         `json:"runId"`
	Name        string         `json:"name"`
	PieceName   string         `json:"pieceName"`
	ActionName  string         `json:"actionName"`
	Input       map[string]any `json:"input"`
	PrevOutputs map[string]any `json:"prevOutputs"`
}

StepInput is the activity's per-step input. Owner is copied verbatim from FlowRunInput.Owner by the workflow — an activity can never widen its own scope.

type StepOutput

type StepOutput struct {
	Name   string `json:"name"`
	Output any    `json:"output"`
}

StepOutput is the activity's result for one step.

func ExecuteStepActivity

func ExecuteStepActivity(ctx context.Context, in StepInput) (StepOutput, error)

ExecuteStepActivity is the side-effecting body of ONE step. It is THE isolation boundary: RunContext.Token is bound to in.Owner (the VALIDATED org set at flow-start) and the step's own connector id, so a flow authored by org A can never reach org B's connection nor another provider's secret. in.Owner is used for token custody and NOTHING from in.Input can change it.

type StepSettings

type StepSettings struct {
	PieceName    string         `json:"pieceName,omitempty"`
	PieceVersion string         `json:"pieceVersion,omitempty"`
	ActionName   string         `json:"actionName,omitempty"`
	TriggerName  string         `json:"triggerName,omitempty"`
	Input        map[string]any `json:"input,omitempty"`
}

StepSettings is the flattened union of PieceTriggerSettings / PieceActionSettings / CodeActionSettings — the fields the engine actually reads to dispatch a step: which piece, which action/trigger, and the input map. (The builder-only settings — propertySettings, sampleData, errorHandlingOptions — round-trip opaquely via the raw version JSON in the store; they are not modeled here.)

type Store

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

Store is the automations database. ONE SQLite file — the system namespace's "automations" — holds every org's flows, versions, and runs; tenant isolation is the `org` column, physical on EVERY uniqueness + lookup index (each leads with org). MaxOpenConns(1) serializes writes against the single-writer WAL file.

func (*Store) ClaimMeter

func (s *Store) ClaimMeter(ctx context.Context, org, id string) (bool, error)

ClaimMeter atomically flips the run's metered flag 0→1 for (org,id) and reports whether THIS call won the flip. It is the exactly-once billing gate: only the winner meters + audits the run, so no entrypoint double-bills a run (MED-1). The (org,id) predicate keeps the claim tenant-scoped.

func (*Store) Close

func (s *Store) Close() error

Close closes the underlying database. Idempotent-safe via sql.DB.

func (*Store) CountRunsSince

func (s *Store) CountRunsSince(ctx context.Context, org string, since int64) (int, error)

CountRunsSince counts an org's run rows created at/after `since` (unix millis). It is the DURABLE per-org run-rate signal (it reads persisted rows, so the bound survives a restart), scoped to org and served by ix_auto_runs_org_created.

func (*Store) CreateFlow

func (s *Store) CreateFlow(ctx context.Context, f Flow) (Flow, error)

func (*Store) CreateRun

func (s *Store) CreateRun(ctx context.Context, r FlowRun) (FlowRun, error)

func (*Store) CreateRunIfAbsent

func (s *Store) CreateRunIfAbsent(ctx context.Context, r FlowRun) (bool, error)

CreateRunIfAbsent inserts a run row keyed on its id, doing NOTHING if it already exists. It reports whether THIS call created the row. Idempotent by run id (=the workflow execution id), so a retried run-start bookkeeping step, or a manual handler + the durable path racing to record the same run, converge on one row.

func (*Store) CreateVersion

func (s *Store) CreateVersion(ctx context.Context, v FlowVersion) (FlowVersion, error)

CreateVersion inserts a version. The flow must exist in the SAME org (validated here, not by a SQL FK, so a cross-tenant flow_id can never anchor a version).

func (*Store) DeleteFlow

func (s *Store) DeleteFlow(ctx context.Context, org, id string) (bool, error)

DeleteFlow removes a flow and all its versions + runs within the org. One transaction so a partial delete never strands a version/run.

func (*Store) DeleteFlowTriggers

func (s *Store) DeleteFlowTriggers(ctx context.Context, org, flowID string) error

DeleteFlowTriggers removes a flow's webhook subscription (on disable/delete). It is idempotent — deleting an absent subscription is a no-op — so a disable can always fail-closed by calling it regardless of the flow's current trigger type.

func (*Store) DeleteRun

func (s *Store) DeleteRun(ctx context.Context, org, id string) error

DeleteRun removes a single run row for (org,id). Used to roll back a run whose durable START failed transiently — so the run id (the idempotency key) is NOT burned and a redelivery can retry, rather than being permanently dropped.

func (*Store) GetFlow

func (s *Store) GetFlow(ctx context.Context, org, id string) (Flow, error)

func (*Store) GetRun

func (s *Store) GetRun(ctx context.Context, org, id string) (FlowRun, error)

func (*Store) GetVersion

func (s *Store) GetVersion(ctx context.Context, org, id string) (FlowVersion, error)

func (*Store) LatestVersion

func (s *Store) LatestVersion(ctx context.Context, org, flowID string) (FlowVersion, error)

LatestVersion returns the most-recently-created version for a flow, or errNotFound if the flow has none.

func (*Store) ListFlows

func (s *Store) ListFlows(ctx context.Context, org string, limit int) ([]Flow, error)

func (*Store) ListRuns

func (s *Store) ListRuns(ctx context.Context, org, flowID string, limit int) ([]FlowRun, error)

func (*Store) ListVersions

func (s *Store) ListVersions(ctx context.Context, org, flowID string, limit int) ([]FlowVersion, error)

func (*Store) MatchTriggers

func (s *Store) MatchTriggers(ctx context.Context, org, provider, event string) ([]triggerSub, error)

MatchTriggers returns the subscriptions whose (provider,event) matches an inbound event, scoped to org — the LEADING index column, so tenant isolation is physical: a query for org A can never return org B's rows. Bounded by maxLimit.

func (*Store) UpdateFlow

func (s *Store) UpdateFlow(ctx context.Context, f Flow) (Flow, error)

UpdateFlow persists the mutable flow fields (status, folder, published version, metadata) for (org,id). RowsAffected==0 ⇒ errNotFound (cross-tenant or missing).

func (*Store) UpdateRunStatus

func (s *Store) UpdateRunStatus(ctx context.Context, org, id string, status FlowRunStatus, finish, updated int64) error

UpdateRunStatus persists a terminal/observed status transition for (org,id).

func (*Store) UpdateVersion

func (s *Store) UpdateVersion(ctx context.Context, v FlowVersion) (FlowVersion, error)

UpdateVersion replaces a version's editable content (display name, trigger tree, valid, state). RowsAffected==0 ⇒ errNotFound.

func (*Store) UpsertTrigger

func (s *Store) UpsertTrigger(ctx context.Context, org, provider, event, flowID, versionID string) error

UpsertTrigger records (or refreshes) a flow's webhook subscription. Idempotent on (org, flow_id): re-enabling or re-publishing a flow overwrites provider/event/version so exactly one subscription per flow ever exists.

type Trigger

type Trigger struct {
	Name        string
	DisplayName string
	Description string
	Strategy    TriggerStrategy
	Props       []PropSpec
}

Trigger is one entry point of a connector. Strategy selects POLLING (cron) vs WEBHOOK vs MANUAL. Phase-1 triggers are catalogue/metadata only (the entry wiring lives in the flow's root trigger + enable/disable); a Trigger carries no Run.

type TriggerEvent

type TriggerEvent struct {
	Source    string         // provider slug == the flow trigger's pieceName ("github","stripe","slack",…)
	Name      string         // event name == the flow trigger's triggerName ("push","charge","message",…)
	DedupeKey string         // idempotency key; "" ⇒ every delivery is its own run
	Depth     int            // causation depth (in-platform hops so far); 0 = external origin
	Payload   map[string]any // the event body, readable by actions as {{trigger.field}}
}

TriggerEvent is the ONE inbound shape every trigger source hands to Deliver. Source/Name select which flows fire (they equal a flow trigger's pieceName / triggerName); Payload is threaded into each run as {{trigger.*}}; DedupeKey makes a re-delivered event idempotent (a flow fires at most once per DedupeKey); Depth is the causation depth (0 = external origin), propagated by in-platform producers so a cycle terminates.

type TriggerStrategy

type TriggerStrategy string

TriggerStrategy mirrors trigger/index.ts TriggerStrategy. It selects HOW a flow starts: POLLING → a cron schedule on the tasks engine; WEBHOOK/APP_WEBHOOK → an inbound HTTP event; MANUAL → an explicit /run.

const (
	StrategyPolling    TriggerStrategy = "POLLING"
	StrategyWebhook    TriggerStrategy = "WEBHOOK"
	StrategyAppWebhook TriggerStrategy = "APP_WEBHOOK"
	StrategyManual     TriggerStrategy = "MANUAL"
)

type UpdateActionRequest

type UpdateActionRequest struct {
	Name     string       `json:"name"`
	Type     string       `json:"type"`
	Settings StepSettings `json:"settings"`
}

UpdateActionRequest replaces the settings of the step named Name.

type UpdateTriggerRequest

type UpdateTriggerRequest FlowTrigger

UpdateTriggerRequest replaces the root trigger node.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL