import { useCallback, useEffect, useRef } from 'react'

import { createLogger } from '@sim/logger'

import {

BLOCK_OPERATIONS,

BLOCKS_OPERATIONS,

EDGES_OPERATIONS,

OPERATION_TARGETS,

SUBBLOCK_OPERATIONS,

SUBFLOW_OPERATIONS,

VARIABLE_OPERATIONS,

WORKFLOW_OPERATIONS,

} from '@sim/realtime-protocol/constants'

import { generateId } from '@sim/utils/id'

import { useQueryClient } from '@tanstack/react-query'

import { isEqual } from 'es-toolkit'

import type { Edge } from 'reactflow'

import { useShallow } from 'zustand/react/shallow'

import { requestJson } from '@/lib/api/client/request'

import { getWorkflowStateContract } from '@/lib/api/contracts'

import { useSession } from '@/lib/auth/auth-client'

import {

type WorkflowSearchSubflowFieldId,

workflowSearchSubflowFieldMatchesExpected,

} from '@/lib/workflows/search-replace/subflow-fields'

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 { useNotificationStore } from '@/stores/notifications'

import {

registerEmitFunctions,

useOperationQueue,

useOperationQueueStore,

} from '@/stores/operation-queue/store'

import { usePanelEditorStore } from '@/stores/panel'

import { useCodeUndoRedoStore, useUndoRedoStore } from '@/stores/undo-redo'

import { useVariablesStore } from '@/stores/variables/store'

import { useWorkflowDiffStore } from '@/stores/workflow-diff/store'

import {

applyWorkflowStateToStores,

WORKFLOW_DIFF_SETTLED_EVENT,

} from '@/stores/workflow-diff/utils'

import { useWorkflowRegistry } from '@/stores/workflows/registry/store'

import { useSubBlockStore } from '@/stores/workflows/subblock/store'

import { filterNewEdges, filterValidEdges, mergeSubblockState } from '@/stores/workflows/utils'

import { useWorkflowStore } from '@/stores/workflows/workflow/store'

import type {

BlockState,

Loop,

Parallel,

Position,

WorkflowState,

} from '@/stores/workflows/workflow/types'

import { findAllDescendantNodes, isBlockProtected } from '@/stores/workflows/workflow/utils'

const logger = createLogger('CollaborativeWorkflow')

export function useCollaborativeWorkflow() {

const queryClient = useQueryClient()

const undoRedo = useUndoRedo()

const isUndoRedoInProgress = useRef(false)

const lastDiffOperationId = useRef<string | null>(null)

useEffect(() => {

const moveHandler = (e: any) => {

const { blockId, before, after } = e.detail || {}

if (!blockId || !before || !after) return

if (isUndoRedoInProgress.current) return

undoRedo.recordBatchMoveBlocks([{ blockId, before, after }])

}

const parentUpdateHandler = (e: any) => {

const { blockId, oldParentId, newParentId, oldPosition, newPosition, affectedEdges } =

e.detail || {}

if (!blockId) return

if (isUndoRedoInProgress.current) return

undoRedo.recordUpdateParent(

blockId,

oldParentId,

newParentId,

oldPosition,

newPosition,

affectedEdges

)

}

const diffOperationHandler = (e: any) => {

const {

type,

baselineSnapshot,

proposedState,

diffAnalysis,

beforeAccept,

afterAccept,

beforeReject,

afterReject,

} = e.detail || {}

// Don't record during undo/redo operations

if (isUndoRedoInProgress.current) return

// Generate a unique ID for this diff operation to prevent duplicates

// Use block keys from the relevant states for each operation type

let stateForId

if (type === 'apply-diff') {

stateForId = proposedState

} else if (type === 'accept-diff') {

stateForId = afterAccept

} else if (type === 'reject-diff') {

stateForId = afterReject

}

const blockKeys = stateForId?.blocks ? Object.keys(stateForId.blocks).sort().join(',') : ''

const operationId = `${type}-${blockKeys}`

if (lastDiffOperationId.current === operationId) {

logger.debug('Skipping duplicate diff operation', { type, operationId })

return // Skip duplicate

}

lastDiffOperationId.current = operationId

if (type === 'apply-diff' && baselineSnapshot && proposedState) {

undoRedo.recordApplyDiff(baselineSnapshot, proposedState, diffAnalysis)

} else if (type === 'accept-diff' && beforeAccept && afterAccept) {

undoRedo.recordAcceptDiff(beforeAccept, afterAccept, diffAnalysis, baselineSnapshot)

} else if (type === 'reject-diff' && beforeReject && afterReject) {

undoRedo.recordRejectDiff(beforeReject, afterReject, diffAnalysis, baselineSnapshot)

}

}

window.addEventListener('workflow-record-move', moveHandler)

window.addEventListener('workflow-record-parent-update', parentUpdateHandler)

window.addEventListener('record-diff-operation', diffOperationHandler)

return () => {

window.removeEventListener('workflow-record-move', moveHandler)

window.removeEventListener('workflow-record-parent-update', parentUpdateHandler)

window.removeEventListener('record-diff-operation', diffOperationHandler)

}

}, [undoRedo])

const {

isConnected,

currentWorkflowId,

emitWorkflowOperation,

emitSubblockUpdate,

emitVariableUpdate,

onWorkflowOperation,

onSubblockUpdate,

onVariableUpdate,

onWorkflowDeleted,

onWorkflowReverted,

onWorkflowUpdated,

onWorkflowDeployed,

onOperationConfirmed,

onOperationFailed,

} = useSocket()

const activeWorkflowId = useWorkflowRegistry((state) => state.activeWorkflowId)

const { data: session } = useSession()

const { hasActiveDiff, isShowingDiff } = useWorkflowDiffStore(

useShallow((state) => ({

hasActiveDiff: state.hasActiveDiff,

isShowingDiff: state.isShowingDiff,

}))

)

const isBaselineDiffView = hasActiveDiff && !isShowingDiff

// Track if we're applying remote changes to avoid infinite loops

const isApplyingRemoteChange = useRef(false)

const reloadSequencesRef = useRef<Record<string, number>>({})

const {

addToQueue,

confirmOperation,

failOperation,

cancelOperationsForBlock,

cancelOperationsForVariable,

} = useOperationQueue()

// Register emit functions with operation queue store

useEffect(() => {

const registeredWorkflowId =

isConnected && currentWorkflowId === activeWorkflowId ? currentWorkflowId : null

registerEmitFunctions(

emitWorkflowOperation,

emitSubblockUpdate,

emitVariableUpdate,

registeredWorkflowId

)

}, [

activeWorkflowId,

currentWorkflowId,

emitWorkflowOperation,

emitSubblockUpdate,

emitVariableUpdate,

isConnected,

])

useEffect(() => {

const handleWorkflowOperation = (data: any) => {

const { operation, target, payload, userId, metadata } = data

if (isApplyingRemoteChange.current) return

// Filter broadcasts by workflowId to prevent cross-workflow updates

if (metadata?.workflowId && metadata.workflowId !== activeWorkflowId) {

logger.debug('Ignoring workflow operation for different workflow', {

broadcastWorkflowId: metadata.workflowId,

activeWorkflowId,

})

return

}

logger.info(`Received ${operation} on ${target} from user ${userId}`)

// Apply the operation to local state

isApplyingRemoteChange.current = true

try {

if (target === OPERATION_TARGETS.BLOCK) {

switch (operation) {

case BLOCK_OPERATIONS.UPDATE_NAME:

useWorkflowStore.getState().updateBlockName(payload.id, payload.name)

break

case BLOCK_OPERATIONS.UPDATE_ADVANCED_MODE:

useWorkflowStore.getState().setBlockAdvancedMode(payload.id, payload.advancedMode)

break

case BLOCK_OPERATIONS.UPDATE_CANONICAL_MODE:

useWorkflowStore

.getState()

.setBlockCanonicalMode(payload.id, payload.canonicalId, payload.canonicalMode)

break

}

} else if (target === OPERATION_TARGETS.BLOCKS) {

switch (operation) {

case BLOCKS_OPERATIONS.BATCH_UPDATE_POSITIONS: {

const { updates } = payload

if (Array.isArray(updates)) {

useWorkflowStore.getState().batchUpdatePositions(updates)

}

break

}

}

} else if (target === OPERATION_TARGETS.SUBBLOCK) {

switch (operation) {

case SUBBLOCK_OPERATIONS.BATCH_UPDATE: {

const { updates } = payload

if (Array.isArray(updates)) {

updates.forEach(

(update: { blockId: string; subblockId: string; value: unknown }) => {

useSubBlockStore

.getState()

.setValue(update.blockId, update.subblockId, update.value)

useWorkflowStore

.getState()

.syncDynamicHandleSubblockValue(

update.blockId,

update.subblockId,

update.value

)

}

)

}

break

}

}

} else if (target === OPERATION_TARGETS.EDGES) {

switch (operation) {

case EDGES_OPERATIONS.BATCH_REMOVE_EDGES: {

const { ids } = payload

if (Array.isArray(ids) && ids.length > 0) {

useWorkflowStore.getState().batchRemoveEdges(ids)

const updatedBlocks = useWorkflowStore.getState().blocks

const updatedEdges = useWorkflowStore.getState().edges

const graph = {

blocksById: updatedBlocks,

edgesById: Object.fromEntries(updatedEdges.map((e) => [e.id, e])),

}

const undoRedoStore = useUndoRedoStore.getState()

const stackKeys = Object.keys(undoRedoStore.stacks)

stackKeys.forEach((key) => {

const [wfId, uId] = key.split(':')

if (wfId === activeWorkflowId) {

undoRedoStore.pruneInvalidEntries(wfId, uId, graph)

}

})

}

break

}

case EDGES_OPERATIONS.BATCH_ADD_EDGES: {

const { edges } = payload

if (Array.isArray(edges) && edges.length > 0) {

const blocks = useWorkflowStore.getState().blocks

const currentEdges = useWorkflowStore.getState().edges

const validEdges = filterValidEdges(edges, blocks)

const newEdges = filterNewEdges(validEdges, currentEdges)

if (newEdges.length > 0) {

useWorkflowStore.getState().batchAddEdges(newEdges, { skipValidation: true })

}

}

break

}

}

} else if (target === OPERATION_TARGETS.SUBFLOW) {

switch (operation) {

case SUBFLOW_OPERATIONS.UPDATE:

// Handle subflow configuration updates (loop/parallel type changes, etc.)

if (payload.type === 'loop') {

const { config } = payload

if (config.loopType !== undefined) {

useWorkflowStore.getState().updateLoopType(payload.id, config.loopType)

}

if (config.iterations !== undefined) {

useWorkflowStore.getState().updateLoopCount(payload.id, config.iterations)

}

if (config.forEachItems !== undefined) {

useWorkflowStore.getState().setLoopForEachItems(payload.id, config.forEachItems)

}

if (config.whileCondition !== undefined) {

useWorkflowStore

.getState()

.setLoopWhileCondition(payload.id, config.whileCondition)

}

if (config.doWhileCondition !== undefined) {

useWorkflowStore

.getState()

.setLoopDoWhileCondition(payload.id, config.doWhileCondition)

}

} else if (payload.type === 'parallel') {

const { config } = payload

if (config.parallelType !== undefined) {

useWorkflowStore.getState().updateParallelType(payload.id, config.parallelType)

}

if (config.count !== undefined) {

useWorkflowStore.getState().updateParallelCount(payload.id, config.count)

}

if (config.batchSize !== undefined) {

useWorkflowStore.getState().updateParallelBatchSize(payload.id, config.batchSize)

}

if (config.distribution !== undefined) {

useWorkflowStore

.getState()

.updateParallelCollection(payload.id, config.distribution)

}

}

break

}

} else if (target === OPERATION_TARGETS.VARIABLE) {

switch (operation) {

case VARIABLE_OPERATIONS.ADD:

useVariablesStore.getState().addVariable(

{

workflowId: payload.workflowId,

name: payload.name,

type: payload.type,

value: payload.value,

},

payload.id

)

break

case VARIABLE_OPERATIONS.UPDATE:

if (payload.field === 'name') {

useVariablesStore

.getState()

.updateVariable(payload.variableId, { name: payload.value })

} else if (payload.field === 'value') {

useVariablesStore

.getState()

.updateVariable(payload.variableId, { value: payload.value })

} else if (payload.field === 'type') {

useVariablesStore

.getState()

.updateVariable(payload.variableId, { type: payload.value })

}

break

case VARIABLE_OPERATIONS.REMOVE:

useVariablesStore.getState().deleteVariable(payload.variableId)

break

}

} else if (target === OPERATION_TARGETS.WORKFLOW) {

switch (operation) {

case WORKFLOW_OPERATIONS.REPLACE_STATE:

if (payload.state) {

logger.info('Received workflow state replacement from remote user', {

userId,

blockCount: Object.keys(payload.state.blocks || {}).length,

edgeCount: (payload.state.edges || []).length,

hasActiveDiff,

isShowingDiff,

})

useWorkflowStore.getState().replaceWorkflowState(payload.state)

// Extract and apply subblock values

const subBlockValues: Record<string, Record<string, any>> = {}

Object.entries(payload.state.blocks || {}).forEach(

([blockId, block]: [string, any]) => {

subBlockValues[blockId] = {}

Object.entries(block.subBlocks || {}).forEach(

([subBlockId, subBlock]: [string, any]) => {

subBlockValues[blockId][subBlockId] = subBlock.value

}

)

}

)

if (activeWorkflowId) {

useSubBlockStore.getState().setWorkflowValues(activeWorkflowId, subBlockValues)

}

logger.info('Successfully applied remote workflow state replacement')

}

break

}

}

if (target === OPERATION_TARGETS.BLOCKS) {

switch (operation) {

case BLOCKS_OPERATIONS.BATCH_ADD_BLOCKS: {

const { blocks, edges, subBlockValues: addedSubBlockValues } = payload

logger.info('Received batch-add-blocks from remote user', {

userId,

blockCount: (blocks || []).length,

edgeCount: (edges || []).length,

})

if (blocks && blocks.length > 0) {

useWorkflowStore

.getState()

.batchAddBlocks(blocks, edges || [], addedSubBlockValues || {})

}

logger.info('Successfully applied batch-add-blocks from remote user')

break

}

case BLOCKS_OPERATIONS.BATCH_REMOVE_BLOCKS: {

const { ids } = payload

logger.info('Received batch-remove-blocks from remote user', {

userId,

count: (ids || []).length,

})

if (ids && ids.length > 0) {

useWorkflowStore.getState().batchRemoveBlocks(ids)

}

logger.info('Successfully applied batch-remove-blocks from remote user')

break

}

case BLOCKS_OPERATIONS.BATCH_TOGGLE_ENABLED: {

const { blockIds } = payload

logger.info('Received batch-toggle-enabled from remote user', {

userId,

count: (blockIds || []).length,

})

if (blockIds && blockIds.length > 0) {

useWorkflowStore.getState().batchToggleEnabled(blockIds)

}

logger.info('Successfully applied batch-toggle-enabled from remote user')

break

}

case BLOCKS_OPERATIONS.BATCH_TOGGLE_HANDLES: {

const { blockIds } = payload

logger.info('Received batch-toggle-handles from remote user', {

userId,

count: (blockIds || []).length,

})

if (blockIds && blockIds.length > 0) {

useWorkflowStore.getState().batchToggleHandles(blockIds)

}

logger.info('Successfully applied batch-toggle-handles from remote user')

break

}

case BLOCKS_OPERATIONS.BATCH_TOGGLE_LOCKED: {

const { blockIds } = payload

logger.info('Received batch-toggle-locked from remote user', {

userId,

count: (blockIds || []).length,

})

if (blockIds && blockIds.length > 0) {

useWorkflowStore.getState().batchToggleLocked(blockIds)

}

logger.info('Successfully applied batch-toggle-locked from remote user')

break

}

case BLOCKS_OPERATIONS.BATCH_UPDATE_PARENT: {

const { updates } = payload

logger.info('Received batch-update-parent from remote user', {

userId,

count: (updates || []).length,

})

if (updates && updates.length > 0) {

useWorkflowStore.getState().batchUpdateBlocksWithParent(

updates.map(

(u: { id: string; parentId: string; position: { x: number; y: number } }) => ({

id: u.id,

position: u.position,

parentId: u.parentId || undefined,

})

)

)

}

logger.info('Successfully applied batch-update-parent from remote user')

break

}

}

}

} catch (error) {

logger.error('Error applying remote operation:', error)

} finally {

isApplyingRemoteChange.current = false

}

}

const handleSubblockUpdate = (data: any) => {

const { workflowId, blockId, subblockId, value, userId } = data

if (isApplyingRemoteChange.current) return

// Filter broadcasts by workflowId to prevent cross-workflow updates

if (workflowId && workflowId !== activeWorkflowId) {

logger.debug('Ignoring subblock update for different workflow', {

broadcastWorkflowId: workflowId,

activeWorkflowId,

})

return

}

logger.info(`Received subblock update from user ${userId}: ${blockId}.${subblockId}`)

isApplyingRemoteChange.current = true

try {

useSubBlockStore.getState().setValue(blockId, subblockId, value)

useWorkflowStore.getState().syncDynamicHandleSubblockValue(blockId, subblockId, value)

const blockType = useWorkflowStore.getState().blocks?.[blockId]?.type

if (activeWorkflowId && blockType === 'function' && subblockId === 'code') {

useCodeUndoRedoStore.getState().clear(activeWorkflowId, blockId, subblockId)

}

} catch (error) {

logger.error('Error applying remote subblock update:', error)

} finally {

isApplyingRemoteChange.current = false

}

}

const handleVariableUpdate = (data: any) => {

const { workflowId, variableId, field, value, userId } = data

if (isApplyingRemoteChange.current) return

// Filter broadcasts by workflowId to prevent cross-workflow updates

if (workflowId && workflowId !== activeWorkflowId) {

logger.debug('Ignoring variable update for different workflow', {

broadcastWorkflowId: workflowId,

activeWorkflowId,

})

return

}

logger.info(`Received variable update from user ${userId}: ${variableId}.${field}`)

isApplyingRemoteChange.current = true

try {

if (field === 'name') {

useVariablesStore.getState().updateVariable(variableId, { name: value })

} else if (field === 'value') {

useVariablesStore.getState().updateVariable(variableId, { value })

} else if (field === 'type') {

useVariablesStore.getState().updateVariable(variableId, { type: value })

}

} catch (error) {

logger.error('Error applying remote variable update:', error)

} finally {

isApplyingRemoteChange.current = false

}

}

const handleWorkflowDeleted = (data: any) => {

const { workflowId } = data

logger.warn(`Workflow ${workflowId} has been deleted`)

if (activeWorkflowId === workflowId) {

logger.info(

`Currently active workflow ${workflowId} was deleted, stopping collaborative operations`

)

const currentUserId = session?.user?.id || 'unknown'

useUndoRedoStore.getState().clear(workflowId, currentUserId)

isApplyingRemoteChange.current = false

}

}

const reloadWorkflowFromApi = async (workflowId: string, reason: string): Promise<boolean> => {

const reloadSequence = (reloadSequencesRef.current[workflowId] ?? 0) + 1

reloadSequencesRef.current[workflowId] = reloadSequence

const isLatestReload = () => reloadSequencesRef.current[workflowId] === reloadSequence

const pendingExternalUpdateAtStart =

useWorkflowDiffStore.getState().pendingExternalUpdates[workflowId] ?? 0

useWorkflowDiffStore.getState().setWorkflowReconciliationInProgress(workflowId, true)

const failLatestReconciliation = (message: string) => {

if (!isLatestReload()) return

const diffStore = useWorkflowDiffStore.getState()

if ((diffStore.pendingExternalUpdates[workflowId] ?? 0) <= pendingExternalUpdateAtStart) {

diffStore.clearExternalUpdatePending(workflowId)

}

diffStore.setWorkflowReconciliationInProgress(workflowId, false)

diffStore.setWorkflowReconciliationError(workflowId, message)

if ((useWorkflowDiffStore.getState().pendingExternalUpdates[workflowId] ?? 0) > 0) {

window.dispatchEvent(

new CustomEvent(WORKFLOW_DIFF_SETTLED_EVENT, { detail: { workflowId } })

)

}

}

// The contract's `state` is `workflowStateSchema` (loose at the wire

// level — `subBlocks.value` is `unknown`, optional flags omitted),

// but downstream consumers (replaceWorkflowState, the undo/redo

// graph) operate on the store's narrower `WorkflowState`. The

// server-of-record persists store-shaped values, so the runtime

// shape is the store type; we narrow once here at the trust

// boundary instead of sprinkling per-field casts.

let workflowState: WorkflowState | null = null

try {

const responseData = await requestJson(getWorkflowStateContract, {

params: { id: workflowId },

})

const wireState = responseData.data?.state

if (wireState) {

// double-cast-allowed: workflowStateSchema is structurally a supertype of the store's WorkflowState (subBlocks.value is `unknown`, optional booleans, etc.); the server persists store-shaped values so the runtime shape matches

workflowState = wireState as unknown as WorkflowState

if (Object.hasOwn(responseData.data, 'variables')) {

workflowState.variables = responseData.data.variables || {}

}

}

} catch (error) {

logger.error(`Failed to fetch workflow data after ${reason}`, { error })

failLatestReconciliation(

'Failed to sync the latest workflow changes. Refresh and try again.'

)

return false

}

if (!isLatestReload()) {

logger.debug(`Ignoring stale workflow reload after ${reason}`, { workflowId })

return false

}

if (!workflowState) {

logger.error(`No state found in workflow data after ${reason}`, { workflowId })

failLatestReconciliation('No workflow state was returned while syncing latest changes.')

return false

}

if (useWorkflowRegistry.getState().activeWorkflowId !== workflowId) {

logger.debug(`Ignoring workflow reload after active workflow changed`, { workflowId })

if (isLatestReload()) {

useWorkflowDiffStore.getState().setWorkflowReconciliationInProgress(workflowId, false)

}

return false

}

const diffStateBeforeApply = useWorkflowDiffStore.getState()

const pendingExternalUpdateBeforeApply =

diffStateBeforeApply.pendingExternalUpdates[workflowId] ?? 0

if (

diffStateBeforeApply.hasActiveDiff ||

pendingExternalUpdateBeforeApply > pendingExternalUpdateAtStart ||

useOperationQueueStore.getState().hasPendingOperations(workflowId)

) {

logger.info(`Deferring workflow reload apply after ${reason}`, { workflowId })

useWorkflowDiffStore.getState().markExternalUpdatePending(workflowId)

if (isLatestReload()) {

useWorkflowDiffStore.getState().setWorkflowReconciliationInProgress(workflowId, false)

if (useWorkflowRegistry.getState().activeWorkflowId === workflowId) {

void replayPendingExternalUpdate(

workflowId,

'deferred external update after reload apply was skipped'

)

}

}

return false

}

isApplyingRemoteChange.current = true

try {

const stateToApply: WorkflowState = {

blocks: workflowState.blocks || {},

edges: workflowState.edges || [],

loops: workflowState.loops || {},

parallels: workflowState.parallels || {},

lastSaved: workflowState.lastSaved || Date.now(),

}

if (Object.hasOwn(workflowState, 'variables')) {

stateToApply.variables = workflowState.variables || {}

}

applyWorkflowStateToStores(workflowId, stateToApply)

const graph = {

blocksById: workflowState.blocks || {},

edgesById: Object.fromEntries((workflowState.edges || []).map((e) => [e.id, e])),

}

const undoRedoStore = useUndoRedoStore.getState()

const stackKeys = Object.keys(undoRedoStore.stacks)

stackKeys.forEach((key) => {

const [wfId, userId] = key.split(':')

if (wfId === workflowId) {

undoRedoStore.pruneInvalidEntries(wfId, userId, graph)

}

})

logger.info(`Successfully reloaded workflow state after ${reason}`, { workflowId })

const diffStore = useWorkflowDiffStore.getState()

const pendingExternalUpdate = diffStore.pendingExternalUpdates[workflowId] ?? 0

if (pendingExternalUpdate <= pendingExternalUpdateAtStart) {

diffStore.clearExternalUpdatePending(workflowId)

}

diffStore.setWorkflowReconciliationError(workflowId, null)

return true

} finally {

isApplyingRemoteChange.current = false

if (isLatestReload()) {

useWorkflowDiffStore.getState().setWorkflowReconciliationInProgress(workflowId, false)

if (useWorkflowRegistry.getState().activeWorkflowId === workflowId) {

void replayPendingExternalUpdate(

workflowId,

'deferred external update after reconciliation'

)

}

}

}

}

const replayPendingExternalUpdate = async (workflowId: string, reason: string) => {

const diffStore = useWorkflowDiffStore.getState()

if (

useWorkflowRegistry.getState().activeWorkflowId !== workflowId ||

diffStore.hasActiveDiff ||

diffStore.reconcilingWorkflows[workflowId] ||

!diffStore.pendingExternalUpdates[workflowId]

) {

return

}

const queueStore = useOperationQueueStore.getState()

if (queueStore.hasPendingOperations(workflowId)) {

return

}

try {

await reloadWorkflowFromApi(workflowId, reason)

} catch (error) {

logger.error(`Error reloading workflow state after ${reason}:`, error)

}

}

const handleWorkflowReverted = async (data: any) => {

const { workflowId } = data

logger.info(`Workflow ${workflowId} has been reverted to deployed state`)

if (activeWorkflowId !== workflowId) return

useWorkflowDiffStore.getState().markRemoteUpdateSeen(workflowId)

try {

await reloadWorkflowFromApi(workflowId, 'revert')

} catch (error) {

logger.error('Error reloading workflow state after revert:', error)

}

}

const handleWorkflowUpdated = async (data: any) => {

const { workflowId } = data

logger.info(`Workflow ${workflowId} has been updated externally`)

if (activeWorkflowId !== workflowId) return

const diffStore = useWorkflowDiffStore.getState()

const { hasActiveDiff } = diffStore

if (hasActiveDiff) {

logger.info('Deferring workflow-updated: active diff in progress', { workflowId })

diffStore.markExternalUpdatePending(workflowId)

return

}

if (diffStore.reconcilingWorkflows[workflowId]) {

logger.info('Deferring workflow-updated: workflow reconciliation is in progress', {

workflowId,

})

diffStore.markExternalUpdatePending(workflowId)

return

}

const operationQueue = useOperationQueueStore.getState()

if (operationQueue.hasPendingOperations(workflowId)) {

logger.info('Deferring workflow-updated: local operations are still pending', {

workflowId,

})

diffStore.markExternalUpdatePending(workflowId)

void operationQueue.waitForWorkflowOperations(workflowId).then((ready) => {

if (!ready) {

const latestQueue = useOperationQueueStore.getState()

if (latestQueue.hasPendingOperations(workflowId) && !latestQueue.hasOperationError) {

return

}

const diffStore = useWorkflowDiffStore.getState()

diffStore.clearExternalUpdatePending(workflowId)

diffStore.setWorkflowReconciliationError(

workflowId,

'Failed to save local workflow changes before syncing external updates.'

)

return

}

void replayPendingExternalUpdate(workflowId, 'deferred external update after local save')

})

return

}

diffStore.markRemoteUpdateSeen(workflowId)

try {

await reloadWorkflowFromApi(workflowId, 'external update')

} catch (error) {

logger.error('Error reloading workflow state after external update:', error)

}

}

const handleDiffSettled = async (event: Event) => {

const customEvent = event as CustomEvent<{ workflowId?: string }>

const workflowId = customEvent.detail?.workflowId

if (!workflowId || activeWorkflowId !== workflowId) return

const diffStore = useWorkflowDiffStore.getState()

if (!diffStore.pendingExternalUpdates[workflowId]) return

await replayPendingExternalUpdate(workflowId, 'deferred external update')

}

const handleWorkflowDeployed = (data: any) => {

const { workflowId } = data

logger.info(`Workflow ${workflowId} deployment state changed`)

if (workflowId !== activeWorkflowId) return

invalidateDeploymentQueries(queryClient, workflowId)

}

const handleOperationConfirmed = (data: any) => {

const { operationId } = data

logger.debug('Operation confirmed', { operationId })

confirmOperation(operationId)

if (activeWorkflowId) {

void replayPendingExternalUpdate(

activeWorkflowId,

'deferred external update after operation confirm'

)

}

}

const handleOperationFailed = (data: any) => {

const { operationId, error, retryable } = data

logger.warn('Operation failed', { operationId, error, retryable })

failOperation(operationId, retryable)

}

onWorkflowOperation(handleWorkflowOperation)

onSubblockUpdate(handleSubblockUpdate)

onVariableUpdate(handleVariableUpdate)

onWorkflowDeleted(handleWorkflowDeleted)

onWorkflowReverted(handleWorkflowReverted)

onWorkflowUpdated(handleWorkflowUpdated)

onWorkflowDeployed(handleWorkflowDeployed)

onOperationConfirmed(handleOperationConfirmed)

onOperationFailed(handleOperationFailed)

window.addEventListener(WORKFLOW_DIFF_SETTLED_EVENT, handleDiffSettled)

if (activeWorkflowId) {

void replayPendingExternalUpdate(

activeWorkflowId,

'pending external update after workflow activation'

)

}

return () => {

window.removeEventListener(WORKFLOW_DIFF_SETTLED_EVENT, handleDiffSettled)

}

}, [

onWorkflowOperation,

onSubblockUpdate,

onVariableUpdate,

onWorkflowDeleted,

onWorkflowReverted,

onWorkflowUpdated,

onWorkflowDeployed,

onOperationConfirmed,

onOperationFailed,

activeWorkflowId,

queryClient,

confirmOperation,

failOperation,

emitWorkflowOperation,

])

const executeQueuedOperation = useCallback(

(operation: string, target: string, payload: any, localAction: () => void) => {

if (isApplyingRemoteChange.current) {

return

}

// Skip socket operations when viewing baseline diff (readonly)

if (isBaselineDiffView) {

logger.debug('Skipping socket operation while viewing baseline diff:', operation)

return

}

// Queue operations if we have an active workflow - queue handles socket readiness

if (!activeWorkflowId) {

logger.debug('Skipping operation - no active workflow', { operation, target })

return

}

const operationId = generateId()

addToQueue({

id: operationId,

operation: {

operation,

target,

payload,

},

workflowId: activeWorkflowId,

userId: session?.user?.id || 'unknown',

})

localAction()

},

[addToQueue, session?.user?.id, isBaselineDiffView, activeWorkflowId]

)

const collaborativeBatchUpdatePositions = useCallback(

(

updates: Array<{ id: string; position: Position }>,

options?: {

previousPositions?: Map<string, { x: number; y: number; parentId?: string }>

}

) => {

if (isBaselineDiffView) {

return

}

if (!activeWorkflowId) {

logger.debug('Skipping batch position update - no active workflow')

return

}

if (updates.length === 0) return

const operationId = generateId()

addToQueue({

id: operationId,

operation: {

operation: BLOCKS_OPERATIONS.BATCH_UPDATE_POSITIONS,

target: OPERATION_TARGETS.BLOCKS,

payload: { updates },

},

workflowId: activeWorkflowId || '',

userId: session?.user?.id || 'unknown',

})

useWorkflowStore.getState().batchUpdatePositions(updates)

if (options?.previousPositions && options.previousPositions.size > 0) {

const moves = updates

.filter((u) => options.previousPositions!.has(u.id))

.map((u) => {

const prev = options.previousPositions!.get(u.id)!

const block = useWorkflowStore.getState().blocks[u.id]

return {

blockId: u.id,

before: prev,

after: {

x: u.position.x,

y: u.position.y,

parentId: block?.data?.parentId,

},

}

})

.filter((m) => m.before.x !== m.after.x || m.before.y !== m.after.y)

if (moves.length > 0) {