fix(workflow): canonicalize realtime edge handles

Author: andresdjasso

apps/realtime/src/database/operations.ts

@@ -32,6 +32,8 @@ import {
   isKnownWorkflowTriggerBlock,
   isWorkflowAnnotationOnlyBlockType,
   isWorkflowBlockProtected,
+  normalizeWorkflowEdgeSourceHandle,
+  normalizeWorkflowEdgeTargetHandle,
 } from '@sim/workflow-types/workflow'
 import { and, eq, inArray, isNull, or, sql } from 'drizzle-orm'
 import { drizzle } from 'drizzle-orm/postgres-js'
@@ -57,13 +59,21 @@ function toEdgeHandles(edge: PersistedEdgeRecord) {
 }
 
 interface EdgeAddCandidate {
-  id?: string
+  id: string
   source: string
   target: string
   sourceHandle?: string | null
   targetHandle?: string | null
 }
 
+function canonicalizeEdgeAddCandidate(edge: EdgeAddCandidate): EdgeAddCandidate {
+  return {
+    ...edge,
+    sourceHandle: normalizeWorkflowEdgeSourceHandle(edge.sourceHandle),
+    targetHandle: normalizeWorkflowEdgeTargetHandle(edge.targetHandle),
+  }
+}
+
 interface FilterEdgesForPersistResult<T> {
   safeEdges: T[]
   droppedCounts: Record<string, number>
@@ -283,8 +293,8 @@ async function insertAutoConnectEdge(
     workflowId,
     sourceBlockId: autoConnectEdge.source,
     targetBlockId: autoConnectEdge.target,
-    sourceHandle: autoConnectEdge.sourceHandle || null,
-    targetHandle: autoConnectEdge.targetHandle || null,
+    sourceHandle: normalizeWorkflowEdgeSourceHandle(autoConnectEdge.sourceHandle),
+    targetHandle: normalizeWorkflowEdgeTargetHandle(autoConnectEdge.targetHandle),
   })
   logger.debug(
     `Added auto-connect edge ${autoConnectEdge.id}: ${autoConnectEdge.source} -> ${autoConnectEdge.target}`
@@ -1051,8 +1061,8 @@ async function handleBlocksOperationTx(
         // 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) => ({
+        const candidates: EdgeAddCandidate[] = (edges as Array<Record<string, unknown>>).map((e) =>
+          canonicalizeEdgeAddCandidate({
             id: e.id as string,
             source: e.source as string,
             target: e.target as string,
@@ -1078,8 +1088,8 @@ async function handleBlocksOperationTx(
             workflowId,
             sourceBlockId: edge.source,
             targetBlockId: edge.target,
-            sourceHandle: edge.sourceHandle || null,
-            targetHandle: edge.targetHandle || null,
+            sourceHandle: normalizeWorkflowEdgeSourceHandle(edge.sourceHandle),
+            targetHandle: normalizeWorkflowEdgeTargetHandle(edge.targetHandle),
           }))
 
           await tx
@@ -1536,18 +1546,17 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str
         throw new Error('Missing required fields for add edge operation')
       }
 
+      const candidate = canonicalizeEdgeAddCandidate({
+        id: payload.id,
+        source: payload.source,
+        target: payload.target,
+        sourceHandle: payload.sourceHandle ?? null,
+        targetHandle: payload.targetHandle ?? null,
+      })
       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,
-          },
-        ]
+        [candidate]
       )
 
       if (safeEdges.length === 0) {
@@ -1561,13 +1570,14 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str
         break
       }
 
+      const [safeEdge] = safeEdges
       await tx.insert(workflowEdges).values({
-        id: payload.id,
+        id: safeEdge.id,
         workflowId,
-        sourceBlockId: payload.source,
-        targetBlockId: payload.target,
-        sourceHandle: payload.sourceHandle || null,
-        targetHandle: payload.targetHandle || null,
+        sourceBlockId: safeEdge.source,
+        targetBlockId: safeEdge.target,
+        sourceHandle: normalizeWorkflowEdgeSourceHandle(safeEdge.sourceHandle),
+        targetHandle: normalizeWorkflowEdgeTargetHandle(safeEdge.targetHandle),
       })
 
       logger.debug(`Added edge ${payload.id}: ${payload.source} -> ${payload.target}`)
@@ -1782,13 +1792,15 @@ async function handleEdgesOperationTx(
 
       logger.info(`Batch adding ${edges.length} edges to workflow ${workflowId}`)
 
-      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 candidates: EdgeAddCandidate[] = (edges as Array<Record<string, unknown>>).map((e) =>
+        canonicalizeEdgeAddCandidate({
+          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)
@@ -1811,8 +1823,8 @@ async function handleEdgesOperationTx(
         workflowId,
         sourceBlockId: edge.source,
         targetBlockId: edge.target,
-        sourceHandle: edge.sourceHandle || null,
-        targetHandle: edge.targetHandle || null,
+        sourceHandle: normalizeWorkflowEdgeSourceHandle(edge.sourceHandle),
+        targetHandle: normalizeWorkflowEdgeTargetHandle(edge.targetHandle),
       }))
 
       await tx
@@ -2179,16 +2191,34 @@ async function handleWorkflowOperationTx(
 
       // Insert all edges from the new state
       if (edges && edges.length > 0) {
-        const edgeValues = edges.map((edge: any) => ({
+        const canonicalEdges = (edges as Array<Record<string, unknown>>).map((edge) =>
+          canonicalizeEdgeAddCandidate({
+            id: edge.id as string,
+            source: edge.source as string,
+            target: edge.target as string,
+            sourceHandle: (edge.sourceHandle as string | null) ?? null,
+            targetHandle: (edge.targetHandle as string | null) ?? null,
+          })
+        )
+        const uniqueEdges = filterUniqueWorkflowEdges(canonicalEdges, [])
+        const edgeValues = uniqueEdges.map((edge) => ({
           id: edge.id,
           workflowId,
           sourceBlockId: edge.source,
           targetBlockId: edge.target,
-          sourceHandle: edge.sourceHandle || null,
-          targetHandle: edge.targetHandle || null,
+          sourceHandle: edge.sourceHandle ?? null,
+          targetHandle: edge.targetHandle ?? null,
         }))
 
-        await tx.insert(workflowEdges).values(edgeValues)
+        if (uniqueEdges.length < edges.length) {
+          logger.info(`Dropped ${edges.length - uniqueEdges.length} duplicate edge(s)`, {
+            operation: WORKFLOW_OPERATIONS.REPLACE_STATE,
+          })
+        }
+
+        if (edgeValues.length > 0) {
+          await tx.insert(workflowEdges).values(edgeValues)
+        }
       }
 
       // Insert all loops from the new state

apps/sim/lib/workflows/persistence/utils.test.ts

@@ -362,13 +362,15 @@ describe('Database Helpers', () => {
         name: 'Start Block',
         position: { x: 100, y: 100 },
         enabled: true,
+        errorEnabled: false,
         horizontalHandles: true,
         height: 150,
         subBlocks: { input: { id: 'input', type: 'short-input' as const, value: 'test' } },
         outputs: { result: { type: 'string' } },
         data: { parentId: undefined, extent: undefined, width: 350 },
         advancedMode: false,
         triggerMode: false,
+        locked: undefined,
       })
 
       expect(result?.edges[0]).toEqual({

apps/sim/stores/workflows/utils.test.ts

@@ -9,7 +9,7 @@ import {
 import type { Edge } from 'reactflow'
 import { describe, expect, it } from 'vitest'
 import { normalizeName } from '@/executor/constants'
-import { getUniqueBlockName, regenerateBlockIds } from './utils'
+import { filterNewEdges, getUniqueBlockName, regenerateBlockIds } from './utils'
 
 describe('normalizeName', () => {
   it.concurrent('should convert to lowercase', () => {
@@ -107,6 +107,33 @@ describe('normalizeName', () => {
   })
 })
 
+describe('filterNewEdges', () => {
+  const makeEdge = (id: string, sourceHandle: string, targetHandle: string): Edge => ({
+    id,
+    source: 'source',
+    target: 'target',
+    sourceHandle,
+    targetHandle,
+  })
+
+  it('treats legacy positioned handles as the same logical connection', () => {
+    const currentEdges = [makeEdge('legacy', 'source-left', 'target-top')]
+    const candidates = [
+      makeEdge('current', 'source-right', 'target-left'),
+      makeEdge('legacy-vertical', 'source-bottom', 'target-bottom'),
+    ]
+
+    expect(filterNewEdges(candidates, currentEdges)).toEqual([])
+  })
+
+  it('keeps semantic routing handles distinct', () => {
+    const currentEdges = [makeEdge('true-route', 'condition-true', 'target-left')]
+    const candidates = [makeEdge('false-route', 'condition-false', 'target-left')]
+
+    expect(filterNewEdges(candidates, currentEdges)).toEqual(candidates)
+  })
+})
+
 describe('getUniqueBlockName', () => {
   it('should return "Start" for starter blocks', () => {
     expect(getUniqueBlockName('Start', {})).toBe('Start')

packages/workflow-types/src/workflow.ts

@@ -309,14 +309,28 @@ export function getHorizontalWorkflowHandleSide(
   return pointerX < cardWidth / 2 ? 'left' : 'right'
 }
 
-// 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
+/**
+ * Returns the canonical persisted source handle used for edge identity.
+ *
+ * Falsy-coalescing mirrors persistence, where an empty handle is stored as
+ * `null`, while positioned legacy outputs collapse onto the right-side port.
+ */
+export function normalizeWorkflowEdgeSourceHandle(
+  handle: string | null | undefined
+): string | null {
+  return normalizePositionedSourceHandleId(handle || null)
+}
+
+/**
+ * Returns the canonical persisted target handle used for edge identity.
+ *
+ * Falsy-coalescing mirrors persistence, where an empty handle is stored as
+ * `null`, while legacy vertical inputs collapse onto the left-side port.
+ */
+export function normalizeWorkflowEdgeTargetHandle(
+  handle: string | null | undefined
+): string | null {
+  return normalizePositionedTargetHandleId(handle || null)
 }
 
 function isDuplicateWorkflowEdge(
@@ -325,11 +339,11 @@ function isDuplicateWorkflowEdge(
 ): boolean {
   return (
     edge.source === existing.source &&
-    normalizeWorkflowEdgeHandle(edge.sourceHandle) ===
-      normalizeWorkflowEdgeHandle(existing.sourceHandle) &&
+    normalizeWorkflowEdgeSourceHandle(edge.sourceHandle) ===
+      normalizeWorkflowEdgeSourceHandle(existing.sourceHandle) &&
     edge.target === existing.target &&
-    normalizeWorkflowEdgeHandle(edge.targetHandle) ===
-      normalizeWorkflowEdgeHandle(existing.targetHandle)
+    normalizeWorkflowEdgeTargetHandle(edge.targetHandle) ===
+      normalizeWorkflowEdgeTargetHandle(existing.targetHandle)
   )
 }