diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index cd5666e558..15a59bd74a 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -362,19 +362,7 @@ export function useCollaborativeWorkflow() { const { operationId, error, retryable } = data logger.warn('Operation failed', { operationId, error, retryable }) - const retryFunction = (operation: any) => { - const { operation: op, target, payload } = operation.operation - - if (op === 'subblock-update' && target === 'subblock') { - // Use subblock-update channel for subblock operations - emitSubblockUpdate(payload.blockId, payload.subblockId, payload.value, operation.id) - } else { - // Use workflow-operation channel for block/edge/subflow operations - emitWorkflowOperation(op, target, payload, operation.id) - } - } - - failOperation(operationId, retryFunction) + failOperation(operationId) } // Register event handlers @@ -478,14 +466,38 @@ export function useCollaborativeWorkflow() { autoConnectEdge, // Include edge data for atomic operation } + // Skip if applying remote changes + if (isApplyingRemoteChange.current) { + workflowStore.addBlock(id, type, name, position, data, parentId, extent) + if (autoConnectEdge) { + workflowStore.addEdge(autoConnectEdge) + } + return + } + + // Generate operation ID for queue tracking + const operationId = crypto.randomUUID() + + // Add to queue for retry mechanism + addToQueue({ + id: operationId, + operation: { + operation: 'add', + target: 'block', + payload: completeBlockData, + }, + workflowId: activeWorkflowId || '', + userId: session?.user?.id || 'unknown', + }) + + // Apply locally first (immediate UI feedback) workflowStore.addBlock(id, type, name, position, data, parentId, extent) if (autoConnectEdge) { workflowStore.addEdge(autoConnectEdge) } - if (!isApplyingRemoteChange.current) { - emitWorkflowOperation('add', 'block', completeBlockData) - } + // Emit to server with operation ID for tracking + emitWorkflowOperation('add', 'block', completeBlockData, operationId) return } diff --git a/apps/sim/stores/operation-queue/store.ts b/apps/sim/stores/operation-queue/store.ts index f6a96c118f..aa1decafc9 100644 --- a/apps/sim/stores/operation-queue/store.ts +++ b/apps/sim/stores/operation-queue/store.ts @@ -13,7 +13,7 @@ export interface QueuedOperation { workflowId: string timestamp: number retryCount: number - status: 'pending' | 'confirmed' | 'failed' + status: 'pending' | 'processing' | 'confirmed' | 'failed' userId: string } @@ -24,9 +24,11 @@ interface OperationQueueState { addToQueue: (operation: Omit) => void confirmOperation: (operationId: string) => void - failOperation: (operationId: string, emitFunction: (operation: QueuedOperation) => void) => void + failOperation: (operationId: string) => void handleOperationTimeout: (operationId: string) => void handleSocketReconnection: () => void + processNextOperation: () => void + triggerOfflineMode: () => void clearError: () => void } @@ -60,8 +62,28 @@ export const useOperationQueueStore = create((set, get) => addToQueue: (operation) => { const state = get() + // Check for duplicate operation ID const existingOp = state.operations.find((op) => op.id === operation.id) if (existingOp) { + logger.debug('Skipping duplicate operation', { operationId: operation.id }) + return + } + + // Check for duplicate operation content (same operation on same target with same payload) + const duplicateContent = state.operations.find( + (op) => + op.operation.operation === operation.operation.operation && + op.operation.target === operation.operation.target && + JSON.stringify(op.operation.payload) === JSON.stringify(operation.operation.payload) && + op.workflowId === operation.workflowId + ) + if (duplicateContent) { + logger.debug('Skipping duplicate operation content', { + operationId: operation.id, + existingOperationId: duplicateContent.id, + operation: operation.operation.operation, + target: operation.operation.target, + }) return } @@ -77,20 +99,12 @@ export const useOperationQueueStore = create((set, get) => operation: queuedOp.operation, }) - const timeoutId = setTimeout(() => { - logger.warn('Operation timeout - no server response after 5 seconds', { - operationId: queuedOp.id, - }) - operationTimeouts.delete(queuedOp.id) - - get().handleOperationTimeout(queuedOp.id) - }, 5000) - - operationTimeouts.set(queuedOp.id, timeoutId) - set((state) => ({ operations: [...state.operations, queuedOp], })) + + // Start processing if not already processing + get().processNextOperation() }, confirmOperation: (operationId) => { @@ -114,10 +128,13 @@ export const useOperationQueueStore = create((set, get) => remainingOps: newOperations.length, }) - set({ operations: newOperations }) + set({ operations: newOperations, isProcessing: false }) + + // Process next operation in queue + get().processNextOperation() }, - failOperation: (operationId: string, emitFunction: (operation: QueuedOperation) => void) => { + failOperation: (operationId: string) => { const state = get() const operation = state.operations.find((op) => op.id === operationId) if (!operation) { @@ -140,42 +157,23 @@ export const useOperationQueueStore = create((set, get) => retryCount: newRetryCount, }) + // Update retry count and mark as pending for retry + set((state) => ({ + operations: state.operations.map((op) => + op.id === operationId + ? { ...op, retryCount: newRetryCount, status: 'pending' as const } + : op + ), + isProcessing: false, // Allow processing to continue + })) + + // Schedule retry const timeout = setTimeout(() => { - if (operation.workflowId !== currentWorkflowId) { - logger.warn('Cancelling retry - workflow changed', { - operationId, - operationWorkflow: operation.workflowId, - currentWorkflow: currentWorkflowId, - }) - retryTimeouts.delete(operationId) - set((state) => ({ - operations: state.operations.filter((op) => op.id !== operationId), - })) - return - } - - emitFunction(operation) retryTimeouts.delete(operationId) - - // Create new operation timeout for this retry attempt - const newTimeoutId = setTimeout(() => { - logger.warn('Retry operation timeout - no server response after 5 seconds', { - operationId, - }) - operationTimeouts.delete(operationId) - get().handleOperationTimeout(operationId) - }, 5000) - - operationTimeouts.set(operationId, newTimeoutId) + get().processNextOperation() }, delay) retryTimeouts.set(operationId, timeout) - - set((state) => ({ - operations: state.operations.map((op) => - op.id === operationId ? { ...op, retryCount: newRetryCount } : op - ), - })) } else { logger.error('Operation failed after max retries, triggering offline mode', { operationId }) get().triggerOfflineMode() @@ -194,21 +192,74 @@ export const useOperationQueueStore = create((set, get) => operationId, }) - const retryFunction = (operation: any) => { - const { operation: op, target, payload } = operation.operation + get().failOperation(operationId) + }, - if (op === 'subblock-update' && target === 'subblock') { - if (emitSubblockUpdate) { - emitSubblockUpdate(payload.blockId, payload.subblockId, payload.value, operation.id) - } - } else { - if (emitWorkflowOperation) { - emitWorkflowOperation(op, target, payload, operation.id) - } + processNextOperation: () => { + const state = get() + + // Don't process if already processing + if (state.isProcessing) { + return + } + + // Find the first pending operation (FIFO - first in, first out) + const nextOperation = state.operations.find((op) => op.status === 'pending') + if (!nextOperation) { + return // No pending operations + } + + // Check workflow context + if (nextOperation.workflowId !== currentWorkflowId) { + logger.warn('Cancelling operation - workflow changed', { + operationId: nextOperation.id, + operationWorkflow: nextOperation.workflowId, + currentWorkflow: currentWorkflowId, + }) + set((state) => ({ + operations: state.operations.filter((op) => op.id !== nextOperation.id), + })) + // Try next operation + get().processNextOperation() + return + } + + // Mark as processing + set((state) => ({ + operations: state.operations.map((op) => + op.id === nextOperation.id ? { ...op, status: 'processing' as const } : op + ), + isProcessing: true, + })) + + logger.debug('Processing operation sequentially', { + operationId: nextOperation.id, + operation: nextOperation.operation, + retryCount: nextOperation.retryCount, + }) + + // Emit the operation + const { operation: op, target, payload } = nextOperation.operation + if (op === 'subblock-update' && target === 'subblock') { + if (emitSubblockUpdate) { + emitSubblockUpdate(payload.blockId, payload.subblockId, payload.value, nextOperation.id) + } + } else { + if (emitWorkflowOperation) { + emitWorkflowOperation(op, target, payload, nextOperation.id) } } - get().failOperation(operationId, retryFunction) + // Create operation timeout + const timeoutId = setTimeout(() => { + logger.warn('Operation timeout - no server response after 5 seconds', { + operationId: nextOperation.id, + }) + operationTimeouts.delete(nextOperation.id) + get().handleOperationTimeout(nextOperation.id) + }, 5000) + + operationTimeouts.set(nextOperation.id, timeoutId) }, handleSocketReconnection: () => { @@ -275,6 +326,7 @@ export function useOperationQueue() { confirmOperation: store.confirmOperation, failOperation: store.failOperation, handleSocketReconnection: store.handleSocketReconnection, + processNextOperation: store.processNextOperation, triggerOfflineMode: store.triggerOfflineMode, clearError: store.clearError, }