diff --git a/apps/realtime/src/database/operations.ts b/apps/realtime/src/database/operations.ts index 7f514bda77f..697ba4099c0 100644 --- a/apps/realtime/src/database/operations.ts +++ b/apps/realtime/src/database/operations.ts @@ -24,7 +24,15 @@ import { import { randomFloat } from '@sim/utils/random' import { loadWorkflowFromNormalizedTablesRaw } from '@sim/workflow-persistence/load' import { mergeSubBlockValues } from '@sim/workflow-persistence/subblocks' -import { isWorkflowBlockProtected } from '@sim/workflow-types/workflow' +import { + filterAcyclicEdges, + filterUniqueWorkflowEdges, + getWorkflowBlockNameConflict, + getWorkflowEdgeScopeDropReason, + isKnownWorkflowTriggerBlock, + isWorkflowAnnotationOnlyBlockType, + isWorkflowBlockProtected, +} from '@sim/workflow-types/workflow' import { and, eq, inArray, isNull, or, sql } from 'drizzle-orm' import { drizzle } from 'drizzle-orm/postgres-js' import postgres from 'postgres' @@ -32,6 +40,171 @@ import { env } from '@/env' const logger = createLogger('SocketDatabase') +interface PersistedEdgeRecord { + sourceBlockId: string + targetBlockId: string + sourceHandle: string | null + targetHandle: string | null +} + +function toEdgeHandles(edge: PersistedEdgeRecord) { + return { + source: edge.sourceBlockId, + target: edge.targetBlockId, + sourceHandle: edge.sourceHandle, + targetHandle: edge.targetHandle, + } +} + +interface EdgeAddCandidate { + id?: string + source: string + target: string + sourceHandle?: string | null + targetHandle?: string | null +} + +interface FilterEdgesForPersistResult { + safeEdges: T[] + droppedCounts: Record + droppedDuplicates: number + droppedCyclic: number +} + +/** + * Runs the full server-side edge-add validation pipeline — missing block, + * protected target, annotation-only block, trigger-block target, scope + * boundary, duplicate, cycle — against one or more candidate edges. Shared by + * the single-edge `ADD` and `BATCH_ADD_EDGES` handlers so the two paths + * can't drift out of sync with each other, the same way the client and + * realtime layers already share these rules via `@sim/workflow-types`. + */ +async function filterEdgesForPersist( + tx: any, + workflowId: string, + candidates: T[] +): Promise> { + const connectedBlockIds = new Set() + for (const edge of candidates) { + connectedBlockIds.add(edge.source) + connectedBlockIds.add(edge.target) + } + + const connectedBlocks = await tx + .select({ + id: workflowBlocks.id, + type: workflowBlocks.type, + locked: workflowBlocks.locked, + triggerMode: workflowBlocks.triggerMode, + data: workflowBlocks.data, + }) + .from(workflowBlocks) + .where( + and( + eq(workflowBlocks.workflowId, workflowId), + inArray(workflowBlocks.id, Array.from(connectedBlockIds)) + ) + ) + + type EdgeBlockRecord = (typeof connectedBlocks)[number] + const blocksById: Record = Object.fromEntries( + connectedBlocks.map((b: EdgeBlockRecord) => [b.id, b]) + ) + + const parentIds = new Set() + for (const block of connectedBlocks) { + const parentId = (block.data as Record | null)?.parentId as string | undefined + if (parentId && !blocksById[parentId]) { + parentIds.add(parentId) + } + } + + if (parentIds.size > 0) { + const parentBlocks = await tx + .select({ + id: workflowBlocks.id, + type: workflowBlocks.type, + locked: workflowBlocks.locked, + data: workflowBlocks.data, + }) + .from(workflowBlocks) + .where( + and( + eq(workflowBlocks.workflowId, workflowId), + inArray(workflowBlocks.id, Array.from(parentIds)) + ) + ) + for (const b of parentBlocks) { + blocksById[b.id] = b + } + } + + const droppedCounts: Record = {} + const countDrop = (reason: string) => { + droppedCounts[reason] = (droppedCounts[reason] ?? 0) + 1 + } + + // Mirrors the client's validateEdges rule order: missing block, protected + // target, annotation-only block, trigger-block target, scope boundary. + // `scopeDropReason`'s detail (which names the specific block ids/scopes) + // is dropped in favor of a stable 'scope boundary' key so droppedCounts + // aggregates across edges instead of gaining one free-text key per edge. + const structurallyValidEdges = candidates.filter((edge) => { + const sourceBlock = blocksById[edge.source] + const targetBlock = blocksById[edge.target] + + if (!sourceBlock || !targetBlock) { + countDrop('missing block') + return false + } + if (isWorkflowBlockProtected(edge.target, blocksById)) { + countDrop('protected target block') + return false + } + if ( + isWorkflowAnnotationOnlyBlockType(sourceBlock.type) || + isWorkflowAnnotationOnlyBlockType(targetBlock.type) + ) { + countDrop('annotation-only block') + return false + } + if (isKnownWorkflowTriggerBlock(targetBlock)) { + countDrop('trigger block target') + return false + } + if (getWorkflowEdgeScopeDropReason(edge, blocksById)) { + countDrop('scope boundary') + return false + } + return true + }) + + if (structurallyValidEdges.length === 0) { + return { safeEdges: [], droppedCounts, droppedDuplicates: 0, droppedCyclic: 0 } + } + + const existingEdgesForCycleCheck: PersistedEdgeRecord[] = await tx + .select({ + sourceBlockId: workflowEdges.sourceBlockId, + targetBlockId: workflowEdges.targetBlockId, + sourceHandle: workflowEdges.sourceHandle, + targetHandle: workflowEdges.targetHandle, + }) + .from(workflowEdges) + .where(eq(workflowEdges.workflowId, workflowId)) + const existingEdgeEndpoints = existingEdgesForCycleCheck.map(toEdgeHandles) + + const uniqueEdges = filterUniqueWorkflowEdges(structurallyValidEdges, existingEdgeEndpoints) + const safeEdges = filterAcyclicEdges(uniqueEdges, existingEdgeEndpoints) + + return { + safeEdges, + droppedCounts, + droppedDuplicates: structurallyValidEdges.length - uniqueEdges.length, + droppedCyclic: uniqueEdges.length - safeEdges.length, + } +} + // Both realtime pools (this socketDb + the shared @sim/db pool) resolve the // realtime-keyed URL when set, falling back to the shared DATABASE_URL. const connectionString = @@ -229,6 +402,15 @@ export async function persistWorkflowOperation(workflowId: string, operation: an } await db.transaction(async (tx) => { + // This UPDATE is also this workflow's write-serialization point, not + // just a timestamp bump: it takes a row lock on `workflow` for the + // rest of the transaction, so two concurrent persistWorkflowOperation + // calls for the same workflowId cannot interleave — the second blocks + // here until the first commits or rolls back. Every handler dispatched + // below (including the edge-add validate-then-insert sequence in + // filterEdgesForPersist) relies on this to read a consistent snapshot; + // do not make this UPDATE conditional/skippable as an optimization + // without replacing the serialization it provides. await tx .update(workflow) .set({ updatedAt: new Date(timestamp) }) @@ -406,6 +588,22 @@ async function handleBlockOperationTx( } } + const siblingBlocks = await tx + .select({ id: workflowBlocks.id, name: workflowBlocks.name }) + .from(workflowBlocks) + .where(eq(workflowBlocks.workflowId, workflowId)) + const siblingNamesById: Record = Object.fromEntries( + siblingBlocks.map((b: { id: string; name: string }) => [b.id, b.name]) + ) + + const nameConflict = getWorkflowBlockNameConflict(payload.id, payload.name, siblingNamesById) + if (nameConflict) { + logger.info( + `Skipping rename of block ${payload.id} - name conflict: ${nameConflict.reason}` + ) + break + } + await tx .update(workflowBlocks) .set({ @@ -820,27 +1018,54 @@ async function handleBlocksOperationTx( } if (edges && edges.length > 0) { - const edgeValues = edges.map((edge: Record) => ({ - id: edge.id as string, - workflowId, - sourceBlockId: edge.source as string, - targetBlockId: edge.target as string, - sourceHandle: (edge.sourceHandle as string | null) || null, - targetHandle: (edge.targetHandle as string | null) || null, - })) + // Runs after the block insert above, so filterEdgesForPersist's + // blocksById lookup (a plain `tx.select` from `workflowBlocks`) also + // sees the blocks this same batch just inserted — reads observe a + // transaction's own prior writes. + const candidates: EdgeAddCandidate[] = (edges as Array>).map( + (e) => ({ + id: e.id as string, + source: e.source as string, + target: e.target as string, + sourceHandle: (e.sourceHandle as string | null) ?? null, + targetHandle: (e.targetHandle as string | null) ?? null, + }) + ) - await tx - .insert(workflowEdges) - .values(edgeValues) - .onConflictDoUpdate({ - target: workflowEdges.id, - set: { - sourceBlockId: sql`excluded.source_block_id`, - targetBlockId: sql`excluded.target_block_id`, - sourceHandle: sql`excluded.source_handle`, - targetHandle: sql`excluded.target_handle`, - }, + const { safeEdges, droppedCounts, droppedDuplicates, droppedCyclic } = + await filterEdgesForPersist(tx, workflowId, candidates) + + if (safeEdges.length < edges.length) { + logger.info(`Dropped ${edges.length - safeEdges.length} invalid edge(s)`, { + droppedCounts, + droppedDuplicates, + droppedCyclic, }) + } + + if (safeEdges.length > 0) { + const edgeValues = safeEdges.map((edge) => ({ + id: edge.id as string, + workflowId, + sourceBlockId: edge.source, + targetBlockId: edge.target, + sourceHandle: edge.sourceHandle || null, + targetHandle: edge.targetHandle || null, + })) + + await tx + .insert(workflowEdges) + .values(edgeValues) + .onConflictDoUpdate({ + target: workflowEdges.id, + set: { + sourceBlockId: sql`excluded.source_block_id`, + targetBlockId: sql`excluded.target_block_id`, + sourceHandle: sql`excluded.source_handle`, + targetHandle: sql`excluded.target_handle`, + }, + }) + } } if (loops && Object.keys(loops).length > 0) { @@ -1282,57 +1507,28 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str throw new Error('Missing required fields for add edge operation') } - const edgeBlocks = await tx - .select({ - id: workflowBlocks.id, - locked: workflowBlocks.locked, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, [payload.source, payload.target]) - ) - ) - - type EdgeBlockRecord = (typeof edgeBlocks)[number] - const blocksById: Record = Object.fromEntries( - edgeBlocks.map((b: EdgeBlockRecord) => [b.id, b]) + const { safeEdges, droppedCounts, droppedDuplicates } = await filterEdgesForPersist( + tx, + workflowId, + [ + { + id: payload.id, + source: payload.source, + target: payload.target, + sourceHandle: payload.sourceHandle ?? null, + targetHandle: payload.targetHandle ?? null, + }, + ] ) - const parentIds = new Set() - for (const block of edgeBlocks) { - const parentId = (block.data as Record | null)?.parentId as - | string - | undefined - if (parentId && !blocksById[parentId]) { - parentIds.add(parentId) - } - } - - // Fetch parent blocks if needed - if (parentIds.size > 0) { - const parentBlocks = await tx - .select({ - id: workflowBlocks.id, - locked: workflowBlocks.locked, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, Array.from(parentIds)) - ) - ) - for (const b of parentBlocks) { - blocksById[b.id] = b - } - } - - if (isWorkflowBlockProtected(payload.target, blocksById)) { - logger.info(`Skipping edge add - target block is protected`) + if (safeEdges.length === 0) { + const [structuralDropReason] = Object.keys(droppedCounts) + const reason = structuralDropReason + ? structuralDropReason + : droppedDuplicates > 0 + ? 'duplicate edge already exists' + : `would create a cycle: ${payload.source} -> ${payload.target}` + logger.info(`Skipping edge add - ${reason}`) break } @@ -1557,81 +1753,37 @@ async function handleEdgesOperationTx( logger.info(`Batch adding ${edges.length} edges to workflow ${workflowId}`) - // Get all connected block IDs to check lock status - const connectedBlockIds = new Set() - edges.forEach((e: Record) => { - connectedBlockIds.add(e.source as string) - connectedBlockIds.add(e.target as string) - }) - - // Fetch blocks to check lock status - const connectedBlocks = await tx - .select({ - id: workflowBlocks.id, - locked: workflowBlocks.locked, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, Array.from(connectedBlockIds)) - ) - ) - - type AddEdgeBlockRecord = (typeof connectedBlocks)[number] - const blocksById: Record = Object.fromEntries( - connectedBlocks.map((b: AddEdgeBlockRecord) => [b.id, b]) - ) + const candidates: EdgeAddCandidate[] = (edges as Array>).map((e) => ({ + id: e.id as string, + source: e.source as string, + target: e.target as string, + sourceHandle: (e.sourceHandle as string | null) ?? null, + targetHandle: (e.targetHandle as string | null) ?? null, + })) - // Collect parent IDs that need to be fetched - const parentIds = new Set() - for (const block of connectedBlocks) { - const parentId = (block.data as Record | null)?.parentId as - | string - | undefined - if (parentId && !blocksById[parentId]) { - parentIds.add(parentId) - } - } + const { safeEdges, droppedCounts, droppedDuplicates, droppedCyclic } = + await filterEdgesForPersist(tx, workflowId, candidates) - // Fetch parent blocks if needed - if (parentIds.size > 0) { - const parentBlocks = await tx - .select({ - id: workflowBlocks.id, - locked: workflowBlocks.locked, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, Array.from(parentIds)) - ) - ) - for (const b of parentBlocks) { - blocksById[b.id] = b - } + if (safeEdges.length < edges.length) { + logger.info(`Dropped ${edges.length - safeEdges.length} invalid edge(s)`, { + droppedCounts, + droppedDuplicates, + droppedCyclic, + }) } - // Filter edges - only add edges where target block is not protected - const safeEdges = (edges as Array>).filter( - (e) => !isWorkflowBlockProtected(e.target as string, blocksById) - ) - if (safeEdges.length === 0) { - logger.info('All edges target protected blocks, skipping add') + logger.info('All edges were invalid, duplicate, or would create a cycle, skipping add') return } - const edgeValues = safeEdges.map((edge: Record) => ({ + const edgeValues = safeEdges.map((edge) => ({ id: edge.id as string, workflowId, - sourceBlockId: edge.source as string, - targetBlockId: edge.target as string, - sourceHandle: (edge.sourceHandle as string | null) || null, - targetHandle: (edge.targetHandle as string | null) || null, + sourceBlockId: edge.source, + targetBlockId: edge.target, + sourceHandle: edge.sourceHandle || null, + targetHandle: edge.targetHandle || null, })) await tx diff --git a/apps/sim/executor/constants.ts b/apps/sim/executor/constants.ts index b5883c61905..affd77355c8 100644 --- a/apps/sim/executor/constants.ts +++ b/apps/sim/executor/constants.ts @@ -1,3 +1,7 @@ +import { + normalizeWorkflowBlockName, + RESERVED_WORKFLOW_BLOCK_NAMES, +} from '@sim/workflow-types/workflow' import { getMaxExecutionTimeout } from '@/lib/core/execution-limits' import type { LoopType, ParallelType } from '@/lib/workflows/types' @@ -151,11 +155,12 @@ export const SPECIAL_REFERENCE_PREFIXES = [ REFERENCE.PREFIX.VARIABLE, ] as const -export const RESERVED_BLOCK_NAMES = [ - REFERENCE.PREFIX.LOOP, - REFERENCE.PREFIX.PARALLEL, - REFERENCE.PREFIX.VARIABLE, -] as const +/** + * Delegates to the shared implementation in `@sim/workflow-types` so the + * client store and the realtime persistence layer agree on the same reserved + * names. Values intentionally mirror REFERENCE.PREFIX.{LOOP,PARALLEL,VARIABLE} above. + */ +export const RESERVED_BLOCK_NAMES = RESERVED_WORKFLOW_BLOCK_NAMES export const LOOP_REFERENCE = { ITERATION: 'iteration', @@ -478,12 +483,10 @@ export function escapeRegExp(value: string): string { * spaces and dots. Used for both block names and variable names to ensure * consistent matching. * - * Dots are stripped because `.` is the reference path delimiter — a name like - * "Trigger.dev 1" must normalize to "triggerdev1" so the reference - * `` parses unambiguously. Dotted names could never be - * referenced before (the first path segment cut the name at the dot), so - * stripping dots cannot break any previously working reference. + * Delegates to the shared implementation in `@sim/workflow-types` so the + * client store and the realtime persistence layer normalize block names + * identically when checking for reserved/duplicate names. */ export function normalizeName(name: string): string { - return name.toLowerCase().replace(/\s+/g, '').replace(/\./g, '') + return normalizeWorkflowBlockName(name) } diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index f24ed617827..ebab233c7dd 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -12,6 +12,7 @@ import { WORKFLOW_OPERATIONS, } from '@sim/realtime-protocol/constants' import { generateId } from '@sim/utils/id' +import { getWorkflowBlockNameConflict } from '@sim/workflow-types/workflow' import { useQueryClient } from '@tanstack/react-query' import { isEqual } from 'es-toolkit' import type { Edge } from 'reactflow' @@ -27,7 +28,6 @@ import { isSyntheticToolSubBlockId } from '@/lib/workflows/tool-input/synthetic- import { useSocket } from '@/app/workspace/providers/socket-provider' import { getBlock } from '@/blocks' import { getSubBlocksDependingOnChange } from '@/blocks/utils' -import { normalizeName, RESERVED_BLOCK_NAMES } from '@/executor/constants' import { invalidateDeploymentQueries } from '@/hooks/queries/deployments' import { useUndoRedo } from '@/hooks/use-undo-redo' import { @@ -54,7 +54,11 @@ import type { Position, WorkflowState, } from '@/stores/workflows/workflow/types' -import { findAllDescendantNodes, isBlockProtected } from '@/stores/workflows/workflow/utils' +import { + filterAcyclicEdges, + findAllDescendantNodes, + isBlockProtected, +} from '@/stores/workflows/workflow/utils' const logger = createLogger('CollaborativeWorkflow') @@ -1028,27 +1032,26 @@ export function useCollaborativeWorkflow() { } const trimmedName = name.trim() - const normalizedNewName = normalizeName(trimmedName) + const currentBlocks = useWorkflowStore.getState().blocks + const siblingNamesById = Object.fromEntries( + Object.entries(currentBlocks).map(([blockId, b]) => [blockId, b.name]) + ) + const conflict = getWorkflowBlockNameConflict(id, trimmedName, siblingNamesById) - if (!normalizedNewName) { + if (conflict?.reason === 'empty') { logger.error('Cannot rename block to empty name') toast.error('Block name cannot be empty') return { success: false, error: 'Block name cannot be empty' } } - if ((RESERVED_BLOCK_NAMES as readonly string[]).includes(normalizedNewName)) { + if (conflict?.reason === 'reserved') { logger.error(`Cannot rename block to reserved name: "${trimmedName}"`) toast.error(`"${trimmedName}" is a reserved name and cannot be used`) return { success: false, error: `"${trimmedName}" is a reserved name` } } - const currentBlocks = useWorkflowStore.getState().blocks - const conflictingBlock = Object.entries(currentBlocks).find( - ([blockId, block]) => blockId !== id && normalizeName(block.name) === normalizedNewName - ) - - if (conflictingBlock) { - const conflictName = conflictingBlock[1].name + if (conflict?.reason === 'duplicate') { + const conflictName = currentBlocks[conflict.conflictingBlockId as string].name logger.error(`Cannot rename block to "${trimmedName}" - conflicts with "${conflictName}"`) toast.error(`Block name "${trimmedName}" already exists`) return { success: false, error: `Block name "${trimmedName}" already exists` } @@ -1423,7 +1426,12 @@ export function useCollaborativeWorkflow() { const currentEdges = useWorkflowStore.getState().edges const validEdges = filterValidEdges(edges, blocks) const newEdges = filterNewEdges(validEdges, currentEdges) - if (newEdges.length === 0) return false + // Reject cyclic edges here, before they are queued for realtime/DB + // persistence — the local store also runs this check, but only after + // an unfiltered payload would already be enqueued. Filtering once + // here keeps the queued payload and the local store in agreement. + const acyclicEdges = filterAcyclicEdges(newEdges, currentEdges) + if (acyclicEdges.length === 0) return false const operationId = generateId() @@ -1432,16 +1440,16 @@ export function useCollaborativeWorkflow() { operation: { operation: EDGES_OPERATIONS.BATCH_ADD_EDGES, target: OPERATION_TARGETS.EDGES, - payload: { edges: newEdges }, + payload: { edges: acyclicEdges }, }, workflowId: activeWorkflowId || '', userId: session?.user?.id || 'unknown', }) - useWorkflowStore.getState().batchAddEdges(newEdges, { skipValidation: true }) + useWorkflowStore.getState().batchAddEdges(acyclicEdges, { skipValidation: true }) if (!options?.skipUndoRedo) { - newEdges.forEach((edge) => undoRedo.recordAddEdge(edge.id)) + acyclicEdges.forEach((edge) => undoRedo.recordAddEdge(edge.id)) } return true diff --git a/apps/sim/stores/workflows/utils.ts b/apps/sim/stores/workflows/utils.ts index 5505a7738f1..a57f04b95af 100644 --- a/apps/sim/stores/workflows/utils.ts +++ b/apps/sim/stores/workflows/utils.ts @@ -1,5 +1,6 @@ import { generateId } from '@sim/utils/id' import { mergeSubblockStateWithValues } from '@sim/workflow-persistence/subblocks' +import { filterUniqueWorkflowEdges } from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import { DEFAULT_DUPLICATE_OFFSET } from '@/lib/workflows/autolayout/constants' import { getEffectiveBlockOutputs } from '@/lib/workflows/blocks/block-outputs' @@ -32,16 +33,7 @@ export function filterValidEdges(edges: Edge[], blocks: Record { - if (edge.source === edge.target) return false - return !currentEdges.some( - (e) => - e.source === edge.source && - e.sourceHandle === edge.sourceHandle && - e.target === edge.target && - e.targetHandle === edge.targetHandle - ) - }) + return filterUniqueWorkflowEdges(edgesToAdd, currentEdges) } export interface RegeneratedState { diff --git a/apps/sim/stores/workflows/workflow/edge-validation.test.ts b/apps/sim/stores/workflows/workflow/edge-validation.test.ts new file mode 100644 index 00000000000..d9c4cffda58 --- /dev/null +++ b/apps/sim/stores/workflows/workflow/edge-validation.test.ts @@ -0,0 +1,99 @@ +import type { Edge } from 'reactflow' +import { describe, expect, it } from 'vitest' +import { validateEdges } from '@/stores/workflows/workflow/edge-validation' +import type { BlockState } from '@/stores/workflows/workflow/types' + +function makeBlock(id: string, type: string, overrides?: Partial): BlockState { + return { + id, + type, + name: id, + position: { x: 0, y: 0 }, + subBlocks: {}, + outputs: {}, + enabled: true, + ...overrides, + } +} + +function makeEdge(id: string, source: string, target: string): Edge { + return { id, source, target, type: 'default' } +} + +describe('validateEdges', () => { + it('accepts an edge between two root-scope blocks', () => { + const blocks = { + a: makeBlock('a', 'starter'), + b: makeBlock('b', 'function'), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(1) + expect(result.dropped).toHaveLength(0) + }) + + it('drops an edge referencing a missing block', () => { + const blocks = { a: makeBlock('a', 'starter') } + const result = validateEdges([makeEdge('e1', 'a', 'missing')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toBe('edge references a missing block') + }) + + it('drops an edge touching an annotation-only (note) block', () => { + const blocks = { + a: makeBlock('a', 'note'), + b: makeBlock('b', 'function'), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toBe('edge references an annotation-only block') + }) + + it('drops an edge targeting a trigger block', () => { + const blocks = { + a: makeBlock('a', 'function'), + b: makeBlock('b', 'function', { triggerMode: true }), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toBe('trigger blocks cannot be edge targets') + }) + + it('drops an edge crossing loop scope boundaries', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + inner: makeBlock('inner', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + outer: makeBlock('outer', 'function'), + } + const result = validateEdges([makeEdge('e1', 'inner', 'outer')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toContain('different scopes') + }) + + it('accepts an edge from a loop container into its own child', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + inner: makeBlock('inner', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + } + const result = validateEdges([makeEdge('e1', 'loop', 'inner')], blocks) + expect(result.valid).toHaveLength(1) + }) + + it('accepts an edge from a loop child back out to its own container', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + inner: makeBlock('inner', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + } + const result = validateEdges([makeEdge('e1', 'inner', 'loop')], blocks) + expect(result.valid).toHaveLength(1) + }) + + it('accepts edges between two siblings inside the same loop', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + a: makeBlock('a', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + b: makeBlock('b', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(1) + }) +}) diff --git a/apps/sim/stores/workflows/workflow/edge-validation.ts b/apps/sim/stores/workflows/workflow/edge-validation.ts index 034e87e177d..b6b0539f8cd 100644 --- a/apps/sim/stores/workflows/workflow/edge-validation.ts +++ b/apps/sim/stores/workflows/workflow/edge-validation.ts @@ -1,6 +1,9 @@ +import { + getWorkflowEdgeScopeDropReason, + isWorkflowAnnotationOnlyBlockType, +} from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import { TriggerUtils } from '@/lib/workflows/triggers/triggers' -import { isAnnotationOnlyBlock } from '@/executor/constants' import type { BlockState } from '@/stores/workflows/workflow/types' interface DroppedEdge { @@ -13,40 +16,6 @@ export interface EdgeValidationResult { dropped: DroppedEdge[] } -function isContainerBlock(block: BlockState | undefined): boolean { - return block?.type === 'loop' || block?.type === 'parallel' -} - -function getParentId(block: BlockState | undefined): string | null { - return block?.data?.parentId ?? null -} - -function getScopeDropReason(edge: Edge, blocks: Record): string | null { - const sourceBlock = blocks[edge.source] - const targetBlock = blocks[edge.target] - - if (!sourceBlock || !targetBlock) { - return 'edge references a missing block' - } - - const sourceParent = getParentId(sourceBlock) - const targetParent = getParentId(targetBlock) - - if (sourceParent === targetParent) { - return null - } - - if (targetParent === edge.source && isContainerBlock(sourceBlock)) { - return null - } - - if (sourceParent === edge.target && isContainerBlock(targetBlock)) { - return null - } - - return `blocks are in different scopes (${sourceParent ?? 'root'} -> ${targetParent ?? 'root'})` -} - export function validateEdges( edges: Edge[], blocks: Record @@ -63,7 +32,10 @@ export function validateEdges( continue } - if (isAnnotationOnlyBlock(sourceBlock.type) || isAnnotationOnlyBlock(targetBlock.type)) { + if ( + isWorkflowAnnotationOnlyBlockType(sourceBlock.type) || + isWorkflowAnnotationOnlyBlockType(targetBlock.type) + ) { dropped.push({ edge, reason: 'edge references an annotation-only block' }) continue } @@ -73,7 +45,7 @@ export function validateEdges( continue } - const scopeDropReason = getScopeDropReason(edge, blocks) + const scopeDropReason = getWorkflowEdgeScopeDropReason(edge, blocks) if (scopeDropReason) { dropped.push({ edge, reason: scopeDropReason }) continue diff --git a/apps/sim/stores/workflows/workflow/store.test.ts b/apps/sim/stores/workflows/workflow/store.test.ts index 860418bd7ee..f5072971faf 100644 --- a/apps/sim/stores/workflows/workflow/store.test.ts +++ b/apps/sim/stores/workflows/workflow/store.test.ts @@ -334,6 +334,86 @@ describe('workflow store', () => { const state = useWorkflowStore.getState() expectEdgeCount(state, 1) }) + + it('should not add duplicate connections that appear twice within the same batch', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'End', { x: 200, y: 0 }) + + batchAddEdges([ + { id: 'e1', source: 'block-1', target: 'block-2' }, + { id: 'e2', source: 'block-1', target: 'block-2' }, + ]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 1) + expect(state.edges.some((e) => e.id === 'e1')).toBe(true) + expect(state.edges.some((e) => e.id === 'e2')).toBe(false) + }) + + it('should treat an empty-string handle as equivalent to no handle when deduping', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'End', { x: 200, y: 0 }) + + batchAddEdges([ + { id: 'e1', source: 'block-1', target: 'block-2', sourceHandle: '' }, + { id: 'e2', source: 'block-1', target: 'block-2' }, + ]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 1) + expect(state.edges.some((e) => e.id === 'e1')).toBe(true) + expect(state.edges.some((e) => e.id === 'e2')).toBe(false) + }) + + it('should not add a self-loop edge', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + + batchAddEdges([{ id: 'e1', source: 'block-1', target: 'block-1' }]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 0) + }) + + it('should reject an edge that would create a cycle', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'Middle', { x: 200, y: 0 }) + addBlock('block-3', 'function', 'End', { x: 400, y: 0 }) + + batchAddEdges([{ id: 'e1', source: 'block-1', target: 'block-2' }]) + batchAddEdges([{ id: 'e2', source: 'block-2', target: 'block-3' }]) + // block-3 -> block-1 would close the loop back to block-1 + batchAddEdges([{ id: 'e3', source: 'block-3', target: 'block-1' }]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 2) + expect(state.edges.some((e) => e.id === 'e3')).toBe(false) + }) + + it('should reject a cyclic edge within the same batch', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'Middle', { x: 200, y: 0 }) + addBlock('block-3', 'function', 'End', { x: 400, y: 0 }) + + batchAddEdges([ + { id: 'e1', source: 'block-1', target: 'block-2' }, + { id: 'e2', source: 'block-2', target: 'block-3' }, + { id: 'e3', source: 'block-3', target: 'block-1' }, + ]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 2) + expect(state.edges.some((e) => e.id === 'e3')).toBe(false) + }) }) describe('batchRemoveEdges', () => { @@ -1593,6 +1673,18 @@ describe('workflow store', () => { expect(state.blocks.block2.name).toBe('Employee Length') }) + it('should reject reserved names (loop, parallel, variable)', () => { + const { updateBlockName } = useWorkflowStore.getState() + + for (const reserved of ['loop', 'Parallel', 'VARIABLE']) { + const result = updateBlockName('block1', reserved) + expect(result.success).toBe(false) + } + + const state = useWorkflowStore.getState() + expect(state.blocks.block1.name).toBe('Column AD') + }) + it('should return false when trying to rename a non-existent block', () => { const { updateBlockName } = useWorkflowStore.getState() diff --git a/apps/sim/stores/workflows/workflow/store.ts b/apps/sim/stores/workflows/workflow/store.ts index 75b4953a2bc..02d9a58761e 100644 --- a/apps/sim/stores/workflows/workflow/store.ts +++ b/apps/sim/stores/workflows/workflow/store.ts @@ -1,5 +1,6 @@ import { createLogger } from '@sim/logger' import { generateId } from '@sim/utils/id' +import { getWorkflowBlockNameConflict } from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import { create } from 'zustand' import { devtools } from 'zustand/middleware' @@ -8,7 +9,7 @@ import { getDynamicHandleSubblockType, isDynamicHandleSubblock, } from '@/lib/workflows/dynamic-handle-topology' -import { normalizeName, RESERVED_BLOCK_NAMES } from '@/executor/constants' +import { normalizeName } from '@/executor/constants' import { useSubBlockStore } from '@/stores/workflows/subblock/store' import { filterNewEdges, @@ -25,11 +26,11 @@ import type { } from '@/stores/workflows/workflow/types' import { clampParallelBatchSize, + filterAcyclicEdges, findAllDescendantNodes, generateLoopBlocks, generateParallelBlocks, isBlockProtected, - wouldCreateCycle, } from '@/stores/workflows/workflow/utils' import { normalizeWorkflowState } from '@/stores/workflows/workflow/validation' @@ -401,11 +402,10 @@ export const useWorkflowStore = create()( // Skip validation if already validated by caller (e.g., collaborative layer) const validEdges = options?.skipValidation ? edges : filterValidEdges(edges, blocks) const filtered = filterNewEdges(validEdges, currentEdges) - const newEdges = [...currentEdges] - - for (const edge of filtered) { - if (wouldCreateCycle([...newEdges], edge.source, edge.target)) continue - newEdges.push({ + const acyclicEdges = filterAcyclicEdges(filtered, currentEdges) + const newEdges = [ + ...currentEdges, + ...acyclicEdges.map((edge) => ({ id: edge.id || generateId(), source: edge.source, target: edge.target, @@ -413,8 +413,8 @@ export const useWorkflowStore = create()( targetHandle: edge.targetHandle, type: edge.type || 'default', data: edge.data || {}, - }) - } + })), + ] set({ blocks: { ...blocks }, @@ -657,26 +657,26 @@ export const useWorkflowStore = create()( const oldBlock = get().blocks[id] if (!oldBlock) return { success: false, changedSubblocks: [] } - const normalizedNewName = normalizeName(name) + const currentBlocks = get().blocks + const siblingNamesById = Object.fromEntries( + Object.entries(currentBlocks).map(([blockId, block]) => [blockId, block.name]) + ) + const conflict = getWorkflowBlockNameConflict(id, name, siblingNamesById) - if (!normalizedNewName) { + if (conflict?.reason === 'empty') { logger.error(`Cannot rename block to empty name`) return { success: false, changedSubblocks: [] } } - const currentBlocks = get().blocks - const conflictingBlock = Object.entries(currentBlocks).find( - ([blockId, block]) => blockId !== id && normalizeName(block.name) === normalizedNewName - ) - - if (conflictingBlock) { + if (conflict?.reason === 'duplicate') { + const conflictingBlock = currentBlocks[conflict.conflictingBlockId as string] logger.error( - `Cannot rename block to "${name}" - conflicts with "${conflictingBlock[1].name}"` + `Cannot rename block to "${name}" - conflicts with "${conflictingBlock.name}"` ) return { success: false, changedSubblocks: [] } } - if ((RESERVED_BLOCK_NAMES as readonly string[]).includes(normalizedNewName)) { + if (conflict?.reason === 'reserved') { logger.error(`Cannot rename block to reserved name: "${name}"`) return { success: false, changedSubblocks: [] } } diff --git a/apps/sim/stores/workflows/workflow/utils.ts b/apps/sim/stores/workflows/workflow/utils.ts index a7077dc0903..9413beeea99 100644 --- a/apps/sim/stores/workflows/workflow/utils.ts +++ b/apps/sim/stores/workflows/workflow/utils.ts @@ -1,6 +1,8 @@ import { + filterAcyclicEdges as filterAcyclicWorkflowEdges, isWorkflowBlockAncestorLocked, isWorkflowBlockProtected, + wouldCreateCycle as wouldCreateWorkflowEdgeCycle, } from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import type { BlockState, Loop, Parallel } from '@/stores/workflows/workflow/types' @@ -19,50 +21,20 @@ export function clampParallelBatchSize(batchSize: unknown): number { /** * Check if adding an edge would create a cycle in the graph. - * Uses depth-first search to detect if the source node is reachable from the target node. - * - * @param edges - Current edges in the graph - * @param sourceId - Source node ID of the proposed edge - * @param targetId - Target node ID of the proposed edge - * @returns true if adding this edge would create a cycle + * Delegates to the shared implementation in `@sim/workflow-types` so the + * client store, the collaborative queueing layer, and the realtime + * persistence layer all agree on the same cyclic edges. */ export function wouldCreateCycle(edges: Edge[], sourceId: string, targetId: string): boolean { - if (sourceId === targetId) { - return true - } - - const adjacencyList = new Map() - for (const edge of edges) { - if (!adjacencyList.has(edge.source)) { - adjacencyList.set(edge.source, []) - } - adjacencyList.get(edge.source)!.push(edge.target) - } - - const visited = new Set() - - function canReachSource(currentNode: string): boolean { - if (currentNode === sourceId) { - return true - } - - if (visited.has(currentNode)) { - return false - } - - visited.add(currentNode) - - const neighbors = adjacencyList.get(currentNode) || [] - for (const neighbor of neighbors) { - if (canReachSource(neighbor)) { - return true - } - } - - return false - } + return wouldCreateWorkflowEdgeCycle(edges, sourceId, targetId) +} - return canReachSource(targetId) +/** + * Filters a batch of candidate edges down to the ones that do not create a + * cycle against `currentEdges`, evaluated incrementally within the batch. + */ +export function filterAcyclicEdges(edgesToAdd: Edge[], currentEdges: Edge[]): Edge[] { + return filterAcyclicWorkflowEdges(edgesToAdd, currentEdges) } /** diff --git a/packages/workflow-types/src/workflow.ts b/packages/workflow-types/src/workflow.ts index 06b9692ddba..c4aa764aecf 100644 --- a/packages/workflow-types/src/workflow.ts +++ b/packages/workflow-types/src/workflow.ts @@ -129,6 +129,257 @@ export function isWorkflowBlockProtected( return Boolean(block.locked || isWorkflowBlockAncestorLocked(blockId, blocks)) } +export interface WorkflowEdgeEndpoints { + source: string + target: string +} + +/** + * Checks whether adding an edge would create a cycle in the graph, via DFS + * reachability from the proposed target back to the proposed source. + * + * Shared between the client store, the collaborative queueing layer, and the + * realtime persistence layer so all three agree on the same cyclic edges — + * an edge rejected by one must never be accepted (and persisted) by another. + */ +export function wouldCreateCycle( + edges: WorkflowEdgeEndpoints[], + sourceId: string, + targetId: string +): boolean { + if (sourceId === targetId) { + return true + } + + const adjacencyList = new Map() + for (const edge of edges) { + if (!adjacencyList.has(edge.source)) { + adjacencyList.set(edge.source, []) + } + adjacencyList.get(edge.source)!.push(edge.target) + } + + const visited = new Set() + + function canReachSource(currentNode: string): boolean { + if (currentNode === sourceId) { + return true + } + + if (visited.has(currentNode)) { + return false + } + + visited.add(currentNode) + + const neighbors = adjacencyList.get(currentNode) || [] + for (const neighbor of neighbors) { + if (canReachSource(neighbor)) { + return true + } + } + + return false + } + + return canReachSource(targetId) +} + +/** + * Filters a batch of candidate edges down to the ones that do not create a + * cycle, evaluated incrementally so edges within the same batch that would + * chain into a cycle are also rejected. + */ +export function filterAcyclicEdges( + edgesToAdd: T[], + currentEdges: WorkflowEdgeEndpoints[] +): T[] { + const workingEdges: WorkflowEdgeEndpoints[] = [...currentEdges] + const acyclicEdges: T[] = [] + + for (const edge of edgesToAdd) { + if (wouldCreateCycle(workingEdges, edge.source, edge.target)) continue + workingEdges.push(edge) + acyclicEdges.push(edge) + } + + return acyclicEdges +} + +/** Edge endpoints that also identify the specific handle used on each side. */ +export interface WorkflowEdgeHandles extends WorkflowEdgeEndpoints { + sourceHandle?: string | null + targetHandle?: string | null +} + +// Falsy-coalesce (not nullish-coalesce): persistence normalizes a missing +// handle to `null` via `edge.sourceHandle || null` (see +// apps/realtime/src/database/operations.ts), which also maps `''` to +// `null`. Comparing with `??` would treat `''` and `null` as distinct +// handles pre-insert while both are written as the same `null` value, +// letting a `sourceHandle: ''` edge slip past the duplicate check. +function normalizeWorkflowEdgeHandle(handle: string | null | undefined): string | null { + return handle || null +} + +function isDuplicateWorkflowEdge( + edge: WorkflowEdgeHandles, + existing: WorkflowEdgeHandles +): boolean { + return ( + edge.source === existing.source && + normalizeWorkflowEdgeHandle(edge.sourceHandle) === + normalizeWorkflowEdgeHandle(existing.sourceHandle) && + edge.target === existing.target && + normalizeWorkflowEdgeHandle(edge.targetHandle) === + normalizeWorkflowEdgeHandle(existing.targetHandle) + ) +} + +/** + * Filters a batch of candidate edges down to the ones that are not + * self-loops and do not duplicate an edge already present (same source, + * target, and handles), evaluated incrementally so two duplicate edges + * within the same batch are also caught (only the first survives). Shared + * between the client store, the collaborative queueing layer, and the + * realtime persistence layer. + */ +export function filterUniqueWorkflowEdges( + edgesToAdd: T[], + currentEdges: WorkflowEdgeHandles[] +): T[] { + const workingEdges: WorkflowEdgeHandles[] = [...currentEdges] + const uniqueEdges: T[] = [] + + for (const edge of edgesToAdd) { + if (edge.source === edge.target) continue + if (workingEdges.some((existing) => isDuplicateWorkflowEdge(edge, existing))) continue + workingEdges.push(edge) + uniqueEdges.push(edge) + } + + return uniqueEdges +} + +const WORKFLOW_CONTAINER_BLOCK_TYPES = new Set(['loop', 'parallel']) +const WORKFLOW_ANNOTATION_ONLY_BLOCK_TYPE = 'note' +/** Legacy trigger block type — see TRIGGER_TYPES.STARTER in apps/sim/lib/workflows/triggers/triggers.ts. */ +const LEGACY_STARTER_BLOCK_TYPE = 'starter' + +export interface WorkflowEdgeScopeBlock extends WorkflowLockBlock { + type?: string +} + +/** + * Checks whether an edge crosses loop/parallel scope boundaries — an edge is + * only valid within the same container, or between a container and its + * direct parent/child. + */ +export function getWorkflowEdgeScopeDropReason( + edge: WorkflowEdgeEndpoints, + blocks: Record +): string | null { + const sourceBlock = blocks[edge.source] + const targetBlock = blocks[edge.target] + + if (!sourceBlock || !targetBlock) { + return 'edge references a missing block' + } + + const sourceParent = getWorkflowBlockParentId(sourceBlock) ?? null + const targetParent = getWorkflowBlockParentId(targetBlock) ?? null + + if (sourceParent === targetParent) { + return null + } + + if (targetParent === edge.source && WORKFLOW_CONTAINER_BLOCK_TYPES.has(sourceBlock.type ?? '')) { + return null + } + + if (sourceParent === edge.target && WORKFLOW_CONTAINER_BLOCK_TYPES.has(targetBlock.type ?? '')) { + return null + } + + return `blocks are in different scopes (${sourceParent ?? 'root'} -> ${targetParent ?? 'root'})` +} + +/** True when a block's type is the annotation-only "note" block, which cannot participate in edges. */ +export function isWorkflowAnnotationOnlyBlockType(blockType: string | undefined): boolean { + return blockType === WORKFLOW_ANNOTATION_ONLY_BLOCK_TYPE +} + +export interface WorkflowTriggerCapableBlock { + type?: string + triggerMode?: boolean +} + +/** + * Portable subset of trigger-block detection: an explicit `triggerMode` + * toggle, or the legacy starter block type. Does NOT cover blocks whose + * trigger status comes from the block registry's `category: 'triggers'` + * field (most modern trigger blocks) — that classification only exists in + * the client's block registry (apps/sim/blocks), which apps/realtime is + * architecturally forbidden from importing (see + * scripts/check-monorepo-boundaries.ts). Persisting a redundant "isTrigger" + * flag on the block row to cover that case would itself be a driftable + * duplicate, so that case remains client-enforced only; see + * TriggerUtils.isTriggerBlock in apps/sim/lib/workflows/triggers/triggers.ts + * for the full (client-only) check. + */ +export function isKnownWorkflowTriggerBlock(block: WorkflowTriggerCapableBlock): boolean { + return block.triggerMode === true || block.type === LEGACY_STARTER_BLOCK_TYPE +} + +/** + * Names reserved for reference-path prefixes (``, ``, + * ``) — a block cannot take one of these as its (normalized) name. + */ +export const RESERVED_WORKFLOW_BLOCK_NAMES = ['loop', 'parallel', 'variable'] as const + +/** + * Normalizes a block name into its reference-safe form: lowercased, whitespace + * stripped, dots stripped. Dots are stripped because `.` is the reference path + * delimiter — a name like "Trigger.dev 1" must normalize to "triggerdev1" so + * the reference `` parses unambiguously. + */ +export function normalizeWorkflowBlockName(name: string): string { + return name.toLowerCase().replace(/\s+/g, '').replace(/\./g, '') +} + +export interface WorkflowBlockNameConflict { + reason: 'empty' | 'reserved' | 'duplicate' + conflictingBlockId?: string +} + +/** + * Checks whether renaming `blockId` to `name` would produce an empty name, a + * reserved name, or collide with another block's normalized name. Block names + * are used as `` reference identifiers, so two blocks landing + * on the same normalized name makes reference resolution ambiguous — shared + * between the client store and the realtime persistence layer so both agree + * on the same conflicts. + */ +export function getWorkflowBlockNameConflict( + blockId: string, + name: string, + siblingNamesById: Record +): WorkflowBlockNameConflict | null { + const normalized = normalizeWorkflowBlockName(name) + if (!normalized) return { reason: 'empty' } + + const conflict = Object.entries(siblingNamesById).find( + ([id, siblingName]) => id !== blockId && normalizeWorkflowBlockName(siblingName) === normalized + ) + if (conflict) return { reason: 'duplicate', conflictingBlockId: conflict[0] } + + if ((RESERVED_WORKFLOW_BLOCK_NAMES as readonly string[]).includes(normalized)) { + return { reason: 'reserved' } + } + + return null +} + export interface SubBlockState { id: string type: SubBlockType