From 96834b75308ff11b3ccbbf8d45654a1461ed41cc Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Fri, 1 Aug 2025 14:26:19 -0700 Subject: [PATCH 1/6] fix(sockets): add sockets event for tag / env var dropdown selections to be unit op --- .../components/sub-block/components/code.tsx | 7 +- .../sub-block/components/combobox.tsx | 5 +- .../document-tag-entry/document-tag-entry.tsx | 45 ++++++- .../sub-block/components/long-input.tsx | 7 +- .../sub-block/components/short-input.tsx | 6 +- apps/sim/contexts/socket-context.tsx | 31 +++++ apps/sim/hooks/use-tag-selection.ts | 25 ++++ apps/sim/socket-server/handlers/subblocks.ts | 115 ++++++++++++++++++ 8 files changed, 233 insertions(+), 8 deletions(-) create mode 100644 apps/sim/hooks/use-tag-selection.ts diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/code.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/code.tsx index e1b5d823ac6..294b199951a 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/code.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/code.tsx @@ -14,6 +14,7 @@ import { WandPromptBar } from '@/app/workspace/[workspaceId]/w/[workflowId]/comp import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value' import { useWand } from '@/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-wand' import type { GenerationType } from '@/blocks/types' +import { useTagSelection } from '@/hooks/use-tag-selection' import { useSubBlockStore } from '@/stores/workflows/subblock/store' const logger = createLogger('Code') @@ -164,6 +165,8 @@ export function Code({ }, }) + const emitTagSelection = useTagSelection(blockId, subBlockId) + // Use preview value when in preview mode, otherwise use store value or prop value const value = isPreview ? previewValue : propValue !== undefined ? propValue : storeValue @@ -306,7 +309,7 @@ export function Code({ const handleTagSelect = (newValue: string) => { if (!isPreview) { setCode(newValue) - setStoreValue(newValue) + emitTagSelection(newValue) } setShowTags(false) setActiveSourceBlockId(null) @@ -319,7 +322,7 @@ export function Code({ const handleEnvVarSelect = (newValue: string) => { if (!isPreview) { setCode(newValue) - setStoreValue(newValue) + emitTagSelection(newValue) } setShowEnvVars(false) diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/combobox.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/combobox.tsx index d0e957ff44f..7e1f51a2e44 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/combobox.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/combobox.tsx @@ -10,6 +10,7 @@ import { createLogger } from '@/lib/logs/console/logger' import { cn } from '@/lib/utils' import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value' import type { SubBlockConfig } from '@/blocks/types' +import { useTagSelection } from '@/hooks/use-tag-selection' const logger = createLogger('ComboBox') @@ -53,6 +54,8 @@ export function ComboBox({ const [activeSourceBlockId, setActiveSourceBlockId] = useState(null) const [highlightedIndex, setHighlightedIndex] = useState(-1) + const emitTagSelection = useTagSelection(blockId, subBlockId) + const inputRef = useRef(null) const overlayRef = useRef(null) const dropdownRef = useRef(null) @@ -330,7 +333,7 @@ export function ComboBox({ // Environment variable and tag selection handler const handleEnvVarSelect = (newValue: string) => { if (!isPreview) { - setStoreValue(newValue) + emitTagSelection(newValue) } } diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx index c4d05265625..105822e751b 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx @@ -11,6 +11,7 @@ import { cn } from '@/lib/utils' import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value' import type { SubBlockConfig } from '@/blocks/types' import { useKnowledgeBaseTagDefinitions } from '@/hooks/use-knowledge-base-tag-definitions' +import { useTagSelection } from '@/hooks/use-tag-selection' interface DocumentTagRow { id: string @@ -47,6 +48,8 @@ export function DocumentTagEntry({ // Use KB tag definitions hook to get available tags const { tagDefinitions, isLoading } = useKnowledgeBaseTagDefinitions(knowledgeBaseId) + const emitTagSelection = useTagSelection(blockId, subBlock.id) + // State for dropdown visibility - one for each row const [dropdownStates, setDropdownStates] = useState>({}) // State for type dropdown visibility - one for each row @@ -191,6 +194,45 @@ export function DocumentTagEntry({ setStoreValue(jsonString) } + // Special handler for tag dropdown selections that uses immediate emission + const handleTagDropdownSelection = (rowIndex: number, column: string, value: string) => { + if (isPreview || disabled) return + + const updatedRows = [...rows].map((row, idx) => { + if (idx === rowIndex) { + const newCells = { ...row.cells, [column]: value } + + // Auto-select type when existing tag is selected + if (column === 'tagName' && value) { + const tagDef = tagDefinitions.find( + (def) => def.displayName.toLowerCase() === value.toLowerCase() + ) + if (tagDef) { + newCells.type = tagDef.fieldType + } + } + + return { + ...row, + cells: newCells, + } + } + return row + }) + + // Store all rows including empty ones - don't auto-remove + const dataToStore = updatedRows.map((row) => ({ + id: row.id, + tagName: row.cells.tagName || '', + fieldType: row.cells.type || 'text', + value: row.cells.value || '', + })) + + const jsonString = dataToStore.length > 0 ? JSON.stringify(dataToStore) : '' + + emitTagSelection(jsonString) + } + const handleAddRow = () => { if (isPreview || disabled) return @@ -520,7 +562,8 @@ export function DocumentTagEntry({ { - handleCellChange(activeTagDropdown.rowIndex, 'value', newValue) + // Use immediate emission for tag dropdown selections + handleTagDropdownSelection(activeTagDropdown.rowIndex, 'value', newValue) setActiveTagDropdown(null) }} blockId={blockId} diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/long-input.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/long-input.tsx index 22a4a84a038..eac806fafb0 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/long-input.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/long-input.tsx @@ -12,6 +12,7 @@ import { WandPromptBar } from '@/app/workspace/[workspaceId]/w/[workflowId]/comp import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value' import { useWand } from '@/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-wand' import type { SubBlockConfig } from '@/blocks/types' +import { useTagSelection } from '@/hooks/use-tag-selection' const logger = createLogger('LongInput') @@ -79,6 +80,8 @@ export function LongInput({ }, }) + const emitTagSelection = useTagSelection(blockId, subBlockId) + const [showEnvVars, setShowEnvVars] = useState(false) const [showTags, setShowTags] = useState(false) const [searchTerm, setSearchTerm] = useState('') @@ -428,7 +431,7 @@ export function LongInput({ if (onChange) { onChange(newValue) } else if (!isPreview) { - setStoreValue(newValue) + emitTagSelection(newValue) } }} searchTerm={searchTerm} @@ -445,7 +448,7 @@ export function LongInput({ if (onChange) { onChange(newValue) } else if (!isPreview) { - setStoreValue(newValue) + emitTagSelection(newValue) } }} blockId={blockId} diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/short-input.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/short-input.tsx index d60d4874e30..1050a084e4b 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/short-input.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/short-input.tsx @@ -8,6 +8,7 @@ import { createLogger } from '@/lib/logs/console/logger' import { cn } from '@/lib/utils' import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value' import type { SubBlockConfig } from '@/blocks/types' +import { useTagSelection } from '@/hooks/use-tag-selection' const logger = createLogger('ShortInput') @@ -57,6 +58,8 @@ export function ShortInput({ const overlayRef = useRef(null) const [activeSourceBlockId, setActiveSourceBlockId] = useState(null) + const emitTagSelection = useTagSelection(blockId, subBlockId) + // Get ReactFlow instance for zoom control const reactFlowInstance = useReactFlow() @@ -288,8 +291,7 @@ export function ShortInput({ if (onChange) { onChange(newValue) } else if (!isPreview) { - // Only update store when not in preview mode - setStoreValue(newValue) + emitTagSelection(newValue) } } diff --git a/apps/sim/contexts/socket-context.tsx b/apps/sim/contexts/socket-context.tsx index 5ff231be315..bb17fd0965b 100644 --- a/apps/sim/contexts/socket-context.tsx +++ b/apps/sim/contexts/socket-context.tsx @@ -50,6 +50,7 @@ interface SocketContextType { value: any, operationId?: string ) => void + emitTagSelection: (blockId: string, subblockId: string, value: any) => void emitCursorUpdate: (cursor: { x: number; y: number }) => void emitSelectionUpdate: (selection: { type: 'block' | 'edge' | 'none'; id?: string }) => void @@ -77,6 +78,7 @@ const SocketContext = createContext({ leaveWorkflow: () => {}, emitWorkflowOperation: () => {}, emitSubblockUpdate: () => {}, + emitTagSelection: () => {}, emitCursorUpdate: () => {}, emitSelectionUpdate: () => {}, onWorkflowOperation: () => {}, @@ -292,6 +294,11 @@ export function SocketProvider({ children, user }: SocketProviderProps) { eventHandlers.current.subblockUpdate?.(data) }) + // Tag selection events (immediate dropdown selections) + socketInstance.on('tag-selection', (data) => { + eventHandlers.current.subblockUpdate?.(data) + }) + // Workflow deletion events socketInstance.on('workflow-deleted', (data) => { logger.warn(`Workflow ${data.workflowId} has been deleted`) @@ -697,6 +704,29 @@ export function SocketProvider({ children, user }: SocketProviderProps) { [socket, currentWorkflowId] ) + // Emit immediate tag/dropdown selections (bypasses operation queue) + const emitTagSelection = useCallback( + (blockId: string, subblockId: string, value: any) => { + // Only emit if socket is connected and we're in a valid workflow room + if (socket && currentWorkflowId) { + socket.emit('tag-selection', { + blockId, + subblockId, + value, + timestamp: Date.now(), + }) + } else { + logger.warn('Cannot emit tag selection: no socket connection or workflow room', { + hasSocket: !!socket, + currentWorkflowId, + blockId, + subblockId, + }) + } + }, + [socket, currentWorkflowId] + ) + // Cursor throttling optimized for database connection health const lastCursorEmit = useRef(0) const emitCursorUpdate = useCallback( @@ -776,6 +806,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) { leaveWorkflow, emitWorkflowOperation, emitSubblockUpdate, + emitTagSelection, emitCursorUpdate, emitSelectionUpdate, diff --git a/apps/sim/hooks/use-tag-selection.ts b/apps/sim/hooks/use-tag-selection.ts new file mode 100644 index 00000000000..3444083393c --- /dev/null +++ b/apps/sim/hooks/use-tag-selection.ts @@ -0,0 +1,25 @@ +import { useCallback } from 'react' +import { useSocket } from '@/contexts/socket-context' +import { useSubBlockStore } from '@/stores/workflows/subblock/store' + +/** + * Hook for handling immediate tag dropdown selections + * This bypasses the debounced operation queue system for instant feedback + */ +export function useTagSelection(blockId: string, subblockId: string) { + const { emitTagSelection } = useSocket() + const subBlockStore = useSubBlockStore() + + const emitTagSelectionValue = useCallback( + (value: any) => { + // Update local store immediately for instant feedback + subBlockStore.setValue(blockId, subblockId, value) + + // Emit to server immediately (no debouncing) + emitTagSelection(blockId, subblockId, value) + }, + [blockId, subblockId, emitTagSelection, subBlockStore] + ) + + return emitTagSelectionValue +} diff --git a/apps/sim/socket-server/handlers/subblocks.ts b/apps/sim/socket-server/handlers/subblocks.ts index a71a1fee193..f45e357a79e 100644 --- a/apps/sim/socket-server/handlers/subblocks.ts +++ b/apps/sim/socket-server/handlers/subblocks.ts @@ -158,4 +158,119 @@ export function setupSubblocksHandlers( }) } }) + + // Handle immediate tag/dropdown selections (no database persistence, just broadcast) + socket.on('tag-selection', async (data) => { + const workflowId = roomManager.getWorkflowIdForSocket(socket.id) + const session = roomManager.getUserSession(socket.id) + + if (!workflowId || !session) { + logger.debug(`Ignoring tag selection: socket not connected to any workflow room`, { + socketId: socket.id, + hasWorkflowId: !!workflowId, + hasSession: !!session, + }) + return + } + + const { blockId, subblockId, value, timestamp } = data + const room = roomManager.getWorkflowRoom(workflowId) + + if (!room) { + logger.debug(`Ignoring tag selection: workflow room not found`, { + socketId: socket.id, + workflowId, + blockId, + subblockId, + }) + return + } + + try { + const userPresence = room.users.get(socket.id) + if (userPresence) { + userPresence.lastActivity = Date.now() + } + + // First, verify that the workflow still exists in the database + const workflowExists = await db + .select({ id: workflow.id }) + .from(workflow) + .where(eq(workflow.id, workflowId)) + .limit(1) + + if (workflowExists.length === 0) { + logger.warn(`Ignoring tag selection: workflow ${workflowId} no longer exists`, { + socketId: socket.id, + blockId, + subblockId, + }) + roomManager.cleanupUserFromRoom(socket.id, workflowId) + return + } + + // Persist to database immediately (same logic as subblock-update but without operation tracking) + let updateSuccessful = false + await db.transaction(async (tx) => { + const [block] = await tx + .select({ subBlocks: workflowBlocks.subBlocks }) + .from(workflowBlocks) + .where(and(eq(workflowBlocks.id, blockId), eq(workflowBlocks.workflowId, workflowId))) + .limit(1) + + if (!block) { + // Block was deleted - this is a normal race condition in collaborative editing + logger.debug( + `Ignoring tag selection for deleted block: ${workflowId}/${blockId}.${subblockId}` + ) + return + } + + const subBlocks = (block.subBlocks as any) || {} + + if (!subBlocks[subblockId]) { + // Create new subblock with minimal structure + subBlocks[subblockId] = { + id: subblockId, + type: 'unknown', // Will be corrected by next collaborative update + value: value, + } + } else { + // Preserve existing id and type, only update value + subBlocks[subblockId] = { + ...subBlocks[subblockId], + value: value, + } + } + + await tx + .update(workflowBlocks) + .set({ + subBlocks: subBlocks, + updatedAt: new Date(), + }) + .where(and(eq(workflowBlocks.id, blockId), eq(workflowBlocks.workflowId, workflowId))) + + updateSuccessful = true + }) + + // Broadcast to other clients if the update was successful + if (updateSuccessful) { + socket.to(workflowId).emit('tag-selection', { + blockId, + subblockId, + value, + timestamp, + senderId: socket.id, + userId: session.userId, + }) + + logger.debug( + `Tag selection persisted and broadcast in workflow ${workflowId}: ${blockId}.${subblockId}` + ) + } + } catch (error) { + logger.error('Error handling tag selection:', error) + } + }) } From 6b3af4d781b497f5732e38663ab18faf19b1e816 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Fri, 1 Aug 2025 14:37:58 -0700 Subject: [PATCH 2/6] do not bypass op queue for tag selections --- apps/sim/hooks/use-collaborative-workflow.ts | 26 ++++++++++++++++++++ apps/sim/hooks/use-tag-selection.ts | 17 +++++-------- apps/sim/stores/operation-queue/store.ts | 2 +- 3 files changed, 33 insertions(+), 12 deletions(-) diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index 6a0ff15e4d1..5b299d2d281 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -22,6 +22,7 @@ export function useCollaborativeWorkflow() { leaveWorkflow, emitWorkflowOperation, emitSubblockUpdate, + emitTagSelection, onWorkflowOperation, onSubblockUpdate, onUserJoined, @@ -723,6 +724,30 @@ export function useCollaborativeWorkflow() { ] ) + // Immediate tag selection (bypasses debouncing for instant collaborative feedback) + const collaborativeSetTagSelection = useCallback( + (blockId: string, subblockId: string, value: any) => { + if (isApplyingRemoteChange.current) return + + if (!currentWorkflowId || activeWorkflowId !== currentWorkflowId) { + logger.debug('Skipping tag selection - not in active workflow', { + currentWorkflowId, + activeWorkflowId, + blockId, + subblockId, + }) + return + } + + // Apply locally first (immediate UI feedback) + subBlockStore.setValue(blockId, subblockId, value) + + // Emit immediately using the tag-selection socket event (no debouncing) + emitTagSelection(blockId, subblockId, value) + }, + [subBlockStore, emitTagSelection, currentWorkflowId, activeWorkflowId] + ) + const collaborativeDuplicateBlock = useCallback( (sourceId: string) => { const sourceBlock = workflowStore.blocks[sourceId] @@ -1019,6 +1044,7 @@ export function useCollaborativeWorkflow() { collaborativeAddEdge, collaborativeRemoveEdge, collaborativeSetSubblockValue, + collaborativeSetTagSelection, // Collaborative loop/parallel operations collaborativeUpdateLoopCount, diff --git a/apps/sim/hooks/use-tag-selection.ts b/apps/sim/hooks/use-tag-selection.ts index 3444083393c..357111e95d2 100644 --- a/apps/sim/hooks/use-tag-selection.ts +++ b/apps/sim/hooks/use-tag-selection.ts @@ -1,24 +1,19 @@ import { useCallback } from 'react' -import { useSocket } from '@/contexts/socket-context' -import { useSubBlockStore } from '@/stores/workflows/subblock/store' +import { useCollaborativeWorkflow } from '@/hooks/use-collaborative-workflow' /** * Hook for handling immediate tag dropdown selections - * This bypasses the debounced operation queue system for instant feedback + * Uses the collaborative workflow system but with immediate processing */ export function useTagSelection(blockId: string, subblockId: string) { - const { emitTagSelection } = useSocket() - const subBlockStore = useSubBlockStore() + const { collaborativeSetTagSelection } = useCollaborativeWorkflow() const emitTagSelectionValue = useCallback( (value: any) => { - // Update local store immediately for instant feedback - subBlockStore.setValue(blockId, subblockId, value) - - // Emit to server immediately (no debouncing) - emitTagSelection(blockId, subblockId, value) + // Use the collaborative system with immediate processing (no debouncing) + collaborativeSetTagSelection(blockId, subblockId, value) }, - [blockId, subblockId, emitTagSelection, subBlockStore] + [blockId, subblockId, collaborativeSetTagSelection] ) return emitTagSelectionValue diff --git a/apps/sim/stores/operation-queue/store.ts b/apps/sim/stores/operation-queue/store.ts index 7d65c82d168..bb51131f9b2 100644 --- a/apps/sim/stores/operation-queue/store.ts +++ b/apps/sim/stores/operation-queue/store.ts @@ -59,7 +59,7 @@ export const useOperationQueueStore = create((set, get) => hasOperationError: false, addToQueue: (operation) => { - // Handle debouncing for subblock operations + // Handle debouncing for regular subblock operations (but not tag selections) if ( operation.operation.operation === 'subblock-update' && operation.operation.target === 'subblock' From 46efa5c560724dfc226f254488faa795d7dbd305 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Fri, 1 Aug 2025 14:39:15 -0700 Subject: [PATCH 3/6] Update apps/sim/socket-server/handlers/subblocks.ts Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --- apps/sim/socket-server/handlers/subblocks.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/sim/socket-server/handlers/subblocks.ts b/apps/sim/socket-server/handlers/subblocks.ts index f45e357a79e..3cfe7e8fcae 100644 --- a/apps/sim/socket-server/handlers/subblocks.ts +++ b/apps/sim/socket-server/handlers/subblocks.ts @@ -159,7 +159,7 @@ export function setupSubblocksHandlers( } }) - // Handle immediate tag/dropdown selections (no database persistence, just broadcast) + // Handle immediate tag/dropdown selections (persist to database and broadcast) socket.on('tag-selection', async (data) => { const workflowId = roomManager.getWorkflowIdForSocket(socket.id) const session = roomManager.getUserSession(socket.id) From 34954714373f9f1edb58ac496dc6c9caeaf4de1b Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Fri, 1 Aug 2025 14:55:20 -0700 Subject: [PATCH 4/6] prevent race cond between subblock update event and tag selection --- apps/sim/contexts/socket-context.tsx | 31 ----- apps/sim/hooks/use-collaborative-workflow.ts | 21 +++- apps/sim/socket-server/handlers/subblocks.ts | 115 ------------------- apps/sim/stores/operation-queue/store.ts | 6 +- 4 files changed, 20 insertions(+), 153 deletions(-) diff --git a/apps/sim/contexts/socket-context.tsx b/apps/sim/contexts/socket-context.tsx index bb17fd0965b..5ff231be315 100644 --- a/apps/sim/contexts/socket-context.tsx +++ b/apps/sim/contexts/socket-context.tsx @@ -50,7 +50,6 @@ interface SocketContextType { value: any, operationId?: string ) => void - emitTagSelection: (blockId: string, subblockId: string, value: any) => void emitCursorUpdate: (cursor: { x: number; y: number }) => void emitSelectionUpdate: (selection: { type: 'block' | 'edge' | 'none'; id?: string }) => void @@ -78,7 +77,6 @@ const SocketContext = createContext({ leaveWorkflow: () => {}, emitWorkflowOperation: () => {}, emitSubblockUpdate: () => {}, - emitTagSelection: () => {}, emitCursorUpdate: () => {}, emitSelectionUpdate: () => {}, onWorkflowOperation: () => {}, @@ -294,11 +292,6 @@ export function SocketProvider({ children, user }: SocketProviderProps) { eventHandlers.current.subblockUpdate?.(data) }) - // Tag selection events (immediate dropdown selections) - socketInstance.on('tag-selection', (data) => { - eventHandlers.current.subblockUpdate?.(data) - }) - // Workflow deletion events socketInstance.on('workflow-deleted', (data) => { logger.warn(`Workflow ${data.workflowId} has been deleted`) @@ -704,29 +697,6 @@ export function SocketProvider({ children, user }: SocketProviderProps) { [socket, currentWorkflowId] ) - // Emit immediate tag/dropdown selections (bypasses operation queue) - const emitTagSelection = useCallback( - (blockId: string, subblockId: string, value: any) => { - // Only emit if socket is connected and we're in a valid workflow room - if (socket && currentWorkflowId) { - socket.emit('tag-selection', { - blockId, - subblockId, - value, - timestamp: Date.now(), - }) - } else { - logger.warn('Cannot emit tag selection: no socket connection or workflow room', { - hasSocket: !!socket, - currentWorkflowId, - blockId, - subblockId, - }) - } - }, - [socket, currentWorkflowId] - ) - // Cursor throttling optimized for database connection health const lastCursorEmit = useRef(0) const emitCursorUpdate = useCallback( @@ -806,7 +776,6 @@ export function SocketProvider({ children, user }: SocketProviderProps) { leaveWorkflow, emitWorkflowOperation, emitSubblockUpdate, - emitTagSelection, emitCursorUpdate, emitSelectionUpdate, diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index 5b299d2d281..76d2cfdd181 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -22,7 +22,6 @@ export function useCollaborativeWorkflow() { leaveWorkflow, emitWorkflowOperation, emitSubblockUpdate, - emitTagSelection, onWorkflowOperation, onSubblockUpdate, onUserJoined, @@ -724,7 +723,7 @@ export function useCollaborativeWorkflow() { ] ) - // Immediate tag selection (bypasses debouncing for instant collaborative feedback) + // Immediate tag selection (uses queue but processes immediately, no debouncing) const collaborativeSetTagSelection = useCallback( (blockId: string, subblockId: string, value: any) => { if (isApplyingRemoteChange.current) return @@ -742,10 +741,22 @@ export function useCollaborativeWorkflow() { // Apply locally first (immediate UI feedback) subBlockStore.setValue(blockId, subblockId, value) - // Emit immediately using the tag-selection socket event (no debouncing) - emitTagSelection(blockId, subblockId, value) + // Use the operation queue but with immediate processing (no debouncing) + const operationId = crypto.randomUUID() + + addToQueue({ + id: operationId, + operation: { + operation: 'subblock-update', + target: 'subblock', + payload: { blockId, subblockId, value }, + }, + workflowId: activeWorkflowId, + userId: session?.user?.id || 'unknown', + immediate: true, + }) }, - [subBlockStore, emitTagSelection, currentWorkflowId, activeWorkflowId] + [subBlockStore, addToQueue, currentWorkflowId, activeWorkflowId, session?.user?.id] ) const collaborativeDuplicateBlock = useCallback( diff --git a/apps/sim/socket-server/handlers/subblocks.ts b/apps/sim/socket-server/handlers/subblocks.ts index f45e357a79e..a71a1fee193 100644 --- a/apps/sim/socket-server/handlers/subblocks.ts +++ b/apps/sim/socket-server/handlers/subblocks.ts @@ -158,119 +158,4 @@ export function setupSubblocksHandlers( }) } }) - - // Handle immediate tag/dropdown selections (no database persistence, just broadcast) - socket.on('tag-selection', async (data) => { - const workflowId = roomManager.getWorkflowIdForSocket(socket.id) - const session = roomManager.getUserSession(socket.id) - - if (!workflowId || !session) { - logger.debug(`Ignoring tag selection: socket not connected to any workflow room`, { - socketId: socket.id, - hasWorkflowId: !!workflowId, - hasSession: !!session, - }) - return - } - - const { blockId, subblockId, value, timestamp } = data - const room = roomManager.getWorkflowRoom(workflowId) - - if (!room) { - logger.debug(`Ignoring tag selection: workflow room not found`, { - socketId: socket.id, - workflowId, - blockId, - subblockId, - }) - return - } - - try { - const userPresence = room.users.get(socket.id) - if (userPresence) { - userPresence.lastActivity = Date.now() - } - - // First, verify that the workflow still exists in the database - const workflowExists = await db - .select({ id: workflow.id }) - .from(workflow) - .where(eq(workflow.id, workflowId)) - .limit(1) - - if (workflowExists.length === 0) { - logger.warn(`Ignoring tag selection: workflow ${workflowId} no longer exists`, { - socketId: socket.id, - blockId, - subblockId, - }) - roomManager.cleanupUserFromRoom(socket.id, workflowId) - return - } - - // Persist to database immediately (same logic as subblock-update but without operation tracking) - let updateSuccessful = false - await db.transaction(async (tx) => { - const [block] = await tx - .select({ subBlocks: workflowBlocks.subBlocks }) - .from(workflowBlocks) - .where(and(eq(workflowBlocks.id, blockId), eq(workflowBlocks.workflowId, workflowId))) - .limit(1) - - if (!block) { - // Block was deleted - this is a normal race condition in collaborative editing - logger.debug( - `Ignoring tag selection for deleted block: ${workflowId}/${blockId}.${subblockId}` - ) - return - } - - const subBlocks = (block.subBlocks as any) || {} - - if (!subBlocks[subblockId]) { - // Create new subblock with minimal structure - subBlocks[subblockId] = { - id: subblockId, - type: 'unknown', // Will be corrected by next collaborative update - value: value, - } - } else { - // Preserve existing id and type, only update value - subBlocks[subblockId] = { - ...subBlocks[subblockId], - value: value, - } - } - - await tx - .update(workflowBlocks) - .set({ - subBlocks: subBlocks, - updatedAt: new Date(), - }) - .where(and(eq(workflowBlocks.id, blockId), eq(workflowBlocks.workflowId, workflowId))) - - updateSuccessful = true - }) - - // Broadcast to other clients if the update was successful - if (updateSuccessful) { - socket.to(workflowId).emit('tag-selection', { - blockId, - subblockId, - value, - timestamp, - senderId: socket.id, - userId: session.userId, - }) - - logger.debug( - `Tag selection persisted and broadcast in workflow ${workflowId}: ${blockId}.${subblockId}` - ) - } - } catch (error) { - logger.error('Error handling tag selection:', error) - } - }) } diff --git a/apps/sim/stores/operation-queue/store.ts b/apps/sim/stores/operation-queue/store.ts index bb51131f9b2..a1a656cc4a9 100644 --- a/apps/sim/stores/operation-queue/store.ts +++ b/apps/sim/stores/operation-queue/store.ts @@ -15,6 +15,7 @@ export interface QueuedOperation { retryCount: number status: 'pending' | 'processing' | 'confirmed' | 'failed' userId: string + immediate?: boolean // Flag for immediate processing (skips debouncing) } interface OperationQueueState { @@ -59,10 +60,11 @@ export const useOperationQueueStore = create((set, get) => hasOperationError: false, addToQueue: (operation) => { - // Handle debouncing for regular subblock operations (but not tag selections) + // Handle debouncing for regular subblock operations (but not immediate ones like tag selections) if ( operation.operation.operation === 'subblock-update' && - operation.operation.target === 'subblock' + operation.operation.target === 'subblock' && + !operation.immediate ) { const { blockId, subblockId } = operation.operation.payload const debounceKey = `${blockId}-${subblockId}` From f16a8af37c672e3f352cbee3d7ec32bc9319b3da Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Fri, 1 Aug 2025 15:08:47 -0700 Subject: [PATCH 5/6] refactor --- .../sub-block/components/condition-input.tsx | 66 +++++++++++++- .../document-tag-entry/document-tag-entry.tsx | 87 +++++++------------ .../knowledge-tag-filters.tsx | 31 ++++++- 3 files changed, 123 insertions(+), 61 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/condition-input.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/condition-input.tsx index 51f5ccf051e..c3f9336cf1d 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/condition-input.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/condition-input.tsx @@ -14,6 +14,7 @@ import { Tooltip, TooltipContent, TooltipTrigger } from '@/components/ui/tooltip import { createLogger } from '@/lib/logs/console/logger' import { cn } from '@/lib/utils' import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value' +import { useTagSelection } from '@/hooks/use-tag-selection' import { useWorkflowStore } from '@/stores/workflows/workflow/store' const logger = createLogger('ConditionInput') @@ -52,6 +53,9 @@ export function ConditionInput({ disabled = false, }: ConditionInputProps) { const [storeValue, setStoreValue] = useSubBlockValue(blockId, subBlockId) + + const emitTagSelection = useTagSelection(blockId, subBlockId) + const containerRef = useRef(null) const [visualLineHeights, setVisualLineHeights] = useState<{ [key: string]: number[] @@ -400,6 +404,64 @@ export function ConditionInput({ ) } + const handleTagSelectImmediate = (blockId: string, newValue: string) => { + if (isPreview || disabled) return + + setConditionalBlocks((blocks) => + blocks.map((block) => + block.id === blockId + ? { + ...block, + value: newValue, + showTags: false, + activeSourceBlockId: null, + } + : block + ) + ) + + const updatedBlocks = conditionalBlocks.map((block) => + block.id === blockId + ? { + ...block, + value: newValue, + showTags: false, + activeSourceBlockId: null, + } + : block + ) + emitTagSelection(JSON.stringify(updatedBlocks)) + } + + const handleEnvVarSelectImmediate = (blockId: string, newValue: string) => { + if (isPreview || disabled) return + + setConditionalBlocks((blocks) => + blocks.map((block) => + block.id === blockId + ? { + ...block, + value: newValue, + showEnvVars: false, + searchTerm: '', + } + : block + ) + ) + + const updatedBlocks = conditionalBlocks.map((block) => + block.id === blockId + ? { + ...block, + value: newValue, + showEnvVars: false, + searchTerm: '', + } + : block + ) + emitTagSelection(JSON.stringify(updatedBlocks)) + } + // Update block titles based on position const updateBlockTitles = (blocks: ConditionalBlock[]): ConditionalBlock[] => { return blocks.map((block, index) => ({ @@ -706,7 +768,7 @@ export function ConditionInput({ {block.showEnvVars && ( handleEnvVarSelect(block.id, newValue)} + onSelect={(newValue) => handleEnvVarSelectImmediate(block.id, newValue)} searchTerm={block.searchTerm} inputValue={block.value} cursorPosition={block.cursorPosition} @@ -723,7 +785,7 @@ export function ConditionInput({ {block.showTags && ( handleTagSelect(block.id, newValue)} + onSelect={(newValue) => handleTagSelectImmediate(block.id, newValue)} blockId={blockId} activeSourceBlockId={block.activeSourceBlockId} inputValue={block.value} diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx index 105822e751b..29a69899f78 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/document-tag-entry/document-tag-entry.tsx @@ -131,33 +131,8 @@ export function DocumentTagEntry({ setStoreValue(jsonString) } - const handleCellChange = (rowIndex: number, column: string, value: string) => { - if (isPreview || disabled) return - - // Check if this is a new tag name that would exceed the limit - if (column === 'tagName' && value.trim()) { - const isExistingTag = tagDefinitions.some( - (def) => def.displayName.toLowerCase() === value.toLowerCase() - ) - - if (!isExistingTag) { - // Count current new tags being created (excluding the current row) - const currentNewTags = rows.filter( - (row, idx) => - idx !== rowIndex && - row.cells.tagName?.trim() && - !tagDefinitions.some( - (def) => def.displayName.toLowerCase() === row.cells.tagName.toLowerCase() - ) - ).length - - if (tagDefinitions.length + currentNewTags >= MAX_TAG_SLOTS) { - // Don't allow creating new tags if we've reached the limit - return - } - } - } - + // Shared helper function for updating rows and generating JSON + const updateRowsAndGenerateJson = (rowIndex: number, column: string, value: string) => { const updatedRows = [...rows].map((row, idx) => { if (idx === rowIndex) { const newCells = { ...row.cells, [column]: value } @@ -180,8 +155,6 @@ export function DocumentTagEntry({ return row }) - // No auto-add rows - user will manually add them with plus button - // Store all rows including empty ones - don't auto-remove const dataToStore = updatedRows.map((row) => ({ id: row.id, @@ -190,46 +163,44 @@ export function DocumentTagEntry({ value: row.cells.value || '', })) - const jsonString = dataToStore.length > 0 ? JSON.stringify(dataToStore) : '' - setStoreValue(jsonString) + return dataToStore.length > 0 ? JSON.stringify(dataToStore) : '' } - // Special handler for tag dropdown selections that uses immediate emission - const handleTagDropdownSelection = (rowIndex: number, column: string, value: string) => { + const handleCellChange = (rowIndex: number, column: string, value: string) => { if (isPreview || disabled) return - const updatedRows = [...rows].map((row, idx) => { - if (idx === rowIndex) { - const newCells = { ...row.cells, [column]: value } + // Check if this is a new tag name that would exceed the limit + if (column === 'tagName' && value.trim()) { + const isExistingTag = tagDefinitions.some( + (def) => def.displayName.toLowerCase() === value.toLowerCase() + ) - // Auto-select type when existing tag is selected - if (column === 'tagName' && value) { - const tagDef = tagDefinitions.find( - (def) => def.displayName.toLowerCase() === value.toLowerCase() - ) - if (tagDef) { - newCells.type = tagDef.fieldType - } - } + if (!isExistingTag) { + // Count current new tags being created (excluding the current row) + const currentNewTags = rows.filter( + (row, idx) => + idx !== rowIndex && + row.cells.tagName?.trim() && + !tagDefinitions.some( + (def) => def.displayName.toLowerCase() === row.cells.tagName.toLowerCase() + ) + ).length - return { - ...row, - cells: newCells, + if (tagDefinitions.length + currentNewTags >= MAX_TAG_SLOTS) { + // Don't allow creating new tags if we've reached the limit + return } } - return row - }) + } - // Store all rows including empty ones - don't auto-remove - const dataToStore = updatedRows.map((row) => ({ - id: row.id, - tagName: row.cells.tagName || '', - fieldType: row.cells.type || 'text', - value: row.cells.value || '', - })) + const jsonString = updateRowsAndGenerateJson(rowIndex, column, value) + setStoreValue(jsonString) + } - const jsonString = dataToStore.length > 0 ? JSON.stringify(dataToStore) : '' + const handleTagDropdownSelection = (rowIndex: number, column: string, value: string) => { + if (isPreview || disabled) return + const jsonString = updateRowsAndGenerateJson(rowIndex, column, value) emitTagSelection(jsonString) } diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/knowledge-tag-filters/knowledge-tag-filters.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/knowledge-tag-filters/knowledge-tag-filters.tsx index bdf7fa9b139..63dd78613ab 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/knowledge-tag-filters/knowledge-tag-filters.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/knowledge-tag-filters/knowledge-tag-filters.tsx @@ -9,6 +9,7 @@ import { Label } from '@/components/ui/label' import { checkTagTrigger, TagDropdown } from '@/components/ui/tag-dropdown' import type { SubBlockConfig } from '@/blocks/types' import { useKnowledgeBaseTagDefinitions } from '@/hooks/use-knowledge-base-tag-definitions' +import { useTagSelection } from '@/hooks/use-tag-selection' import { useSubBlockValue } from '../../hooks/use-sub-block-value' interface TagFilter { @@ -44,6 +45,9 @@ export function KnowledgeTagFilters({ }: KnowledgeTagFiltersProps) { const [storeValue, setStoreValue] = useSubBlockValue(blockId, subBlock.id) + // Hook for immediate tag/dropdown selections + const emitTagSelection = useTagSelection(blockId, subBlock.id) + // Get the knowledge base ID from other sub-blocks const [knowledgeBaseIdValue] = useSubBlockValue(blockId, 'knowledgeBaseId') const knowledgeBaseId = knowledgeBaseIdValue || null @@ -122,6 +126,30 @@ export function KnowledgeTagFilters({ updateFilters(updatedFilters) } + const handleTagDropdownSelection = (rowIndex: number, column: string, value: string) => { + if (isPreview || disabled) return + + const updatedRows = [...rows].map((row, idx) => { + if (idx === rowIndex) { + return { + ...row, + cells: { ...row.cells, [column]: value }, + } + } + return row + }) + + // Convert back to TagFilter format - keep all rows, even empty ones + const updatedFilters = updatedRows.map((row) => ({ + id: row.id, + tagName: row.cells.tagName || '', + tagValue: row.cells.value || '', + })) + + const jsonValue = updatedFilters.length > 0 ? JSON.stringify(updatedFilters) : null + emitTagSelection(jsonValue) + } + const handleAddRow = () => { if (isPreview || disabled) return @@ -336,7 +364,8 @@ export function KnowledgeTagFilters({ { - handleCellChange(activeTagDropdown.rowIndex, 'value', newValue) + // Use immediate emission for tag dropdown selections + handleTagDropdownSelection(activeTagDropdown.rowIndex, 'value', newValue) setActiveTagDropdown(null) }} blockId={blockId} From cc23429736d250a566fd5baf1a04cc02f1800a37 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Fri, 1 Aug 2025 15:20:11 -0700 Subject: [PATCH 6/6] reduce debounce time to 50ms --- apps/sim/stores/operation-queue/store.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/sim/stores/operation-queue/store.ts b/apps/sim/stores/operation-queue/store.ts index a1a656cc4a9..9ddcb2e543a 100644 --- a/apps/sim/stores/operation-queue/store.ts +++ b/apps/sim/stores/operation-queue/store.ts @@ -102,7 +102,7 @@ export const useOperationQueueStore = create((set, get) => })) get().processNextOperation() - }, 100) // 100ms debounce for subblock operations + }, 50) // 50ms debounce for subblock operations - optimized for collaborative editing subblockDebounceTimeouts.set(debounceKey, timeoutId) return