diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx index 762f158a783..5536ae391c5 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx @@ -3,8 +3,6 @@ import { useMemo, useState } from 'react' import { Button, - ButtonGroup, - ButtonGroupItem, ChipCombobox, ChipInput, type ComboboxOptionGroup, @@ -21,28 +19,20 @@ import { import { ArrowLeft, ChevronDown, SquareArrowUpRight, X } from '@sim/emcn/icons' import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' -import { useMutation, useQueryClient } from '@tanstack/react-query' import { findValidationIssue, isValidationError } from '@/lib/api/client/errors' -import { requestJson } from '@/lib/api/client/request' import type { AddWorkflowGroupBodyInput, UpdateWorkflowGroupBodyInput, } from '@/lib/api/contracts/tables' -import { - putWorkflowNormalizedStateContract, - type WorkflowStateContractInput, -} from '@/lib/api/contracts/workflows' import type { ColumnDefinition, WorkflowGroup, WorkflowGroupDependencies, - WorkflowGroupDeploymentMode, WorkflowGroupInputMapping, WorkflowGroupOutput, } from '@/lib/table' import { getColumnId } from '@/lib/table/column-keys' import { columnTypeForLeaf, deriveOutputColumnName } from '@/lib/table/column-naming' -import { columnTypeById } from '@/lib/table/column-types' import { type FlattenOutputsBlockInput, type FlattenOutputsEdgeInput, @@ -58,12 +48,12 @@ import { } from '@/app/workspace/[workspaceId]/tables/[tableId]/components/sidebar-fields' import { PreviewWorkflow } from '@/app/workspace/[workspaceId]/w/components/preview' import { BlockTile } from '@/blocks/block-tile' +import { useDeployedWorkflowState } from '@/hooks/queries/deployments' import { useAddWorkflowGroup, useUpdateColumn, useUpdateWorkflowGroup, } from '@/hooks/queries/tables' -import { useWorkflowState, workflowKeys } from '@/hooks/queries/workflows' import type { WorkflowMetadata } from '@/stores/workflows/registry/types' import { InputMappingSection } from './input-mapping-section' import { RunSettingsSection } from './run-settings-section' @@ -142,25 +132,6 @@ interface BlockOutputGroup { paths: string[] } -interface WorkflowStatePayload { - blocks: Record< - string, - { - type: string - subBlocks?: Record - } & Record - > - edges: unknown[] - loops: unknown - parallels: unknown - lastSaved?: number - isDeployed?: boolean -} - -function tableColumnTypeToInputType(colType: ColumnDefinition['type'] | undefined): string { - return columnTypeById(colType).workflowInputType -} - /** * Right-edge sidebar for workflow group configuration. Three flows: * - create a new group (workflow + outputs + deps), @@ -277,11 +248,6 @@ export function WorkflowSidebarBody({ */ const otherColumns = anchorIdx >= allColumns.length ? allColumns : allColumns.slice(0, anchorIdx) - // Used by the "missing workflow input" suggestion below — for edit-output - // we exclude the column being edited (you can't suggest it as its own - // input). - const anchorColumnName = config.mode === 'edit-output' ? config.columnName : null - // Every left-of-current column is a valid dep — workflow output columns // included. Exclude this group's own outputs (you can't depend on yourself). const ownOutputIds = new Set(existingGroup?.outputs.map((o) => o.columnName) ?? []) @@ -312,11 +278,6 @@ export function WorkflowSidebarBody({ const [autoRun, setAutoRun] = useState(() => existingGroup ? existingGroup.autoRun !== false : false ) - // Which workflow state per-cell runs execute against. Defaults to `'live'` - // (the editable draft) for both new and pre-feature groups. - const [deploymentMode, setDeploymentMode] = useState( - () => existingGroup?.deploymentMode ?? 'live' - ) // Deps default to none selected. With auto-run on, at least one is required // (enforced via `depsValid` below); a legacy group with empty deps will // surface the error on first open until the user picks at least one column. @@ -332,101 +293,21 @@ export function WorkflowSidebarBody({ const [showValidation, setShowValidation] = useState(false) const [nameError, setNameError] = useState(null) - const workflowState = useWorkflowState(selectedWorkflowId || undefined) + const workflowState = useDeployedWorkflowState(selectedWorkflowId || null) - /** Resolves the unified Start block id and its current `inputFormat` field - * names. The "Add inputs" mutation only adds rows for table columns that - * aren't already represented in the start block. */ - const startBlockInputs = useMemo<{ - blockId: string | null - existingNames: Set - existing: InputFormatField[] - }>(() => { + /** Resolves Start-block inputs from the active deployment used by table runs. */ + const startBlockInputs = useMemo(() => { const blocks = (workflowState.data as { blocks?: Record } | null) ?.blocks - if (!blocks) return { blockId: null, existingNames: new Set(), existing: [] } + if (!blocks) return [] const candidate = TriggerUtils.findStartBlock(blocks, 'manual') - if (!candidate) return { blockId: null, existingNames: new Set(), existing: [] } + if (!candidate) return [] const block = blocks[candidate.blockId] as | { subBlocks?: Record } | undefined - const existing = normalizeInputFormatValue(block?.subBlocks?.inputFormat?.value) - return { - blockId: candidate.blockId, - existingNames: new Set(existing.map((f) => f.name).filter((n): n is string => !!n)), - existing, - } + return normalizeInputFormatValue(block?.subBlocks?.inputFormat?.value) }, [workflowState.data]) - const missingInputColumnNames = useMemo(() => { - if (!startBlockInputs.blockId) return [] - const anchor = anchorColumnName - return allColumns - .filter( - (c) => - getColumnId(c) !== anchor && - !c.workflowGroupId && - !startBlockInputs.existingNames.has(c.name) - ) - .map((c) => c.name) - }, [allColumns, anchorColumnName, startBlockInputs]) - - const queryClient = useQueryClient() - const addInputsMutation = useMutation({ - mutationFn: async () => { - const wfId = selectedWorkflowId - const startBlockId = startBlockInputs.blockId - const state = workflowState.data as WorkflowStatePayload | null | undefined - if (!wfId || !startBlockId || !state || missingInputColumnNames.length === 0) { - throw new Error('Nothing to add') - } - const startBlock = state.blocks[startBlockId] - if (!startBlock) throw new Error('Start block missing from workflow') - - const newFields: InputFormatField[] = missingInputColumnNames.map((name) => { - const col = allColumns.find((c) => c.name === name) - return { - id: generateId(), - name, - type: tableColumnTypeToInputType(col?.type), - value: '', - collapsed: false, - } as InputFormatField & { id: string; collapsed: boolean } - }) - - const updatedSubBlock = { - ...(startBlock.subBlocks?.inputFormat ?? { id: 'inputFormat', type: 'input-format' }), - value: [...startBlockInputs.existing, ...newFields], - } - const updatedBlocks = { - ...state.blocks, - [startBlockId]: { - ...startBlock, - subBlocks: { ...startBlock.subBlocks, inputFormat: updatedSubBlock }, - }, - } - - const rawBody = { - blocks: updatedBlocks, - edges: state.edges, - loops: state.loops, - parallels: state.parallels, - lastSaved: state.lastSaved ?? Date.now(), - isDeployed: state.isDeployed ?? false, - } - // double-cast-allowed: WorkflowStatePayload is the loose local view of - // useWorkflowState; round-trip back to the strict PUT body shape. - const body = rawBody as unknown as WorkflowStateContractInput - await requestJson(putWorkflowNormalizedStateContract, { params: { id: wfId }, body }) - }, - onError: (err) => { - toast.error(toError(err).message) - }, - onSettled: () => { - return queryClient.invalidateQueries({ queryKey: workflowKeys.state(selectedWorkflowId) }) - }, - }) - const blockOutputGroups = useMemo(() => { const state = workflowState.data as | { @@ -517,13 +398,13 @@ export function WorkflowSidebarBody({ // Once the Start block's input fields resolve, auto-fill any field that has no // persisted mapping yet but matches a table column by name. Runs once; never // overrides a persisted or user-picked mapping. - if (!inputMappingsHydrated && startBlockInputs.existing.length > 0) { + if (!inputMappingsHydrated && startBlockInputs.length > 0) { // Map a Start input field to the column sharing its name, storing the // column id (the value the dropdowns and persisted mappings key on). const idByColumnName = new Map(depOptions.map((c) => [c.name, getColumnId(c)])) const next = { ...inputMappings } let changed = false - for (const field of startBlockInputs.existing) { + for (const field of startBlockInputs) { if (!field.name || next[field.name]) continue const colId = idByColumnName.get(field.name) if (colId) { @@ -676,7 +557,6 @@ export function WorkflowSidebarBody({ outputs: fullOutputs, ...(newOutputColumns.length > 0 ? { newOutputColumns } : {}), inputMappings: inputMappingsList, - deploymentMode, autoRun, }) toast.success(`Saved "${existingGroup.name ?? 'Workflow'}"`) @@ -708,7 +588,6 @@ export function WorkflowSidebarBody({ dependencies, outputs: groupOutputs, inputMappings: inputMappingsList, - deploymentMode, autoRun, } await addWorkflowGroup.mutateAsync({ group, outputColumns: newOutputColumns }) @@ -815,29 +694,6 @@ export function WorkflowSidebarBody({
- {!isEnrichment && - startBlockInputs.blockId && - missingInputColumnNames.length > 0 && ( - - - - - - Adds {missingInputColumnNames.join(', ')} to the workflow's Start block - - - )}
{workflowState.isLoading ? ( @@ -896,12 +752,16 @@ export function WorkflowSidebarBody({
Workflow ({ label: wf.name, value: wf.id })) ?? []} + options={ + workflows + ?.filter((workflow) => workflow.isDeployed) + .map((workflow) => ({ label: workflow.name, value: workflow.id })) ?? [] + } value={selectedWorkflowId} onChange={(v) => setSelectedWorkflowId(v)} placeholder='Select a workflow' disabled={!workflows || workflows.length === 0 || isEditOutputMode || isEnrichment} - emptyMessage='No manual triggers configured' + emptyMessage='No deployed workflows available' maxHeight={260} searchable searchPlaceholder='Search workflows...' @@ -993,25 +853,8 @@ export function WorkflowSidebarBody({
{showAdvanced && ( <> - {!isEnrichment && ( - <> -
- - - setDeploymentMode(v === 'deployed' ? 'deployed' : 'live') - } - > - Live - Deployed - -
- - - )} ({ appendTableEventMock: vi.fn() })) +const { appendTableEventMock, flattenWorkflowOutputsMock } = vi.hoisted(() => ({ + appendTableEventMock: vi.fn(), + flattenWorkflowOutputsMock: vi.fn(), +})) vi.mock('@/lib/table/events', () => ({ appendTableEvent: appendTableEventMock })) +vi.mock('@/lib/workflows/blocks/flatten-outputs', () => ({ + flattenWorkflowOutputs: flattenWorkflowOutputsMock, +})) beforeEach(() => { vi.clearAllMocks() @@ -50,6 +57,106 @@ const QUEUED_PAYLOAD = { }, } +function latestDeployment( + startType = 'start_trigger' +): Parameters[1] { + return { + blocks: { + start: { + id: 'start', + type: startType, + subBlocks: { + inputFormat: { value: [{ name: 'company', type: 'string' }] }, + }, + }, + agent: { + id: 'agent', + type: 'agent', + subBlocks: {}, + }, + }, + edges: [{ id: 'start-agent', source: 'start', target: 'agent' }], + loops: {}, + parallels: {}, + variables: {}, + isFromNormalizedTables: false, + deploymentVersionId: 'deployment-version-latest', + } as Parameters[1] +} + +describe('latest table workflow deployment mappings', () => { + beforeEach(() => { + flattenWorkflowOutputsMock.mockReturnValue([ + { + blockId: 'agent', + blockName: 'Agent', + blockType: 'agent', + path: 'content', + leafType: 'string', + }, + ]) + }) + + it('accepts saved mappings that the latest active deployment still supports', () => { + expect(() => + assertWorkflowGroupMatchesLatestDeployment( + { + id: 'group-1', + workflowId: 'workflow-1', + outputs: [{ blockId: 'agent', path: 'content', columnName: 'column-output' }], + inputMappings: [{ inputName: 'company', columnName: 'column-company' }], + }, + latestDeployment() + ) + ).not.toThrow() + }) + + it('accepts a canonical split manual start block', () => { + expect(() => + assertWorkflowGroupMatchesLatestDeployment( + { + id: 'group-1', + workflowId: 'workflow-1', + outputs: [{ blockId: 'agent', path: 'content', columnName: 'column-output' }], + inputMappings: [{ inputName: 'company', columnName: 'column-company' }], + }, + latestDeployment('manual_trigger') + ) + ).not.toThrow() + }) + + it('rejects an output mapping removed by the latest active deployment', () => { + expect(() => + assertWorkflowGroupMatchesLatestDeployment( + { + id: 'group-1', + workflowId: 'workflow-1', + outputs: [{ blockId: 'agent', path: 'score', columnName: 'column-output' }], + }, + latestDeployment() + ) + ).toThrow( + 'Workflow group group-1 output agent::score is not available in the latest active deployment' + ) + }) + + it('rejects an input mapping removed by the latest active deployment', () => { + expect(() => + assertWorkflowGroupMatchesLatestDeployment( + { + id: 'group-1', + workflowId: 'workflow-1', + outputs: [{ blockId: 'agent', path: 'content', columnName: 'column-output' }], + inputMappings: [{ inputName: 'website', columnName: 'column-website' }], + }, + latestDeployment() + ) + ).toThrow( + 'Workflow group group-1 input website is not available in the latest active deployment' + ) + }) +}) + describe('table workflow carrier deadline', () => { it('preserves one absolute deadline when a later cascade group creates its controller', () => { vi.useFakeTimers() diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 28264abd776..33c21174d70 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -54,12 +54,58 @@ import { type QueuedWorkflowGroupCellPayload, type WorkflowGroupCellPayload, } from '@/lib/table/workflow-columns' +import { flattenWorkflowOutputs } from '@/lib/workflows/blocks/flatten-outputs' +import { normalizeInputFormatValue } from '@/lib/workflows/input-format' +import type { DeployedWorkflowData } from '@/lib/workflows/persistence/utils' +import { TriggerUtils } from '@/lib/workflows/triggers/triggers' import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' export type { WorkflowGroupCellPayload } const logger = createLogger('TriggerWorkflowGroupCell') +/** + * Fails before workflow execution when saved table mappings are incompatible + * with the latest active deployment loaded for this cell run. + */ +export function assertWorkflowGroupMatchesLatestDeployment( + group: WorkflowGroup, + deployment: DeployedWorkflowData +): void { + const validOutputs = new Set( + flattenWorkflowOutputs(Object.values(deployment.blocks), deployment.edges).map( + (output) => `${output.blockId}::${output.path}` + ) + ) + const invalidOutput = group.outputs.find( + (output) => !validOutputs.has(`${output.blockId}::${output.path}`) + ) + if (invalidOutput) { + throw new Error( + `Workflow group ${group.id} output ${invalidOutput.blockId}::${invalidOutput.path} is not available in the latest active deployment` + ) + } + + const startCandidate = TriggerUtils.findStartBlock(deployment.blocks, 'manual') + if (!startCandidate) { + throw new Error('Workflow is missing a Start trigger') + } + + const validInputNames = new Set( + normalizeInputFormatValue(startCandidate.block.subBlocks?.inputFormat?.value).map( + (input) => input.name + ) + ) + const invalidInput = (group.inputMappings ?? []).find( + (mapping) => !validInputNames.has(mapping.inputName) + ) + if (invalidInput) { + throw new Error( + `Workflow group ${group.id} input ${invalidInput.inputName} is not available in the latest active deployment` + ) + } +} + function requirePayloadBillingAttribution( payload: WorkflowGroupCellPayload ): BillingAttributionSnapshot { @@ -389,19 +435,13 @@ async function runWorkflowAndWriteTerminal( const billingAttribution = requirePayloadBillingAttribution(payload) const timeoutController = createWorkflowGroupAttemptTimeoutController(payload, signal) const attemptSignal = timeoutController.signal - // Read from the live `group`, not the payload: in a cascade the payload is the - // first group's snapshot, so a downstream group with a different version must - // use its own setting (same reason `workflowId` is re-derived per iteration). - const deploymentMode = group.deploymentMode const requestId = `wfgrp-${executionId}` try { return await runWithRequestContext({ requestId }, async () => { const { getRowById } = await import('@/lib/table/rows/service') const { executeWorkflow } = await import('@/lib/workflows/executor/execute-workflow') - const { loadWorkflowFromNormalizedTables, loadDeployedWorkflowState } = await import( - '@/lib/workflows/persistence/utils' - ) + const { loadDeployedWorkflowState } = await import('@/lib/workflows/persistence/utils') const { buildCancelledExecution, createWorkflowCellProgressWriter, @@ -693,32 +733,22 @@ async function runWorkflowAndWriteTerminal( return 'error' } - // `deployed` groups run the workflow's latest active deployment; `live` - // (default) runs the editable draft. A `deployed` group whose workflow - // has never been deployed fails the cell — no silent fallback to draft. - let normalizedData: Awaited> - if (deploymentMode === 'deployed') { - try { - normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId) - } catch (err) { - // Surface the real reason (missing deployment vs. transient DB/migration - // failure) rather than always claiming the workflow isn't deployed. - await writeState({ - status: 'error', - executionId, - jobId: null, - workflowId, - error: toError(err).message, - }) - return 'error' - } - } else { - normalizedData = await loadWorkflowFromNormalizedTables(workflowId) + let normalizedData: Awaited> + try { + normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId) + assertWorkflowGroupMatchesLatestDeployment(group, normalizedData) + } catch (err) { + await writeState({ + status: 'error', + executionId, + jobId: null, + workflowId, + error: toError(err).message, + }) + return 'error' } - const startBlock = normalizedData - ? Object.values(normalizedData.blocks).find((b) => b?.type === 'start_trigger') - : undefined - if (!startBlock) { + const startCandidate = TriggerUtils.findStartBlock(normalizedData.blocks, 'manual') + if (!startCandidate) { await writeState({ status: 'error', executionId, @@ -1006,11 +1036,9 @@ async function runWorkflowAndWriteTerminal( }, executionMode: 'sync', workflowTriggerType: 'table', - triggerBlockId: startBlock.id, - // `deployed` groups execute the latest active deployment; everything - // else runs the editable draft (the table default). Matches the - // state loaded above for start-block / output-block resolution. - useDraftState: deploymentMode !== 'deployed', + triggerBlockId: startCandidate.blockId, + useDraftState: false, + workflowStateOverride: normalizedData, abortSignal: attemptSignal, onBlockStart: progressWriter.onBlockStart, onBlockComplete: progressWriter.onBlockComplete, diff --git a/apps/sim/executor/execution/types.ts b/apps/sim/executor/execution/types.ts index 35cf038381f..ba7e7db6a8a 100644 --- a/apps/sim/executor/execution/types.ts +++ b/apps/sim/executor/execution/types.ts @@ -56,6 +56,7 @@ export interface ExecutionMetadata { edges: Edge[] loops?: Record parallels?: Record + variables?: Record deploymentVersionId?: string } largeValueExecutionIds?: string[] diff --git a/apps/sim/lib/copilot/generated/tool-catalog-v1.ts b/apps/sim/lib/copilot/generated/tool-catalog-v1.ts index b50898eb45a..ed93f30ed58 100644 --- a/apps/sim/lib/copilot/generated/tool-catalog-v1.ts +++ b/apps/sim/lib/copilot/generated/tool-catalog-v1.ts @@ -5312,12 +5312,6 @@ export const TableAutomations: ToolCatalogEntry = { }, }, }, - deploymentMode: { - type: 'string', - description: - 'Which workflow version rows execute: "live" (default, editable draft — edits take effect immediately) or "deployed" (latest active deployment; fails if the workflow was never deployed).', - enum: ['live', 'deployed'], - }, groupId: { type: 'string', description: @@ -6092,12 +6086,6 @@ export const UserTable: ToolCatalogEntry = { }, }, }, - deploymentMode: { - type: 'string', - description: - "Which version of the backing workflow this group's per-row runs execute, for add_workflow_group and update_workflow_group. 'live' (default) runs the editable draft, so later edits take effect immediately. 'deployed' runs the workflow's latest active deployment, pinning rows to a published version — if that workflow has never been deployed the cell fails rather than falling back to the draft. Only meaningful for workflow groups; enrichment groups have no backing workflow.", - enum: ['live', 'deployed'], - }, description: { type: 'string', description: "Table description (optional for 'create')" }, enrichmentId: { type: 'string', diff --git a/apps/sim/lib/copilot/generated/tool-schemas-v1.ts b/apps/sim/lib/copilot/generated/tool-schemas-v1.ts index 7625f38c296..1fecc8476a6 100644 --- a/apps/sim/lib/copilot/generated/tool-schemas-v1.ts +++ b/apps/sim/lib/copilot/generated/tool-schemas-v1.ts @@ -5173,12 +5173,6 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { }, }, }, - deploymentMode: { - type: 'string', - description: - 'Which workflow version rows execute: "live" (default, editable draft — edits take effect immediately) or "deployed" (latest active deployment; fails if the workflow was never deployed).', - enum: ['live', 'deployed'], - }, groupId: { type: 'string', description: @@ -6034,12 +6028,6 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { }, }, }, - deploymentMode: { - type: 'string', - description: - "Which version of the backing workflow this group's per-row runs execute, for add_workflow_group and update_workflow_group. 'live' (default) runs the editable draft, so later edits take effect immediately. 'deployed' runs the workflow's latest active deployment, pinning rows to a published version — if that workflow has never been deployed the cell fails rather than falling back to the draft. Only meaningful for workflow groups; enrichment groups have no backing workflow.", - enum: ['live', 'deployed'], - }, description: { type: 'string', description: "Table description (optional for 'create')", diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.test.ts b/apps/sim/lib/copilot/tools/server/table/user-table.test.ts index 85d39c44e4d..1e96164e1b5 100644 --- a/apps/sim/lib/copilot/tools/server/table/user-table.test.ts +++ b/apps/sim/lib/copilot/tools/server/table/user-table.test.ts @@ -258,6 +258,7 @@ vi.mock('@/lib/workflows/application/context', () => ({ })) vi.mock('@/lib/workflows/application/resolve-workflow-outputs', () => ({ + loadResolvedDeployedWorkflowOutputs: async () => mockExecuteCopilotWorkflowUseCase(), loadResolvedWorkflowOutputs: async () => mockExecuteCopilotWorkflowUseCase(), resolveWorkflowOutputs: { operation: { id: 'workflows.read' } }, })) @@ -959,6 +960,25 @@ describe('userTableServerTool workflow scope', () => { expect(mockAddWorkflowGroup).not.toHaveBeenCalled() }) + it('does not pass a legacy deployment mode into workflow group creation', async () => { + const result = await userTableServerTool.execute( + { + operation: 'add_workflow_group', + args: { + tableId: 'tbl_1', + workflowId: 'workflow-1', + outputs: [{ blockId: 'block-1', path: 'content' }], + deploymentMode: 'live', + }, + }, + buildToolContext() + ) + + expect(result.success).toBe(true) + expect(mockAddWorkflowGroup).toHaveBeenCalledTimes(1) + expect(mockAddWorkflowGroup.mock.calls[0][0].group).not.toHaveProperty('deploymentMode') + }) + it('conceals unknown application failures from tool output', async () => { mockQueryRows.mockRejectedValueOnce(new Error('database host unavailable')) diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.ts b/apps/sim/lib/copilot/tools/server/table/user-table.ts index 372d3aa9898..d48ac344d42 100644 --- a/apps/sim/lib/copilot/tools/server/table/user-table.ts +++ b/apps/sim/lib/copilot/tools/server/table/user-table.ts @@ -62,7 +62,6 @@ import type { TablePredicateInput, TableSchema, WorkflowGroupDependencies, - WorkflowGroupDeploymentMode, } from '@/lib/table/types' import { viewConfigIdsToNames } from '@/lib/table/views/service' import type { ResolvedSecretTraceProvenanceV1 } from '@/executor/utils/resolved-secret-trace-registry' @@ -132,16 +131,6 @@ function resolveAuthorizedWorkflowOutputs( }) } -/** - * Narrows a raw `deploymentMode` arg to the `'live' | 'deployed'` union, or - * `undefined` when absent/invalid (leaving the group's existing value — which - * itself defaults to `'live'`). Lets Mothership choose whether a group's - * per-cell runs execute the live draft or the latest active deployment. - */ -function parseDeploymentMode(value: unknown): WorkflowGroupDeploymentMode | undefined { - return value === 'live' || value === 'deployed' ? value : undefined -} - /** Validates an optional row limit against the policy for the requested surface operation. */ function limitError(limit: unknown, max?: number): string | null { if (limit === undefined) return null @@ -1242,7 +1231,6 @@ export const userTableServerTool: BaseServerTool const dependencies = args.dependencies as WorkflowGroupDependencies | undefined const name = args.name as string | undefined - const deploymentMode = parseDeploymentMode(args.deploymentMode) assertNotAborted() const autoRun = args.autoRun === true const { table: updated, group } = await executeCopilotCreateWorkflowTableGroup(context, { @@ -1252,7 +1240,6 @@ export const userTableServerTool: BaseServerTool outputs: rawOutputs, name, dependencies, - deploymentMode, autoRun, }) return { @@ -1294,7 +1281,6 @@ export const userTableServerTool: BaseServerTool dependencies: args.dependencies as WorkflowGroupDependencies | undefined, outputs: updateOutputs, mappingUpdates, - deploymentMode: parseDeploymentMode(args.deploymentMode), autoRun: typeof args.autoRun === 'boolean' ? args.autoRun : undefined, }) return { diff --git a/apps/sim/lib/table/application/groups.test.ts b/apps/sim/lib/table/application/groups.test.ts index 6eb36a21c5c..59f3135caa7 100644 --- a/apps/sim/lib/table/application/groups.test.ts +++ b/apps/sim/lib/table/application/groups.test.ts @@ -72,7 +72,7 @@ vi.mock('@/lib/workflows/application/context', () => ({ resolveActiveWorkflowApplicationContext: mocks.resolveWorkflowContext, })) vi.mock('@/lib/workflows/application/resolve-workflow-outputs', () => ({ - loadResolvedWorkflowOutputs: mocks.loadWorkflowOutputs, + loadResolvedDeployedWorkflowOutputs: mocks.loadWorkflowOutputs, })) import { v2WorkflowGroupSchema } from '@/lib/api/contracts/v2/tables' diff --git a/apps/sim/lib/table/application/groups.ts b/apps/sim/lib/table/application/groups.ts index f8eef13eaca..485c960b78e 100644 --- a/apps/sim/lib/table/application/groups.ts +++ b/apps/sim/lib/table/application/groups.ts @@ -35,7 +35,7 @@ import { } from '@/lib/table/workflow-groups/service' import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context' import type { ResolveWorkflowOutputsResult } from '@/lib/workflows/application/resolve-workflow-outputs' -import { loadResolvedWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' +import { loadResolvedDeployedWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' import { getEnrichment } from '@/enrichments/registry' import type { EnrichmentConfig } from '@/enrichments/types' @@ -64,7 +64,7 @@ async function resolveWorkflowForAuthorizedTableCommand( workflowId, assertedWorkspaceId: workspaceId, }) - return loadResolvedWorkflowOutputs(workflowContext) + return loadResolvedDeployedWorkflowOutputs(workflowContext) } async function resolveRelatedWorkflowForTableRoute( @@ -1079,6 +1079,11 @@ export const addWorkflowTableGroupOutput = defineAuthorizedTableUseCase({ context.workspaceId ) const outputs = requireWorkflowOutputs(resolvedWorkflow, group.workflowId) + validateRequestedOutputs( + [...group.outputs, { blockId: input.blockId, path: input.path }], + resolvedWorkflow, + group.workflowId + ) const output = outputs.find( (candidate) => candidate.blockId === input.blockId && candidate.path === input.path ) diff --git a/apps/sim/lib/table/workflow-groups/service.ts b/apps/sim/lib/table/workflow-groups/service.ts index 7f526593b5a..9437427fef3 100644 --- a/apps/sim/lib/table/workflow-groups/service.ts +++ b/apps/sim/lib/table/workflow-groups/service.ts @@ -741,6 +741,15 @@ export async function addWorkflowGroupOutput( const [db, ib] = orderKey(b) return da !== db ? da - db : ia - ib }) + const invalidOutput = allGroupOutputs.find( + (output) => !resolvedOrder.has(`${output.blockId}::${output.path}`) + ) + if (invalidOutput) { + throw new OrchestrationError( + 'conflict', + `Workflow group "${data.groupId}" mappings changed concurrently; retry the add.` + ) + } const orderedGroupColIds = allGroupOutputs.map((o) => o.columnName) const updatedGroup: WorkflowGroup = { ...group, diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts index 6ccf06f983a..f750feea8b8 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts @@ -7,7 +7,9 @@ import { OrchestrationError } from '@/lib/core/orchestration/types' const mocks = vi.hoisted(() => ({ flatten: vi.fn(), + loadDeployed: vi.fn(), load: vi.fn(), + NoActiveDeploymentError: class NoActiveDeploymentError extends Error {}, order: vi.fn(), resolveContext: vi.fn(), resolvePermission: vi.fn(), @@ -33,10 +35,15 @@ vi.mock('@/lib/workflows/blocks/flatten-outputs', () => ({ })) vi.mock('@/lib/workflows/persistence/utils', () => ({ + NoActiveDeploymentError: mocks.NoActiveDeploymentError, + loadDeployedWorkflowState: mocks.loadDeployed, loadWorkflowFromNormalizedTables: mocks.load, })) -import { resolveWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' +import { + loadResolvedDeployedWorkflowOutputs, + resolveWorkflowOutputs, +} from '@/lib/workflows/application/resolve-workflow-outputs' const principal = { kind: 'delegated' as const, @@ -59,12 +66,16 @@ describe('resolveWorkflowOutputs', () => { workspaceOrganizationId: null, allowPersonalApiKeys: true, billedAccountUserId: 'billing-owner-1', - workflow: { id: 'workflow-1' }, + workflow: { id: 'workflow-1', isDeployed: true }, }) mocks.load.mockResolvedValue({ blocks: { block1: { id: 'block-1', type: 'agent', name: 'Agent', subBlocks: {} } }, edges: [], }) + mocks.loadDeployed.mockResolvedValue({ + blocks: { block1: { id: 'block-1', type: 'agent', name: 'Agent', subBlocks: {} } }, + edges: [], + }) mocks.flatten.mockReturnValue([ { blockId: 'block-1', @@ -110,6 +121,42 @@ describe('resolveWorkflowOutputs', () => { expect(mocks.load).not.toHaveBeenCalled() }) + it('resolves table mappings from the active deployment state', async () => { + const context = await mocks.resolveContext() + + await expect(loadResolvedDeployedWorkflowOutputs(context)).resolves.toMatchObject({ + workflowId: 'workflow-1', + outputs: [{ blockId: 'block-1', path: 'content' }], + }) + + expect(mocks.loadDeployed).toHaveBeenCalledWith('workflow-1', 'workspace-1') + expect(mocks.load).not.toHaveBeenCalled() + }) + + it('rejects a workflow without an active deployment before resolving mappings', async () => { + const context = { + ...(await mocks.resolveContext()), + workflow: { id: 'workflow-1', isDeployed: false }, + } + + await expect(loadResolvedDeployedWorkflowOutputs(context)).rejects.toMatchObject({ + code: 'validation', + message: 'Workflow must have an active deployment', + }) + expect(mocks.loadDeployed).not.toHaveBeenCalled() + }) + + it('rejects inconsistent deployment metadata without returning draft mappings', async () => { + const context = await mocks.resolveContext() + mocks.loadDeployed.mockRejectedValueOnce(new mocks.NoActiveDeploymentError()) + + await expect(loadResolvedDeployedWorkflowOutputs(context)).rejects.toMatchObject({ + code: 'validation', + message: 'Workflow must have an active deployment', + }) + expect(mocks.load).not.toHaveBeenCalled() + }) + it('rejects expired delegated scope before loading workflow state', async () => { await expect( resolveWorkflowOutputs.execute({ diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts index 209d26107f6..cb60823c006 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts @@ -1,3 +1,4 @@ +import { OrchestrationError } from '@/lib/core/orchestration/types' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { type ActiveWorkflowApplicationContext, @@ -9,7 +10,11 @@ import { flattenWorkflowOutputs, getBlockExecutionOrder, } from '@/lib/workflows/blocks/flatten-outputs' -import { loadWorkflowFromNormalizedTables } from '@/lib/workflows/persistence/utils' +import { + loadDeployedWorkflowState, + loadWorkflowFromNormalizedTables, + NoActiveDeploymentError, +} from '@/lib/workflows/persistence/utils' export interface ResolveWorkflowOutputsInput { workflowId: string @@ -22,14 +27,14 @@ export interface ResolveWorkflowOutputsResult { executionOrderByBlockId: Record } -/** Loads output metadata after a top-level application command has authorized this workflow context. */ -export async function loadResolvedWorkflowOutputs( - context: ActiveWorkflowApplicationContext -): Promise { - const normalized = await loadWorkflowFromNormalizedTables(context.workflowId) - if (!normalized) { - return { workflowId: context.workflowId, outputs: null, executionOrderByBlockId: {} } - } +type ResolvableWorkflowState = + | NonNullable>> + | Awaited> + +function resolveWorkflowOutputsFromState( + workflowId: string, + normalized: ResolvableWorkflowState +): ResolveWorkflowOutputsResult { const blocks = Object.values(normalized.blocks ?? {}).map((block) => ({ id: block.id, type: block.type, @@ -38,12 +43,41 @@ export async function loadResolvedWorkflowOutputs( subBlocks: block.subBlocks as Record | undefined, })) return { - workflowId: context.workflowId, + workflowId, outputs: flattenWorkflowOutputs(blocks, normalized.edges ?? []), executionOrderByBlockId: getBlockExecutionOrder(blocks, normalized.edges ?? []), } } +/** Loads output metadata after a top-level application command has authorized this workflow context. */ +export async function loadResolvedWorkflowOutputs( + context: ActiveWorkflowApplicationContext +): Promise { + const normalized = await loadWorkflowFromNormalizedTables(context.workflowId) + if (!normalized) { + return { workflowId: context.workflowId, outputs: null, executionOrderByBlockId: {} } + } + return resolveWorkflowOutputsFromState(context.workflowId, normalized) +} + +/** Loads output metadata from the active deployment after workflow authorization. */ +export async function loadResolvedDeployedWorkflowOutputs( + context: ActiveWorkflowApplicationContext +): Promise { + if (!context.workflow.isDeployed) { + throw new OrchestrationError('validation', 'Workflow must have an active deployment') + } + try { + const normalized = await loadDeployedWorkflowState(context.workflowId, context.workspaceId) + return resolveWorkflowOutputsFromState(context.workflowId, normalized) + } catch (error) { + if (error instanceof NoActiveDeploymentError) { + throw new OrchestrationError('validation', 'Workflow must have an active deployment') + } + throw error + } +} + export const resolveWorkflowOutputs = defineAuthorizedWorkflowUseCase({ operation: workflowOperations.read, resolveContext: ({ input }: { input: ResolveWorkflowOutputsInput }) => diff --git a/apps/sim/lib/workflows/executor/execute-workflow.test.ts b/apps/sim/lib/workflows/executor/execute-workflow.test.ts index 4240058f7d3..664ff6f511a 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.test.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.test.ts @@ -234,6 +234,32 @@ describe('executeWorkflow', () => { ) }) + it('forwards a trusted immutable workflow state to the execution snapshot', async () => { + const workflowStateOverride = { + blocks: { 'block-1': { id: 'block-1', type: 'start_trigger' } }, + edges: [], + loops: {}, + parallels: {}, + variables: { + 'variable-1': { id: 'variable-1', name: 'deployed', value: 'frozen' }, + }, + deploymentVersionId: 'deployment-version-1', + } + + await executeWorkflow(workflow, 'request-1', { prompt: 'hello' }, 'actor-1', { + enabled: true, + principal, + billingAttribution, + workflowStateOverride, + }) + + const coreParams = executeWorkflowCoreMock.mock.calls[0]?.[0] as { + snapshot: ExecutionSnapshot + } + expect(coreParams.snapshot.metadata.workflowStateOverride).toEqual(workflowStateOverride) + expect(coreParams.snapshot.workflowVariables).toEqual(workflowStateOverride.variables) + }) + it('waits for post-execution persistence before resolving', async () => { let resolvePostExecution!: () => void waitForPostExecutionMock.mockReturnValueOnce( diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts index 8336ea332d2..b2589685fb3 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.ts @@ -48,6 +48,8 @@ export interface ExecuteWorkflowOptions { abortSignal?: AbortSignal /** Use the live/draft workflow state instead of the deployed state. Used by copilot. */ useDraftState?: boolean + /** Immutable workflow state selected by a trusted server-side trigger boundary. */ + workflowStateOverride?: NonNullable /** Stop execution after this block completes. Used for "run until block" feature. */ stopAfterBlockId?: string /** Run-from-block configuration using a prior execution snapshot. */ @@ -142,6 +144,7 @@ export async function executeWorkflow( triggerType, triggerBlockId: streamConfig?.triggerBlockId, useDraftState: streamConfig?.useDraftState ?? false, + workflowStateOverride: streamConfig?.workflowStateOverride, startTime: new Date().toISOString(), isClientSession: false, enforceCredentialAccess: streamConfig?.enforceCredentialAccess ?? false, @@ -163,7 +166,7 @@ export async function executeWorkflow( metadata, workflow, input, - workflow.variables || {}, + streamConfig?.workflowStateOverride?.variables ?? workflow.variables ?? {}, streamConfig?.selectedOutputs || [] ) diff --git a/apps/sim/lib/workflows/orchestration/deploy.test.ts b/apps/sim/lib/workflows/orchestration/deploy.test.ts index 262a45c30d7..08e4f63fce8 100644 --- a/apps/sim/lib/workflows/orchestration/deploy.test.ts +++ b/apps/sim/lib/workflows/orchestration/deploy.test.ts @@ -896,6 +896,7 @@ describe('mutation lock on the orchestration entry points', () => { expect(result.success).toBe(false) expect(result.error).toContain('locked') + expect(result.errorCode).toBe('locked') expect(mockRecordAudit).not.toHaveBeenCalled() }) diff --git a/apps/sim/lib/workflows/orchestration/deploy.ts b/apps/sim/lib/workflows/orchestration/deploy.ts index 5584a52412f..9290143972e 100644 --- a/apps/sim/lib/workflows/orchestration/deploy.ts +++ b/apps/sim/lib/workflows/orchestration/deploy.ts @@ -151,7 +151,7 @@ export async function performFullDeploy( // Backstop for every caller — routes may assert first to render their own 423, // but the copilot deploy tools call this directly. const lockDenial = await workflowLockDenial(workflowId) - if (lockDenial) return { success: false, error: lockDenial, errorCode: 'validation' } + if (lockDenial) return { success: false, error: lockDenial, errorCode: 'locked' } const [workflowRecord] = await db .select() @@ -509,6 +509,7 @@ export interface PerformFullUndeployParams { export interface PerformFullUndeployResult { success: boolean error?: string + errorCode?: OrchestrationErrorCode warnings?: string[] } @@ -526,7 +527,7 @@ export async function performFullUndeploy( const requestId = params.requestId ?? generateRequestId() const lockDenial = await workflowLockDenial(workflowId) - if (lockDenial) return { success: false, error: lockDenial } + if (lockDenial) return { success: false, error: lockDenial, errorCode: 'locked' } const [workflowRecord] = await db .select() @@ -661,7 +662,7 @@ export async function performActivateVersion( const idempotencyKey = params.idempotencyKey ?? generateId() const lockDenial = await workflowLockDenial(workflowId) - if (lockDenial) return { success: false, error: lockDenial, errorCode: 'validation' } + if (lockDenial) return { success: false, error: lockDenial, errorCode: 'locked' } const [versionRow] = await db .select({ diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts new file mode 100644 index 00000000000..1eae182e775 --- /dev/null +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -0,0 +1,362 @@ +/** + * @vitest-environment node + */ +import { dbChainMockFns, resetDbChainMock } from '@sim/testing' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockLoadRuntimeSecrets, mockPerformFullDeploy } = vi.hoisted(() => ({ + mockLoadRuntimeSecrets: vi.fn(), + mockPerformFullDeploy: vi.fn(), +})) + +vi.mock('@sim/runtime-secrets', () => ({ + loadRuntimeSecrets: mockLoadRuntimeSecrets, +})) + +vi.mock('@/lib/workflows/orchestration/deploy', () => ({ + performFullDeploy: mockPerformFullDeploy, +})) + +import { + backfillTableWorkflowDeployments, + deployTableWorkflow, + parseTableWorkflowDeploymentBackfillArgs, + postgresTableWorkflowDeploymentStore, + prepareTableWorkflowDeploymentBackfillEnvironment, + TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE, + type TableWorkflowDeploymentCandidate, + type TableWorkflowDeploymentStore, +} from '@/scripts/backfill-table-workflow-deployments' + +const ORIGINAL_ENV = { + DATABASE_URL: process.env.DATABASE_URL, + DATABASE_URL_WEB: process.env.DATABASE_URL_WEB, + REDIS_TLS_SERVERNAME: process.env.REDIS_TLS_SERVERNAME, + REDIS_URL: process.env.REDIS_URL, + SIM_ENV_SECRET_ID: process.env.SIM_ENV_SECRET_ID, +} + +interface MockSqlQuery { + toSQL(): { sql: string } +} + +function restoreEnvironmentVariable(key: keyof typeof ORIGINAL_ENV): void { + const value = ORIGINAL_ENV[key] + if (value === undefined) { + Reflect.deleteProperty(process.env, key) + } else { + process.env[key] = value + } +} + +function candidate(workflowId: string): TableWorkflowDeploymentCandidate { + return { + workflowId, + workspaceId: 'workspace-1', + userId: 'user-1', + } +} + +function store( + overrides: Partial = {} +): TableWorkflowDeploymentStore { + return { + assertIntegrity: vi.fn().mockResolvedValue(undefined), + listCandidates: vi.fn().mockResolvedValue([]), + isDeployed: vi.fn().mockResolvedValue(false), + ...overrides, + } +} + +describe('backfillTableWorkflowDeployments', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + afterEach(() => { + restoreEnvironmentVariable('DATABASE_URL') + restoreEnvironmentVariable('DATABASE_URL_WEB') + restoreEnvironmentVariable('REDIS_TLS_SERVERNAME') + restoreEnvironmentVariable('REDIS_URL') + restoreEnvironmentVariable('SIM_ENV_SECRET_ID') + }) + + it.each([ + ['production', '/production/sim/env-vars'], + ['staging', '/staging/sim/env-vars'], + ] as const)( + 'loads the %s runtime secret before database modules are needed', + async (environment, runtimeSecretId) => { + Reflect.deleteProperty(process.env, 'DATABASE_URL') + Reflect.deleteProperty(process.env, 'DATABASE_URL_WEB') + Reflect.deleteProperty(process.env, 'REDIS_TLS_SERVERNAME') + Reflect.deleteProperty(process.env, 'REDIS_URL') + Reflect.deleteProperty(process.env, 'SIM_ENV_SECRET_ID') + mockLoadRuntimeSecrets.mockImplementation(async () => { + process.env.DATABASE_URL = `postgres://${environment}/database` + process.env.REDIS_TLS_SERVERNAME = `cache.${environment}.internal` + process.env.REDIS_URL = `rediss://cache.${environment}.internal:6379` + }) + + await prepareTableWorkflowDeploymentBackfillEnvironment([`--environment=${environment}`]) + + expect(process.env.SIM_ENV_SECRET_ID).toBe(runtimeSecretId) + expect(process.env.REDIS_TLS_SERVERNAME).toBeUndefined() + expect(process.env.REDIS_URL).toBeUndefined() + expect(mockLoadRuntimeSecrets).toHaveBeenCalledTimes(1) + } + ) + + it('keeps the existing local DATABASE_URL mode when no environment is requested', async () => { + process.env.DATABASE_URL = 'postgres://local/database' + process.env.REDIS_URL = 'redis://localhost:6379' + + await prepareTableWorkflowDeploymentBackfillEnvironment([]) + + expect(mockLoadRuntimeSecrets).not.toHaveBeenCalled() + expect(process.env.DATABASE_URL).toBe('postgres://local/database') + expect(process.env.REDIS_URL).toBe('redis://localhost:6379') + }) + + it('rejects unsupported, unknown, duplicate, and locally configured staging arguments', async () => { + expect(() => parseTableWorkflowDeploymentBackfillArgs(['--environment=prod'])).toThrow( + 'Unsupported backfill environment: prod' + ) + expect(() => parseTableWorkflowDeploymentBackfillArgs(['--dry-run'])).toThrow( + 'Unknown argument: --dry-run' + ) + expect(() => + parseTableWorkflowDeploymentBackfillArgs(['--environment=staging', '--environment=staging']) + ).toThrow('can only be provided once') + + process.env.DATABASE_URL = 'postgres://local/database' + await expect( + prepareTableWorkflowDeploymentBackfillEnvironment(['--environment=staging']) + ).rejects.toThrow('local configuration cannot override staging') + expect(mockLoadRuntimeSecrets).not.toHaveBeenCalled() + }) + + it('silently excludes missing and archived workflow references from integrity checks', async () => { + await postgresTableWorkflowDeploymentStore.assertIntegrity() + + const referenceQuery = dbChainMockFns.execute.mock.calls[2]?.[0] as MockSqlQuery + const queryText = referenceQuery.toSQL().sql + expect(queryText).toContain('INNER JOIN workflow') + expect(queryText).toContain('workflow.archived_at IS NULL') + expect(queryText).not.toContain('LEFT JOIN workflow') + expect(queryText).not.toContain('workflow.id IS NULL') + }) + + it('deploys bounded keyset pages and verifies the final desired state', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')]) + .mockResolvedValueOnce([candidate('workflow-c')]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]) + const deploymentState = new Map() + const isDeployed = vi + .fn() + .mockImplementation(async (workflowId) => deploymentState.get(workflowId) ?? false) + const deploy = vi.fn(async (workflow: TableWorkflowDeploymentCandidate) => { + deploymentState.set(workflow.workflowId, true) + return { + success: true, + activeDeployment: { + deploymentVersionId: `version-${workflow.workflowId}`, + version: 1, + deployedAt: new Date().toISOString(), + }, + } + }) + const backfillStore = store({ listCandidates, isDeployed }) + + await expect( + backfillTableWorkflowDeployments(backfillStore, deploy, { batchSize: 2 }) + ).resolves.toEqual({ + scanned: 3, + deployed: 3, + alreadyDeployed: 0, + skippedLocked: 0, + }) + expect(listCandidates.mock.calls).toEqual([ + ['', 2], + ['workflow-b', 2], + ['workflow-c', 2], + ['', 1], + ]) + expect(backfillStore.assertIntegrity).toHaveBeenCalledTimes(2) + expect(deploy.mock.calls.map(([workflow]) => workflow.workflowId)).toEqual([ + 'workflow-a', + 'workflow-b', + 'workflow-c', + ]) + }) + + it('does not redeploy an already deployed workflow', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a')]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]) + const deploy = vi.fn() + + await expect( + backfillTableWorkflowDeployments( + store({ + listCandidates, + isDeployed: vi.fn().mockResolvedValue(true), + }), + deploy + ) + ).resolves.toEqual({ + scanned: 1, + deployed: 0, + alreadyDeployed: 1, + skippedLocked: 0, + }) + expect(deploy).not.toHaveBeenCalled() + }) + + it('reports and skips locked workflows while continuing the backfill', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([candidate('workflow-a')]) + .mockResolvedValueOnce([]) + const deploymentState = new Map() + const isDeployed = vi + .fn() + .mockImplementation(async (workflowId) => deploymentState.get(workflowId) ?? false) + const deploy = vi.fn(async (workflow: TableWorkflowDeploymentCandidate) => { + if (workflow.workflowId === 'workflow-a') { + return { + success: false, + error: 'Workflow is locked by its containing folder', + errorCode: 'locked' as const, + } + } + deploymentState.set(workflow.workflowId, true) + return { + success: true, + activeDeployment: { + deploymentVersionId: `version-${workflow.workflowId}`, + version: 1, + deployedAt: new Date().toISOString(), + }, + } + }) + + await expect( + backfillTableWorkflowDeployments(store({ listCandidates, isDeployed }), deploy) + ).resolves.toEqual({ + scanned: 2, + deployed: 1, + alreadyDeployed: 0, + skippedLocked: 1, + }) + expect(deploy.mock.calls.map(([workflow]) => workflow.workflowId)).toEqual([ + 'workflow-a', + 'workflow-b', + ]) + expect(listCandidates.mock.calls.slice(-2)).toEqual([ + ['', 1], + ['workflow-a', 1], + ]) + }) + + it('fails fast when canonical deployment fails', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')]) + const deploy = vi.fn().mockResolvedValue({ + success: false, + error: 'invalid trigger configuration', + }) + + await expect( + backfillTableWorkflowDeployments(store({ listCandidates }), deploy) + ).rejects.toThrow('Failed to deploy table workflow workflow-a: invalid trigger configuration') + expect(deploy).toHaveBeenCalledTimes(1) + }) + + it('fails when deployment does not activate or persist a valid active version', async () => { + const firstList = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a')]) + await expect( + backfillTableWorkflowDeployments(store({ listCandidates: firstList }), async () => ({ + success: true, + activeDeployment: null, + })) + ).rejects.toThrow('did not reach an active deployment state') + + const secondList = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-b')]) + await expect( + backfillTableWorkflowDeployments(store({ listCandidates: secondList }), async () => ({ + success: true, + activeDeployment: { + deploymentVersionId: 'version-b', + version: 1, + deployedAt: new Date().toISOString(), + }, + })) + ).rejects.toThrow('completed without a valid active version') + }) + + it('rejects invalid batch and page behavior before it can loop or skip data', async () => { + const invalidBatchStore = store() + await expect( + backfillTableWorkflowDeployments(invalidBatchStore, vi.fn(), { batchSize: 0 }) + ).rejects.toThrow('positive integer') + expect(invalidBatchStore.assertIntegrity).not.toHaveBeenCalled() + + const oversizedStore = store({ + listCandidates: vi.fn().mockResolvedValue([candidate('workflow-a'), candidate('workflow-b')]), + }) + await expect( + backfillTableWorkflowDeployments(oversizedStore, vi.fn(), { batchSize: 1 }) + ).rejects.toThrow('oversized page') + + const duplicateStore = store({ + listCandidates: vi.fn().mockResolvedValue([candidate('workflow-a'), candidate('workflow-a')]), + }) + await expect( + backfillTableWorkflowDeployments(duplicateStore, vi.fn(), { batchSize: 2 }) + ).rejects.toThrow('duplicate workflow ids') + }) + + it('uses the canonical deployer with backfill attribution and a stable idempotency key', async () => { + mockPerformFullDeploy.mockResolvedValue({ + success: true, + activeDeployment: { + deploymentVersionId: 'version-1', + version: 1, + deployedAt: new Date().toISOString(), + }, + }) + + await deployTableWorkflow(candidate('workflow-1')) + + expect(mockPerformFullDeploy).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + userId: 'user-1', + actorId: 'table-workflow-deployment-backfill', + captureAnalytics: false, + requestId: 'table-workflow-deployment-backfill:v2:workflow-1', + idempotencyKey: 'table-workflow-deployment-backfill:v2:workflow-1', + }) + }) + + it('uses the repository batch-size default', async () => { + const listCandidates = vi.fn().mockResolvedValue([]) + + await backfillTableWorkflowDeployments(store({ listCandidates }), vi.fn()) + + expect(listCandidates).toHaveBeenNthCalledWith(1, '', TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE) + }) +}) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts new file mode 100644 index 00000000000..07f4fd0cf57 --- /dev/null +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -0,0 +1,503 @@ +#!/usr/bin/env bun + +/** + * Deploys every mutable workflow referenced by a table workflow group. + * + * The script is idempotent and resumable: it pages over only workflows that do + * not have the full desired state, and each deployment uses a stable idempotency + * key. It runs the canonical deployment orchestration so webhook, schedule, + * MCP, audit, and notification side effects stay consistent with a + * user-initiated deployment. + * + * Usage: + * DATABASE_URL=... bun run apps/sim/scripts/backfill-table-workflow-deployments.ts + * AWS_PROFILE=sim-admin bun --no-env-file apps/sim/scripts/backfill-table-workflow-deployments.ts --environment=staging + * AWS_PROFILE=sim-admin bun --no-env-file apps/sim/scripts/backfill-table-workflow-deployments.ts --environment=production + */ + +import { createLogger } from '@sim/logger' +import { loadRuntimeSecrets } from '@sim/runtime-secrets' +import { getErrorMessage } from '@sim/utils/errors' +import { sql } from 'drizzle-orm' +import type { PerformFullDeployResult } from '@/lib/workflows/orchestration/deploy' + +const logger = createLogger('BackfillTableWorkflowDeployments') + +export const TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE = 25 +const BACKFILL_ACTOR_ID = 'table-workflow-deployment-backfill' +const BACKFILL_OPERATION_VERSION = 'v2' +const RUNTIME_SECRET_IDS = { + production: '/production/sim/env-vars', + staging: '/staging/sim/env-vars', +} as const +/** Container-private services that a locally executed hosted backfill must not initialize. */ +const LOCAL_HOSTED_OMITTED_VARIABLES = ['REDIS_URL', 'REDIS_TLS_SERVERNAME'] as const + +type TableWorkflowDeploymentBackfillEnvironment = keyof typeof RUNTIME_SECRET_IDS + +interface TableWorkflowDeploymentBackfillCliOptions { + environment?: TableWorkflowDeploymentBackfillEnvironment +} + +export interface TableWorkflowDeploymentCandidate { + workflowId: string + workspaceId: string + userId: string +} + +export interface TableWorkflowDeploymentStore { + assertIntegrity(): Promise + listCandidates( + afterWorkflowId: string, + limit: number + ): Promise + isDeployed(workflowId: string): Promise +} + +export interface TableWorkflowDeploymentSummary { + scanned: number + deployed: number + alreadyDeployed: number + skippedLocked: number +} + +interface TableWorkflowDeploymentBackfillOptions { + batchSize?: number +} + +export type DeployTableWorkflow = ( + candidate: TableWorkflowDeploymentCandidate +) => Promise + +interface InvalidTableSchemaRow extends Record { + table_id: string +} + +interface InvalidWorkflowGroupRow extends Record { + group_index: string + table_id: string +} + +interface InvalidWorkflowReferenceRow extends Record { + table_id: string + table_workspace_id: string + workflow_id: string + workflow_workspace_id: string +} + +interface MultipleActiveVersionsRow extends Record { + active_version_count: number + workflow_id: string +} + +interface CandidateRow extends Record { + user_id: string + workflow_id: string + workspace_id: string +} + +interface DeploymentStateRow extends Record { + active_version_count: number + is_deployed: boolean +} + +function isTableWorkflowDeploymentBackfillEnvironment( + value: string +): value is TableWorkflowDeploymentBackfillEnvironment { + return Object.hasOwn(RUNTIME_SECRET_IDS, value) +} + +/** Parses the deliberately small CLI surface for the backfill. */ +export function parseTableWorkflowDeploymentBackfillArgs( + args: readonly string[] +): TableWorkflowDeploymentBackfillCliOptions { + let environment: TableWorkflowDeploymentBackfillCliOptions['environment'] + + for (const arg of args) { + if (!arg.startsWith('--environment=')) { + throw new Error(`Unknown argument: ${arg}`) + } + if (environment) { + throw new Error('The --environment argument can only be provided once') + } + + const requestedEnvironment = arg.slice('--environment='.length) + if (!isTableWorkflowDeploymentBackfillEnvironment(requestedEnvironment)) { + throw new Error(`Unsupported backfill environment: ${requestedEnvironment || '(empty)'}`) + } + environment = requestedEnvironment + } + + return { environment } +} + +/** Loads staging configuration before modules that read database settings are imported. */ +export async function prepareTableWorkflowDeploymentBackfillEnvironment( + args: readonly string[] +): Promise { + const { environment } = parseTableWorkflowDeploymentBackfillArgs(args) + if (!environment) return + + const runtimeSecretId = RUNTIME_SECRET_IDS[environment] + const configuredSecretId = process.env.SIM_ENV_SECRET_ID + if (configuredSecretId && configuredSecretId !== runtimeSecretId) { + throw new Error( + `SIM_ENV_SECRET_ID is already set to ${configuredSecretId}; expected ${runtimeSecretId}` + ) + } + + const configuredDatabaseVariables = ['DATABASE_URL', 'DATABASE_URL_WEB'].filter( + (key) => key in process.env + ) + if (configuredDatabaseVariables.length > 0) { + throw new Error( + `Unset ${configuredDatabaseVariables.join(', ')} before using --environment=${environment} so local configuration cannot override ${environment}` + ) + } + + process.env.SIM_ENV_SECRET_ID = runtimeSecretId + await loadRuntimeSecrets() + + if (!process.env.DATABASE_URL && !process.env.DATABASE_URL_WEB) { + throw new Error(`${runtimeSecretId} did not provide a database URL`) + } + + for (const key of LOCAL_HOSTED_OMITTED_VARIABLES) { + Reflect.deleteProperty(process.env, key) + } +} + +async function getDatabase() { + const { db } = await import('@sim/db') + return db +} + +function validateCandidatePage( + candidates: TableWorkflowDeploymentCandidate[], + afterWorkflowId: string, + limit: number +): string | null { + if (candidates.length === 0) return null + if (candidates.length > limit) { + throw new Error('Table workflow deployment store returned an oversized page') + } + + const pageIds = new Set(candidates.map((candidate) => candidate.workflowId)) + if (pageIds.size !== candidates.length) { + throw new Error('Table workflow deployment store returned duplicate workflow ids') + } + + const lastWorkflowId = candidates.at(-1)?.workflowId + if (!lastWorkflowId || lastWorkflowId === afterWorkflowId) { + throw new Error('Table workflow deployment store returned a non-advancing page') + } + return lastWorkflowId +} + +async function assertOnlyLockedCandidatesRemain( + store: TableWorkflowDeploymentStore, + lockedWorkflowIds: ReadonlySet +): Promise { + let afterWorkflowId = '' + for (;;) { + const candidates = await store.listCandidates(afterWorkflowId, 1) + const lastWorkflowId = validateCandidatePage(candidates, afterWorkflowId, 1) + if (!lastWorkflowId) return + + const unexpectedCandidate = candidates.find( + (candidate) => !lockedWorkflowIds.has(candidate.workflowId) + ) + if (unexpectedCandidate) { + throw new Error( + `Table workflow deployment backfill left workflow ${unexpectedCandidate.workflowId} undeployed` + ) + } + afterWorkflowId = lastWorkflowId + } +} + +/** Ensures the table group references can be traversed without silently dropping corrupt data. */ +async function assertTableWorkflowIntegrity(): Promise { + const db = await getDatabase() + const [invalidSchema] = await db.execute(sql` + SELECT id AS table_id + FROM user_table_definitions + WHERE schema ? 'workflowGroups' + AND jsonb_typeof(schema->'workflowGroups') IS DISTINCT FROM 'array' + LIMIT 1 + `) + if (invalidSchema) { + throw new Error( + `Table ${invalidSchema.table_id} has a workflowGroups value that is not an array` + ) + } + + const [invalidGroup] = await db.execute(sql` + SELECT + table_definition.id AS table_id, + workflow_group.ordinality::text AS group_index + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) WITH ORDINALITY AS workflow_group(value, ordinality) + WHERE jsonb_typeof(workflow_group.value) IS DISTINCT FROM 'object' + OR NOT (workflow_group.value ? 'workflowId') + OR jsonb_typeof(workflow_group.value->'workflowId') IS DISTINCT FROM 'string' + OR ( + workflow_group.value->>'workflowId' = '' + AND ( + NOT (workflow_group.value ? 'enrichmentId') + OR jsonb_typeof(workflow_group.value->'enrichmentId') IS DISTINCT FROM 'string' + OR workflow_group.value->>'enrichmentId' = '' + ) + ) + LIMIT 1 + `) + if (invalidGroup) { + throw new Error( + `Table ${invalidGroup.table_id} workflow group ${invalidGroup.group_index} has an invalid workflowId` + ) + } + + const [invalidReference] = await db.execute(sql` + WITH table_workflow_groups AS ( + SELECT + table_definition.id AS table_id, + table_definition.workspace_id AS table_workspace_id, + workflow_group.value->>'workflowId' AS workflow_id + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + ) + SELECT + table_workflow_groups.table_id, + table_workflow_groups.table_workspace_id, + table_workflow_groups.workflow_id, + workflow.workspace_id AS workflow_workspace_id + FROM table_workflow_groups + INNER JOIN workflow ON workflow.id = table_workflow_groups.workflow_id + AND workflow.archived_at IS NULL + WHERE table_workflow_groups.workflow_id <> '' + AND workflow.workspace_id IS DISTINCT FROM table_workflow_groups.table_workspace_id + LIMIT 1 + `) + if (invalidReference) { + throw new Error( + `Table ${invalidReference.table_id} in workspace ${invalidReference.table_workspace_id} ` + + `references workflow ${invalidReference.workflow_id} in workspace ${invalidReference.workflow_workspace_id}` + ) + } + + const [multipleActiveVersions] = await db.execute(sql` + WITH referenced_workflow_ids AS ( + SELECT DISTINCT workflow_group.value->>'workflowId' AS workflow_id + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + INNER JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' + AND workflow.archived_at IS NULL + WHERE workflow_group.value->>'workflowId' <> '' + ) + SELECT + deployment_version.workflow_id, + COUNT(*)::int AS active_version_count + FROM workflow_deployment_version AS deployment_version + INNER JOIN referenced_workflow_ids + ON referenced_workflow_ids.workflow_id = deployment_version.workflow_id + WHERE deployment_version.is_active = true + GROUP BY deployment_version.workflow_id + HAVING COUNT(*) > 1 + LIMIT 1 + `) + if (multipleActiveVersions) { + throw new Error( + `Workflow ${multipleActiveVersions.workflow_id} has ${multipleActiveVersions.active_version_count} active deployment versions` + ) + } +} + +export const postgresTableWorkflowDeploymentStore: TableWorkflowDeploymentStore = { + assertIntegrity: assertTableWorkflowIntegrity, + + async listCandidates(afterWorkflowId, limit) { + const db = await getDatabase() + const rows = await db.execute(sql` + WITH referenced_workflow_ids AS ( + SELECT DISTINCT workflow_group.value->>'workflowId' AS workflow_id + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + WHERE workflow_group.value->>'workflowId' <> '' + ) + SELECT + workflow.id AS workflow_id, + workflow.workspace_id AS workspace_id, + workflow.user_id + FROM referenced_workflow_ids + INNER JOIN workflow ON workflow.id = referenced_workflow_ids.workflow_id + AND workflow.archived_at IS NULL + WHERE workflow.id COLLATE "C" > ${afterWorkflowId}::text COLLATE "C" + AND ( + workflow.is_deployed = false + OR NOT EXISTS ( + SELECT 1 + FROM workflow_deployment_version AS active_version + WHERE active_version.workflow_id = workflow.id + AND active_version.is_active = true + ) + ) + ORDER BY workflow.id COLLATE "C" + LIMIT ${limit} + `) + + return rows.map((row) => ({ + workflowId: row.workflow_id, + workspaceId: row.workspace_id, + userId: row.user_id, + })) + }, + + async isDeployed(workflowId) { + const db = await getDatabase() + const [state] = await db.execute(sql` + SELECT + workflow.is_deployed, + (COUNT(deployment_version.id) FILTER (WHERE deployment_version.is_active))::int + AS active_version_count + FROM workflow + LEFT JOIN workflow_deployment_version AS deployment_version + ON deployment_version.workflow_id = workflow.id + WHERE workflow.id = ${workflowId} + GROUP BY workflow.id, workflow.is_deployed + `) + if (!state) { + throw new Error(`Workflow ${workflowId} disappeared during the deployment backfill`) + } + if (state.active_version_count > 1) { + throw new Error( + `Workflow ${workflowId} has ${state.active_version_count} active deployment versions` + ) + } + return state.is_deployed && state.active_version_count === 1 + }, +} + +/** Deploys one table workflow through the same orchestration used by application surfaces. */ +export async function deployTableWorkflow( + candidate: TableWorkflowDeploymentCandidate +): Promise { + const { performFullDeploy } = await import('@/lib/workflows/orchestration/deploy') + return performFullDeploy({ + workflowId: candidate.workflowId, + userId: candidate.userId, + actorId: BACKFILL_ACTOR_ID, + captureAnalytics: false, + requestId: `${BACKFILL_ACTOR_ID}:${BACKFILL_OPERATION_VERSION}:${candidate.workflowId}`, + idempotencyKey: `${BACKFILL_ACTOR_ID}:${BACKFILL_OPERATION_VERSION}:${candidate.workflowId}`, + }) +} + +/** + * Reaches and verifies that every mutable table workflow has an active deployment. + */ +export async function backfillTableWorkflowDeployments( + store: TableWorkflowDeploymentStore, + deploy: DeployTableWorkflow, + options: TableWorkflowDeploymentBackfillOptions = {} +): Promise { + const batchSize = options.batchSize ?? TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE + if (!Number.isInteger(batchSize) || batchSize <= 0) { + throw new Error('Table workflow deployment backfill batch size must be a positive integer') + } + + await store.assertIntegrity() + + const summary: TableWorkflowDeploymentSummary = { + scanned: 0, + deployed: 0, + alreadyDeployed: 0, + skippedLocked: 0, + } + const lockedWorkflowIds = new Set() + let afterWorkflowId = '' + + for (;;) { + const candidates = await store.listCandidates(afterWorkflowId, batchSize) + const lastWorkflowId = validateCandidatePage(candidates, afterWorkflowId, batchSize) + if (!lastWorkflowId) break + + for (const candidate of candidates) { + summary.scanned += 1 + if (await store.isDeployed(candidate.workflowId)) { + summary.alreadyDeployed += 1 + } else { + logger.info('Deploying workflow referenced by a table workflow group', { + workflowId: candidate.workflowId, + workspaceId: candidate.workspaceId, + }) + const result = await deploy(candidate) + if (!result.success) { + if (result.errorCode === 'locked') { + lockedWorkflowIds.add(candidate.workflowId) + summary.skippedLocked += 1 + logger.warn('Skipping locked workflow referenced by a table workflow group', { + workflowId: candidate.workflowId, + workspaceId: candidate.workspaceId, + reason: result.error ?? 'Workflow is locked', + }) + continue + } + throw new Error( + `Failed to deploy table workflow ${candidate.workflowId}: ${result.error ?? 'deployment returned no error'}` + ) + } + if (!result.activeDeployment) { + throw new Error( + `Table workflow ${candidate.workflowId} did not reach an active deployment state` + ) + } + if (!(await store.isDeployed(candidate.workflowId))) { + throw new Error( + `Table workflow ${candidate.workflowId} deployment completed without a valid active version` + ) + } + summary.deployed += 1 + } + } + + afterWorkflowId = lastWorkflowId + } + + await store.assertIntegrity() + await assertOnlyLockedCandidatesRemain(store, lockedWorkflowIds) + + return summary +} + +export async function runTableWorkflowDeploymentBackfill(): Promise { + logger.info('Starting table workflow deployment backfill') + const summary = await backfillTableWorkflowDeployments( + postgresTableWorkflowDeploymentStore, + deployTableWorkflow + ) + logger.info('Table workflow deployment backfill completed', summary) +} + +async function main(): Promise { + await prepareTableWorkflowDeploymentBackfillEnvironment(process.argv.slice(2)) + await runTableWorkflowDeploymentBackfill() +} + +if ((import.meta as { main?: boolean }).main) { + main() + .then(() => process.exit(0)) + .catch((error: unknown) => { + logger.error('Table workflow deployment backfill failed', { + error: getErrorMessage(error), + }) + process.exit(1) + }) +}