opensource/sim
3
0
Fork 0
mirror of https://github.com/simstudioai/sim.git synced 2026-10-25 09:34:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflow-edges): enforce edge/block validation server-side, not just client-side (#5571)

Browse Source 
* fix(workflow-edges): enforce edge/block validation server-side, not just client-side

Dragging a connection that creates a cycle correctly refused to render
client-side, but the cyclic edge was still queued for realtime persistence
and written to the DB, so it reappeared after refresh. Root cause: cycle
detection (and several other edge/block rules) only lived in the client
Zustand store and was never enforced by the realtime persistence layer,
which is the actual source of truth on reload.

- Move wouldCreateCycle, edge scope-boundary, annotation-only-block,
  duplicate-edge, and block-name-conflict checks into @sim/workflow-types
  so the client store, the collaborative queueing layer, and
  apps/realtime's DB write path all share one implementation
- Wire these into apps/realtime/database/operations.ts's edge-add and
  block-rename handlers as the authoritative gate
- Client-side behavior is unchanged (same call sites, same error messages,
  same rule ordering) — verified via existing + new test coverage

* fix(workflow-edges): address review findings on realtime edge validation

- Select triggerMode when fetching blocks for edge-add validation —
  isKnownWorkflowTriggerBlock checked block.triggerMode but the column
  was never fetched from the DB, so trigger-mode blocks could still
  receive an incoming edge (Cursor Bugbot)
- Make filterUniqueWorkflowEdges incremental, so two duplicate edges
  within the same BATCH_ADD_EDGES payload are also deduped instead of
  both surviving (Greptile)

* fix(workflow-edges): normalize empty-string handles in duplicate-edge check

filterUniqueWorkflowEdges compared handles with ??, so a `sourceHandle: ''`
edge wasn't recognized as a duplicate of an existing null-handle edge —
even though both get persisted as the same null value at insert time
(edge.sourceHandle || null). Falsy-coalesce in the comparison so '' and
null/undefined are treated as the same "no handle" state everywhere.
(Greptile)

* improvement(workflow-edges): dedup realtime edge-add validation, reuse block-name-conflict helper

/simplify pass on the already-merged-quality PR before sign-off:

- Extract filterEdgesForPersist in apps/realtime/src/database/operations.ts:
  the single-edge ADD and batch BATCH_ADD_EDGES handlers hand-inlined the
  same six-step validation pipeline (missing block, protected target,
  annotation-only, trigger-target, scope boundary, duplicate, cycle) and
  each independently re-fetched blocksById/existingEdgesForCycleCheck. Two
  copies of one rule in the same file was exactly the drift risk this PR
  otherwise closes across client/server. One shared helper now backs both.
  Net -63 lines despite the new shared function.
- Fix a real bug this surfaced: droppedCounts was keyed by the free-text,
  block-id-bearing scope-boundary message, so it could never aggregate
  across edges/runs. Now keyed by a stable 'scope boundary' reason.
- use-collaborative-workflow.ts's collaborativeUpdateBlockName still
  hand-rolled the empty/reserved/duplicate block-name-conflict checks this
  PR centralized as getWorkflowBlockNameConflict (already adopted by
  store.ts). Switched it to the shared helper, which also fixes a latent
  check-order mismatch between the two (this pre-check ran reserved before
  duplicate; the store's real gate — after this PR's own store.ts change —
  runs duplicate before reserved, so they could disagree on which toast a
  name that was both reserved and duplicate would surface).

* docs(workflow-edges): document the per-workflow write-serialization invariant

No behavior change. Address Greptile P1 on filterEdgesForPersist ('concurrent
duplicate writes can persist without a per-workflow write guard') by
documenting, at the actual mechanism, why the concern doesn't apply here:
persistWorkflowOperation's leading 'UPDATE workflow SET updatedAt ... WHERE
id = workflowId' already takes a row lock that serializes every operation
(including edge adds) for a given workflowId — a second concurrent call
blocks on that UPDATE until the first transaction commits or rolls back, so
the validate-then-insert sequence in filterEdgesForPersist can never
interleave across two writers on the same workflow.

Verified empirically, not just by reading: ran two concurrent transactions
against a throwaway local Postgres against the exact statement shape used
here (UPDATE the parent row, sleep to simulate the read/validate window,
insert, commit). The second transaction's UPDATE blocked for the full
duration of the first's transaction and only proceeded once the first
committed — confirming the row lock, not any application-level guard,
already provides the serialization Greptile flagged as missing.

Added a comment at the lock site (not a second, redundant advisory lock)
so a future change can't silently break this invariant by making the
UPDATE conditional/skippable as a perceived no-op optimization.

* fix(workflow-edges): validate edges in BATCH_ADD_BLOCKS before persisting

Real gap Cursor's PR summary flagged ('BATCH_ADD_BLOCKS edge inserts... not
fully covered by the new server pipeline'), confirmed by reading the code:
this handler bulk-inserted the edges from a block-paste/duplicate/import
payload directly into workflowEdges with zero validation — no missing-block,
protected-target, annotation-only, trigger-target, scope-boundary,
duplicate, or cycle check. A client sending edges through this operation
instead of BATCH_ADD_EDGES could bypass every rule this PR otherwise
enforces server-side, exactly the class of gap the PR exists to close.

Routes it through the same filterEdgesForPersist used by the other two
edge-add handlers. Runs after the block insert in this same handler, so the
shared helper's blocksById lookup also sees the blocks this same batch just
inserted (a transaction observes its own prior writes).
This commit is contained in:
Waleed
2026-07-10 15:25:25 -07:00
committed by GitHub
parent 43e61a1eaf
commit fac8ec0c51
10 changed files with 810 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/realtime/src/database/operations.ts
+287 -135
View File
@@ -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<T> {
safeEdges: T[]
droppedCounts: Record<string, number>
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<T extends EdgeAddCandidate>(
tx: any,
workflowId: string,
candidates: T[]
): Promise<FilterEdgesForPersistResult<T>> {
const connectedBlockIds = new Set<string>()
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<string, EdgeBlockRecord> = Object.fromEntries(
connectedBlocks.map((b: EdgeBlockRecord) => [b.id, b])
)
const parentIds = new Set<string>()
for (const block of connectedBlocks) {
const parentId = (block.data as Record<string, unknown> | 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<string, number> = {}
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<string, string> = 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<string, unknown>) => ({
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,
}))
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`,
},
// 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<Record<string, unknown>>).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,
})
)
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<string, EdgeBlockRecord> = 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<string>()
for (const block of edgeBlocks) {
const parentId = (block.data as Record<string, unknown> | 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<string>()
edges.forEach((e: Record<string, unknown>) => {
connectedBlockIds.add(e.source as string)
connectedBlockIds.add(e.target as string)
})
const candidates: EdgeAddCandidate[] = (edges as Array<Record<string, unknown>>).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,
}))
// Fetch blocks to check lock status
const connectedBlocks = await tx
.select({
id: workflowBlocks.id,
locked: workflowBlocks.locked,
data: workflowBlocks.data,
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,
})
.from(workflowBlocks)
.where(
and(
eq(workflowBlocks.workflowId, workflowId),
inArray(workflowBlocks.id, Array.from(connectedBlockIds))
)
)
type AddEdgeBlockRecord = (typeof connectedBlocks)[number]
const blocksById: Record<string, AddEdgeBlockRecord> = Object.fromEntries(
connectedBlocks.map((b: AddEdgeBlockRecord) => [b.id, b])
)
// Collect parent IDs that need to be fetched
const parentIds = new Set<string>()
for (const block of connectedBlocks) {
const parentId = (block.data as Record<string, unknown> | 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
}
}
// Filter edges - only add edges where target block is not protected
const safeEdges = (edges as Array<Record<string, unknown>>).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<string, unknown>) => ({
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
apps/sim/executor/constants.ts
+14 -11
View File
@@ -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
* `<triggerdev1.output>` 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)
}
apps/sim/hooks/use-collaborative-workflow.ts
+24 -16
View File
@@ -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
apps/sim/stores/workflows/utils.ts
+2 -10
View File
@@ -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<string, BlockStat
}
export function filterNewEdges(edgesToAdd: Edge[], currentEdges: Edge[]): Edge[] {
return edgesToAdd.filter((edge) => {
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 {
apps/sim/stores/workflows/workflow/edge-validation.test.ts
+99
View File
@@ -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>): 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)
})
})
apps/sim/stores/workflows/workflow/edge-validation.ts
+9 -37
View File
@@ -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, BlockState>): 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<string, BlockState>
@@ -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
apps/sim/stores/workflows/workflow/store.test.ts
+92
View File
@@ -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()
apps/sim/stores/workflows/workflow/store.ts
+19 -19
View File
@@ -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<WorkflowStore>()(
// 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<WorkflowStore>()(
targetHandle: edge.targetHandle,
type: edge.type || 'default',
data: edge.data || {},
})
}
})),
]
set({
blocks: { ...blocks },
@@ -657,26 +657,26 @@ export const useWorkflowStore = create<WorkflowStore>()(
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: [] }
}
apps/sim/stores/workflows/workflow/utils.ts
+13 -41
View File
@@ -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
}
return wouldCreateWorkflowEdgeCycle(edges, sourceId, targetId)
}
const adjacencyList = new Map<string, string[]>()
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<string>()
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 against `currentEdges`, evaluated incrementally within the batch.
*/
export function filterAcyclicEdges(edgesToAdd: Edge[], currentEdges: Edge[]): Edge[] {
return filterAcyclicWorkflowEdges(edgesToAdd, currentEdges)
}
/**
packages/workflow-types/src/workflow.ts
+251
View File
@@ -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<string, string[]>()
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<string>()
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<T extends WorkflowEdgeEndpoints>(
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<T extends WorkflowEdgeHandles>(
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, WorkflowEdgeScopeBlock>
): 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 (`<loop.index>`, `<parallel.currentItem>`,
* `<variable.x>`) — 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 `<triggerdev1.output>` 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 `<blockName.field>` 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<string, string>
): 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