fix: sync parent update socket side effects

PlaneInABottle · PlaneInABottle · commit 9504519b4f53 · 2026-03-19T19:42:04.000+03:00
diff --git a/apps/sim/socket/handlers/operations.test.ts b/apps/sim/socket/handlers/operations.test.ts
@@ -3,6 +3,7 @@
  */
 
 import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { ZodError } from 'zod'
 
 const { mockPersistWorkflowOperation, mockCheckRolePermission, mockWorkflowOperationParse } =
   vi.hoisted(() => ({
@@ -26,7 +27,12 @@ vi.mock('@/socket/validation/schemas', () => ({
   },
 }))
 
-import { BLOCKS_OPERATIONS, EDGES_OPERATIONS, OPERATION_TARGETS } from '@/socket/constants'
+import {
+  BLOCK_OPERATIONS,
+  BLOCKS_OPERATIONS,
+  EDGES_OPERATIONS,
+  OPERATION_TARGETS,
+} from '@/socket/constants'
 import { setupOperationsHandlers } from '@/socket/handlers/operations'
 
 describe('setupOperationsHandlers', () => {
@@ -696,6 +702,202 @@ describe('setupOperationsHandlers', () => {
     )
   })
 
+  it('broadcasts single block parent side-effect edges to the whole workflow', async () => {
+    mockPersistWorkflowOperation.mockResolvedValue({
+      removedEdgeIds: ['edge-removed'],
+      addedEdges: [
+        {
+          id: 'edge-added',
+          source: 'loop-1',
+          target: 'block-1',
+          sourceHandle: 'loop-start-source',
+          targetHandle: 'target',
+          type: 'workflowEdge',
+        },
+      ],
+    })
+
+    const socketEmit = vi.fn()
+    const socketRoomEmit = vi.fn()
+    const emitToWorkflow = vi.fn()
+    const socketHandlers = new Map<string, (data: unknown) => Promise<void>>()
+
+    const socket = {
+      id: 'socket-1',
+      on: vi.fn((event: string, handler: (data: unknown) => Promise<void>) => {
+        socketHandlers.set(event, handler)
+      }),
+      emit: socketEmit,
+      to: vi.fn(() => ({
+        emit: socketRoomEmit,
+      })),
+    }
+
+    const roomManager = {
+      io: {} as never,
+      initialize: vi.fn(),
+      isReady: vi.fn(() => true),
+      shutdown: vi.fn(),
+      addUserToRoom: vi.fn(),
+      removeUserFromRoom: vi.fn(),
+      getWorkflowIdForSocket: vi.fn().mockResolvedValue('workflow-1'),
+      getUserSession: vi.fn().mockResolvedValue({ userId: 'user-1', userName: 'Test User' }),
+      getWorkflowUsers: vi.fn().mockResolvedValue([
+        {
+          socketId: 'socket-1',
+          userId: 'user-1',
+          workflowId: 'workflow-1',
+          userName: 'Test User',
+          joinedAt: Date.now(),
+          lastActivity: Date.now(),
+          role: 'admin',
+        },
+      ]),
+      hasWorkflowRoom: vi.fn().mockResolvedValue(true),
+      updateUserActivity: vi.fn(),
+      updateRoomLastModified: vi.fn(),
+      broadcastPresenceUpdate: vi.fn(),
+      emitToWorkflow,
+      getUniqueUserCount: vi.fn(),
+      getTotalActiveConnections: vi.fn(),
+      handleWorkflowDeletion: vi.fn(),
+      handleWorkflowRevert: vi.fn(),
+      handleWorkflowUpdate: vi.fn(),
+    }
+
+    setupOperationsHandlers(socket as never, roomManager)
+
+    const workflowOperationHandler = socketHandlers.get('workflow-operation')
+
+    await workflowOperationHandler?.({
+      operationId: 'op-single-parent',
+      operation: BLOCK_OPERATIONS.UPDATE_PARENT,
+      target: OPERATION_TARGETS.BLOCK,
+      payload: { id: 'block-1', parentId: 'loop-1', position: { x: 10, y: 20 } },
+      timestamp: 123,
+    })
+
+    expect(socketRoomEmit).toHaveBeenCalledWith(
+      'workflow-operation',
+      expect.objectContaining({
+        operation: BLOCK_OPERATIONS.UPDATE_PARENT,
+        target: OPERATION_TARGETS.BLOCK,
+        payload: { id: 'block-1', parentId: 'loop-1', position: { x: 10, y: 20 } },
+      })
+    )
+    expect(emitToWorkflow).toHaveBeenNthCalledWith(
+      1,
+      'workflow-1',
+      'workflow-operation',
+      expect.objectContaining({
+        operation: EDGES_OPERATIONS.BATCH_REMOVE_EDGES,
+        target: OPERATION_TARGETS.EDGES,
+        payload: { ids: ['edge-removed'] },
+      })
+    )
+    expect(emitToWorkflow).toHaveBeenNthCalledWith(
+      2,
+      'workflow-1',
+      'workflow-operation',
+      expect.objectContaining({
+        operation: EDGES_OPERATIONS.BATCH_ADD_EDGES,
+        target: OPERATION_TARGETS.EDGES,
+        payload: {
+          edges: [
+            expect.objectContaining({
+              id: 'edge-added',
+              source: 'loop-1',
+              target: 'block-1',
+            }),
+          ],
+        },
+      })
+    )
+    expect(socketEmit).toHaveBeenCalledWith(
+      'operation-confirmed',
+      expect.objectContaining({
+        operationId: 'op-single-parent',
+        serverTimestamp: expect.any(Number),
+      })
+    )
+  })
+
+  it('includes operationId when a zod error happens after parsing', async () => {
+    const socketEmit = vi.fn()
+    const socketHandlers = new Map<string, (data: unknown) => Promise<void>>()
+
+    const socket = {
+      id: 'socket-1',
+      on: vi.fn((event: string, handler: (data: unknown) => Promise<void>) => {
+        socketHandlers.set(event, handler)
+      }),
+      emit: socketEmit,
+      to: vi.fn(() => ({
+        emit: vi.fn(),
+      })),
+    }
+
+    const roomManager = {
+      io: {} as never,
+      initialize: vi.fn(),
+      isReady: vi.fn(() => true),
+      shutdown: vi.fn(),
+      addUserToRoom: vi.fn(),
+      removeUserFromRoom: vi.fn(),
+      getWorkflowIdForSocket: vi.fn().mockResolvedValue('workflow-1'),
+      getUserSession: vi.fn().mockResolvedValue({ userId: 'user-1', userName: 'Test User' }),
+      getWorkflowUsers: vi.fn().mockResolvedValue([
+        {
+          socketId: 'socket-1',
+          userId: 'user-1',
+          workflowId: 'workflow-1',
+          userName: 'Test User',
+          joinedAt: Date.now(),
+          lastActivity: Date.now(),
+          role: 'admin',
+        },
+      ]),
+      hasWorkflowRoom: vi.fn().mockResolvedValue(true),
+      updateUserActivity: vi.fn().mockRejectedValue(
+        new ZodError([
+          {
+            code: 'custom',
+            path: ['payload'],
+            message: 'Invalid payload',
+          },
+        ])
+      ),
+      updateRoomLastModified: vi.fn(),
+      broadcastPresenceUpdate: vi.fn(),
+      emitToWorkflow: vi.fn(),
+      getUniqueUserCount: vi.fn(),
+      getTotalActiveConnections: vi.fn(),
+      handleWorkflowDeletion: vi.fn(),
+      handleWorkflowRevert: vi.fn(),
+      handleWorkflowUpdate: vi.fn(),
+    }
+
+    setupOperationsHandlers(socket as never, roomManager)
+
+    const workflowOperationHandler = socketHandlers.get('workflow-operation')
+
+    await workflowOperationHandler?.({
+      operationId: 'op-zod-after-parse',
+      operation: BLOCKS_OPERATIONS.BATCH_ADD_BLOCKS,
+      target: OPERATION_TARGETS.BLOCKS,
+      payload: { blocks: [], edges: [], loops: {}, parallels: {}, subBlockValues: {} },
+      timestamp: 123,
+    })
+
+    expect(socketEmit).toHaveBeenCalledWith(
+      'operation-failed',
+      expect.objectContaining({
+        operationId: 'op-zod-after-parse',
+        error: 'Invalid operation format',
+      })
+    )
+  })
+
   it('does not emit edge side-effect syncs when the operation has no handler support', async () => {
     mockPersistWorkflowOperation.mockResolvedValue({
       removedEdgeIds: ['edge-removed'],
diff --git a/apps/sim/socket/handlers/operations.ts b/apps/sim/socket/handlers/operations.ts
@@ -33,6 +33,42 @@ function getPayloadArray(payload: unknown, key: string): unknown[] {
   return Array.isArray(value) ? value : []
 }
 
+function emitParentUpdateSideEffects(
+  roomManager: IRoomManager,
+  workflowId: string,
+  result: Awaited<ReturnType<typeof persistWorkflowOperation>>,
+  operationTimestamp: number,
+  senderId: string,
+  userId: string,
+  userName: string
+) {
+  if (result?.removedEdgeIds && result.removedEdgeIds.length > 0) {
+    roomManager.emitToWorkflow(workflowId, 'workflow-operation', {
+      operation: EDGES_OPERATIONS.BATCH_REMOVE_EDGES,
+      target: OPERATION_TARGETS.EDGES,
+      payload: { ids: result.removedEdgeIds },
+      timestamp: operationTimestamp,
+      senderId,
+      userId,
+      userName,
+      metadata: { workflowId, operationId: crypto.randomUUID() },
+    })
+  }
+
+  if (result?.addedEdges && result.addedEdges.length > 0) {
+    roomManager.emitToWorkflow(workflowId, 'workflow-operation', {
+      operation: EDGES_OPERATIONS.BATCH_ADD_EDGES,
+      target: OPERATION_TARGETS.EDGES,
+      payload: { edges: result.addedEdges },
+      timestamp: operationTimestamp,
+      senderId,
+      userId,
+      userName,
+      metadata: { workflowId, operationId: crypto.randomUUID() },
+    })
+  }
+}
+
 export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
   socket.on('workflow-operation', async (data) => {
     const emitOperationError = (
@@ -553,39 +589,63 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
           metadata: { workflowId, operationId: crypto.randomUUID() },
         })
 
-        // Broadcast edge removals if the parent update cleaned up boundary edges
-        if (result?.removedEdgeIds && result.removedEdgeIds.length > 0) {
-          roomManager.emitToWorkflow(workflowId, 'workflow-operation', {
-            operation: EDGES_OPERATIONS.BATCH_REMOVE_EDGES,
-            target: OPERATION_TARGETS.EDGES,
-            payload: { ids: result.removedEdgeIds },
-            timestamp: operationTimestamp,
-            senderId: socket.id,
-            userId: session.userId,
-            userName: session.userName,
-            metadata: { workflowId, operationId: crypto.randomUUID() },
-          })
-        }
+        emitParentUpdateSideEffects(
+          roomManager,
+          workflowId,
+          result,
+          operationTimestamp,
+          socket.id,
+          session.userId,
+          session.userName
+        )
 
-        // Broadcast auto-connected edges so clients add them to local state
-        if (result?.addedEdges && result.addedEdges.length > 0) {
-          roomManager.emitToWorkflow(workflowId, 'workflow-operation', {
-            operation: EDGES_OPERATIONS.BATCH_ADD_EDGES,
-            target: OPERATION_TARGETS.EDGES,
-            payload: { edges: result.addedEdges },
-            timestamp: operationTimestamp,
-            senderId: socket.id,
-            userId: session.userId,
-            userName: session.userName,
-            metadata: { workflowId, operationId: crypto.randomUUID() },
+        if (operationId) {
+          socket.emit('operation-confirmed', {
+            operationId,
+            serverTimestamp: Date.now(),
+            appliedPayload: broadcastPayload,
           })
         }
 
+        return
+      }
+
+      if (target === OPERATION_TARGETS.BLOCK && operation === BLOCK_OPERATIONS.UPDATE_PARENT) {
+        const result = await persistWorkflowOperation(workflowId, {
+          operation,
+          target,
+          payload,
+          timestamp: operationTimestamp,
+          userId: session.userId,
+        })
+
+        await roomManager.updateRoomLastModified(workflowId)
+
+        socket.to(workflowId).emit('workflow-operation', {
+          operation,
+          target,
+          payload,
+          timestamp: operationTimestamp,
+          senderId: socket.id,
+          userId: session.userId,
+          userName: session.userName,
+          metadata: { workflowId, operationId: crypto.randomUUID() },
+        })
+
+        emitParentUpdateSideEffects(
+          roomManager,
+          workflowId,
+          result,
+          operationTimestamp,
+          socket.id,
+          session.userId,
+          session.userName
+        )
+
         if (operationId) {
           socket.emit('operation-confirmed', {
             operationId,
             serverTimestamp: Date.now(),
-            appliedPayload: broadcastPayload,
           })
         }
 
@@ -784,6 +844,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
       if (error instanceof ZodError) {
         logger.error('Invalid operation format:', error.errors)
         socket.emit('operation-failed', {
+          ...(operationId ? { operationId } : {}),
           error: 'Invalid operation format',
           details: error.errors,
         })
