improvement(mothership): workflow edits via sockets

Author: icecrasher321

apps/sim/app/workspace/providers/socket-provider.tsx

@@ -90,6 +90,7 @@ interface SocketContextType {
   onSelectionUpdate: (handler: (data: any) => void) => void
   onWorkflowDeleted: (handler: (data: any) => void) => void
   onWorkflowReverted: (handler: (data: any) => void) => void
+  onWorkflowUpdated: (handler: (data: any) => void) => void
   onOperationConfirmed: (handler: (data: any) => void) => void
   onOperationFailed: (handler: (data: any) => void) => void
 }
@@ -118,6 +119,7 @@ const SocketContext = createContext<SocketContextType>({
   onSelectionUpdate: () => {},
   onWorkflowDeleted: () => {},
   onWorkflowReverted: () => {},
+  onWorkflowUpdated: () => {},
   onOperationConfirmed: () => {},
   onOperationFailed: () => {},
 })
@@ -155,6 +157,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
     selectionUpdate?: (data: any) => void
     workflowDeleted?: (data: any) => void
     workflowReverted?: (data: any) => void
+    workflowUpdated?: (data: any) => void
     operationConfirmed?: (data: any) => void
     operationFailed?: (data: any) => void
   }>({})
@@ -382,6 +385,11 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
           eventHandlers.current.workflowReverted?.(data)
         })
 
+        socketInstance.on('workflow-updated', (data) => {
+          logger.info(`Workflow ${data.workflowId} has been updated externally`)
+          eventHandlers.current.workflowUpdated?.(data)
+        })
+
         const rehydrateWorkflowStores = async (workflowId: string, workflowState: any) => {
           const [
             { useOperationQueueStore },
@@ -804,6 +812,10 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
     eventHandlers.current.workflowReverted = handler
   }, [])
 
+  const onWorkflowUpdated = useCallback((handler: (data: any) => void) => {
+    eventHandlers.current.workflowUpdated = handler
+  }, [])
+
   const onOperationConfirmed = useCallback((handler: (data: any) => void) => {
     eventHandlers.current.operationConfirmed = handler
   }, [])
@@ -837,6 +849,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
       onSelectionUpdate,
       onWorkflowDeleted,
       onWorkflowReverted,
+      onWorkflowUpdated,
       onOperationConfirmed,
       onOperationFailed,
     }),
@@ -864,6 +877,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
       onSelectionUpdate,
       onWorkflowDeleted,
       onWorkflowReverted,
+      onWorkflowUpdated,
       onOperationConfirmed,
       onOperationFailed,
     ]

apps/sim/hooks/use-collaborative-workflow.ts

@@ -122,6 +122,7 @@ export function useCollaborativeWorkflow() {
     onVariableUpdate,
     onWorkflowDeleted,
     onWorkflowReverted,
+    onWorkflowUpdated,
     onOperationConfirmed,
     onOperationFailed,
   } = useSocket()
@@ -615,6 +616,67 @@ export function useCollaborativeWorkflow() {
       }
     }
 
+    const handleWorkflowUpdated = async (data: any) => {
+      const { workflowId } = data
+      logger.info(`Workflow ${workflowId} has been updated externally`)
+
+      if (activeWorkflowId !== workflowId) return
+
+      const { hasActiveDiff } = useWorkflowDiffStore.getState()
+      if (hasActiveDiff) {
+        logger.info('Skipping workflow-updated: active diff in progress', { workflowId })
+        return
+      }
+
+      try {
+        const response = await fetch(`/api/workflows/${workflowId}`)
+        if (response.ok) {
+          const responseData = await response.json()
+          const workflowData = responseData.data
+
+          if (workflowData?.state) {
+            isApplyingRemoteChange.current = true
+            try {
+              useWorkflowStore.getState().replaceWorkflowState({
+                blocks: workflowData.state.blocks || {},
+                edges: workflowData.state.edges || [],
+                loops: workflowData.state.loops || {},
+                parallels: workflowData.state.parallels || {},
+                lastSaved: workflowData.state.lastSaved || Date.now(),
+                deploymentStatuses: workflowData.state.deploymentStatuses || {},
+              })
+
+              const subblockValues: Record<string, Record<string, any>> = {}
+              Object.entries(workflowData.state.blocks || {}).forEach(([blockId, block]) => {
+                const blockState = block as any
+                subblockValues[blockId] = {}
+                Object.entries(blockState.subBlocks || {}).forEach(([subblockId, subblock]) => {
+                  subblockValues[blockId][subblockId] = (subblock as any).value
+                })
+              })
+
+              useSubBlockStore.setState((state: any) => ({
+                workflowValues: {
+                  ...state.workflowValues,
+                  [workflowId]: subblockValues,
+                },
+              }))
+
+              logger.info(`Successfully applied externally updated workflow state`, { workflowId })
+            } finally {
+              isApplyingRemoteChange.current = false
+            }
+          }
+        } else {
+          logger.error(
+            `Failed to fetch workflow data after external update: ${response.statusText}`
+          )
+        }
+      } catch (error) {
+        logger.error('Error reloading workflow state after external update:', error)
+      }
+    }
+
     const handleOperationConfirmed = (data: any) => {
       const { operationId } = data
       logger.debug('Operation confirmed', { operationId })
@@ -633,6 +695,7 @@ export function useCollaborativeWorkflow() {
     onVariableUpdate(handleVariableUpdate)
     onWorkflowDeleted(handleWorkflowDeleted)
     onWorkflowReverted(handleWorkflowReverted)
+    onWorkflowUpdated(handleWorkflowUpdated)
     onOperationConfirmed(handleOperationConfirmed)
     onOperationFailed(handleOperationFailed)
   }, [
@@ -641,6 +704,7 @@ export function useCollaborativeWorkflow() {
     onVariableUpdate,
     onWorkflowDeleted,
     onWorkflowReverted,
+    onWorkflowUpdated,
     onOperationConfirmed,
     onOperationFailed,
     activeWorkflowId,

apps/sim/lib/copilot/tools/server/workflow/edit-workflow/index.ts

@@ -7,6 +7,7 @@ import {
   type BaseServerTool,
   type ServerToolContext,
 } from '@/lib/copilot/tools/server/base-tool'
+import { env } from '@/lib/core/config/env'
 import { applyTargetedLayout, getTargetedLayoutImpact } from '@/lib/workflows/autolayout'
 import {
   DEFAULT_HORIZONTAL_SPACING,
@@ -287,6 +288,20 @@ export const editWorkflowServerTool: BaseServerTool<EditWorkflowParams, unknown>
 
     logger.info('Workflow state persisted to database', { workflowId })
 
+    try {
+      const socketUrl = env.SOCKET_SERVER_URL || 'http://localhost:3002'
+      await fetch(`${socketUrl}/api/workflow-updated`, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+          'x-api-key': env.INTERNAL_API_SECRET,
+        },
+        body: JSON.stringify({ workflowId }),
+      })
+    } catch (error) {
+      logger.warn('Failed to notify socket server of workflow update', { workflowId, error })
+    }
+
     const sanitizationWarnings = validation.warnings.length > 0 ? validation.warnings : undefined
 
     return {