improvement(sockets): position persistence on drag end, perms call only on joining room (#1571)

icecrasher321 · web-flow · commit 8e43774b5e4d · 2025-10-07T17:31:02.000-07:00
diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/workflow.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/workflow.tsx
@@ -1422,7 +1422,7 @@ const WorkflowContent = React.memo(() => {
       setDraggedNodeId(node.id)
 
       // Emit collaborative position update during drag for smooth real-time movement
-      collaborativeUpdateBlockPosition(node.id, node.position)
+      collaborativeUpdateBlockPosition(node.id, node.position, false)
 
       // Get the current parent ID of the node being dragged
       const currentParentId = blocks[node.id]?.data?.parentId || null
@@ -1608,7 +1608,7 @@ const WorkflowContent = React.memo(() => {
 
       // Emit collaborative position update for the final position
       // This ensures other users see the smooth final position
-      collaborativeUpdateBlockPosition(node.id, node.position)
+      collaborativeUpdateBlockPosition(node.id, node.position, true)
 
       // Record single move entry on drag end to avoid micro-moves
       try {
diff --git a/apps/sim/contexts/socket-context.tsx b/apps/sim/contexts/socket-context.tsx
@@ -600,30 +600,45 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
 
       // Apply light throttling only to position updates for smooth collaborative experience
       const isPositionUpdate = operation === 'update-position' && target === 'block'
+      const { commit = true } = payload || {}
 
       if (isPositionUpdate && payload.id) {
         const blockId = payload.id
 
-        // Store the latest position update
+        if (commit) {
+          socket.emit('workflow-operation', {
+            operation,
+            target,
+            payload,
+            timestamp: Date.now(),
+            operationId,
+          })
+          pendingPositionUpdates.current.delete(blockId)
+          const timeoutId = positionUpdateTimeouts.current.get(blockId)
+          if (timeoutId) {
+            clearTimeout(timeoutId)
+            positionUpdateTimeouts.current.delete(blockId)
+          }
+          return
+        }
+
         pendingPositionUpdates.current.set(blockId, {
           operation,
           target,
           payload,
           timestamp: Date.now(),
-          operationId, // Include operation ID for queue tracking
+          operationId,
         })
 
-        // Check if we already have a pending timeout for this block
         if (!positionUpdateTimeouts.current.has(blockId)) {
-          // Schedule emission with optimized throttling (30fps = ~33ms) to reduce DB load
           const timeoutId = window.setTimeout(() => {
             const latestUpdate = pendingPositionUpdates.current.get(blockId)
             if (latestUpdate) {
               socket.emit('workflow-operation', latestUpdate)
               pendingPositionUpdates.current.delete(blockId)
             }
             positionUpdateTimeouts.current.delete(blockId)
-          }, 33) // 30fps - good balance between smoothness and DB performance
+          }, 33)
 
           positionUpdateTimeouts.current.set(blockId, timeoutId)
         }
diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts
@@ -868,13 +868,19 @@ export function useCollaborativeWorkflow() {
   )
 
   const collaborativeUpdateBlockPosition = useCallback(
-    (id: string, position: Position) => {
-      // Only apply position updates here (no undo recording to avoid micro-moves)
-      executeQueuedDebouncedOperation('update-position', 'block', { id, position }, () =>
+    (id: string, position: Position, commit = true) => {
+      if (commit) {
+        executeQueuedOperation('update-position', 'block', { id, position, commit }, () => {
+          workflowStore.updateBlockPosition(id, position)
+        })
+        return
+      }
+
+      executeQueuedDebouncedOperation('update-position', 'block', { id, position }, () => {
         workflowStore.updateBlockPosition(id, position)
-      )
+      })
     },
-    [executeQueuedDebouncedOperation, workflowStore]
+    [executeQueuedDebouncedOperation, executeQueuedOperation, workflowStore]
   )
 
   const collaborativeUpdateBlockName = useCallback(
diff --git a/apps/sim/hooks/use-undo-redo.ts b/apps/sim/hooks/use-undo-redo.ts
@@ -603,6 +603,7 @@ export function useUndoRedo() {
                 id: moveOp.data.blockId,
                 position: { x: moveOp.data.after.x, y: moveOp.data.after.y },
                 parentId: moveOp.data.after.parentId,
+                commit: true,
                 isUndo: true,
                 originalOpId: entry.id,
               },
@@ -706,6 +707,7 @@ export function useUndoRedo() {
               payload: {
                 id: blockId,
                 position: newPosition,
+                commit: true,
                 isUndo: true,
                 originalOpId: entry.id,
               },
diff --git a/apps/sim/socket-server/database/operations.ts b/apps/sim/socket-server/database/operations.ts
@@ -340,6 +340,10 @@ async function handleBlockOperationTx(
         throw new Error('Missing required fields for update position operation')
       }
 
+      if (payload.commit !== true) {
+        return
+      }
+
       const updateResult = await tx
         .update(workflowBlocks)
         .set({
diff --git a/apps/sim/socket-server/handlers/operations.ts b/apps/sim/socket-server/handlers/operations.ts
@@ -3,7 +3,7 @@ import { createLogger } from '@/lib/logs/console/logger'
 import { persistWorkflowOperation } from '@/socket-server/database/operations'
 import type { HandlerDependencies } from '@/socket-server/handlers/workflow'
 import type { AuthenticatedSocket } from '@/socket-server/middleware/auth'
-import { verifyOperationPermission } from '@/socket-server/middleware/permissions'
+import { checkRolePermission } from '@/socket-server/middleware/permissions'
 import type { RoomManager } from '@/socket-server/rooms/manager'
 import { WorkflowOperationSchema } from '@/socket-server/validation/schemas'
 
@@ -43,36 +43,52 @@ export function setupOperationsHandlers(
       operationId = validatedOperation.operationId
       const { operation, target, payload, timestamp } = validatedOperation
 
-      // Check operation permissions
-      const permissionCheck = await verifyOperationPermission(
-        session.userId,
-        workflowId,
-        operation,
-        target
-      )
-      if (!permissionCheck.allowed) {
-        logger.warn(
-          `User ${session.userId} forbidden from ${operation} on ${target}: ${permissionCheck.reason}`
-        )
-        socket.emit('operation-forbidden', {
-          type: 'INSUFFICIENT_PERMISSIONS',
-          message: permissionCheck.reason || 'Insufficient permissions for this operation',
-          operation,
-          target,
-        })
-        return
-      }
-
-      const userPresence = room.users.get(socket.id)
-      if (userPresence) {
-        userPresence.lastActivity = Date.now()
-      }
-
       // For position updates, preserve client timestamp to maintain ordering
       // For other operations, use server timestamp for consistency
       const isPositionUpdate = operation === 'update-position' && target === 'block'
+      const commitPositionUpdate =
+        isPositionUpdate && 'commit' in payload ? payload.commit === true : false
       const operationTimestamp = isPositionUpdate ? timestamp : Date.now()
 
+      // Skip permission checks for non-committed position updates (broadcasts only, no persistence)
+      if (isPositionUpdate && !commitPositionUpdate) {
+        // Update last activity
+        const userPresence = room.users.get(socket.id)
+        if (userPresence) {
+          userPresence.lastActivity = Date.now()
+        }
+      } else {
+        // Check permissions from cached role for all other operations
+        const userPresence = room.users.get(socket.id)
+        if (!userPresence) {
+          logger.warn(`User presence not found for socket ${socket.id}`)
+          socket.emit('operation-forbidden', {
+            type: 'SESSION_ERROR',
+            message: 'User session not found',
+            operation,
+            target,
+          })
+          return
+        }
+
+        userPresence.lastActivity = Date.now()
+
+        // Check permissions using cached role (no DB query)
+        const permissionCheck = checkRolePermission(userPresence.role, operation)
+        if (!permissionCheck.allowed) {
+          logger.warn(
+            `User ${session.userId} (role: ${userPresence.role}) forbidden from ${operation} on ${target}`
+          )
+          socket.emit('operation-forbidden', {
+            type: 'INSUFFICIENT_PERMISSIONS',
+            message: `${permissionCheck.reason} on '${target}'`,
+            operation,
+            target,
+          })
+          return
+        }
+      }
+
       // Broadcast first for position updates to minimize latency, then persist
       // For other operations, persist first for consistency
       if (isPositionUpdate) {
@@ -94,36 +110,39 @@ export function setupOperationsHandlers(
 
         socket.to(workflowId).emit('workflow-operation', broadcastData)
 
-        // Persist position update asynchronously to avoid blocking real-time updates
-        persistWorkflowOperation(workflowId, {
-          operation,
-          target,
-          payload,
-          timestamp: operationTimestamp,
-          userId: session.userId,
-        }).catch((error) => {
+        if (!commitPositionUpdate) {
+          return
+        }
+
+        try {
+          await persistWorkflowOperation(workflowId, {
+            operation,
+            target,
+            payload,
+            timestamp: operationTimestamp,
+            userId: session.userId,
+          })
+          room.lastModified = Date.now()
+
+          if (operationId) {
+            socket.emit('operation-confirmed', {
+              operationId,
+              serverTimestamp: Date.now(),
+            })
+          }
+        } catch (error) {
           logger.error('Failed to persist position update:', error)
-          // Emit failure for position updates if operationId is provided
+
           if (operationId) {
             socket.emit('operation-failed', {
               operationId,
               error: error instanceof Error ? error.message : 'Database persistence failed',
               retryable: true,
             })
           }
-        })
-
-        room.lastModified = Date.now()
-
-        // Emit confirmation if operationId is provided
-        if (operationId) {
-          socket.emit('operation-confirmed', {
-            operationId,
-            serverTimestamp: Date.now(),
-          })
         }
 
-        return // Early return for position updates
+        return
       }
 
       if (target === 'variable' && ['add', 'remove', 'duplicate'].includes(operation)) {
diff --git a/apps/sim/socket-server/handlers/workflow.ts b/apps/sim/socket-server/handlers/workflow.ts
@@ -46,13 +46,15 @@ export function setupWorkflowHandlers(
 
       logger.info(`Join workflow request from ${userId} (${userName}) for workflow ${workflowId}`)
 
+      let userRole: string
       try {
         const accessInfo = await verifyWorkflowAccess(userId, workflowId)
         if (!accessInfo.hasAccess) {
           logger.warn(`User ${userId} (${userName}) denied access to workflow ${workflowId}`)
           socket.emit('join-workflow-error', { error: 'Access denied to workflow' })
           return
         }
+        userRole = accessInfo.role || 'read'
       } catch (error) {
         logger.warn(`Error verifying workflow access for ${userId}:`, error)
         socket.emit('join-workflow-error', { error: 'Failed to verify workflow access' })
@@ -85,6 +87,7 @@ export function setupWorkflowHandlers(
         socketId: socket.id,
         joinedAt: Date.now(),
         lastActivity: Date.now(),
+        role: userRole,
       }
 
       room.users.set(socket.id, userPresence)
diff --git a/apps/sim/socket-server/index.test.ts b/apps/sim/socket-server/index.test.ts
@@ -40,9 +40,9 @@ vi.mock('@/socket-server/middleware/auth', () => ({
 vi.mock('@/socket-server/middleware/permissions', () => ({
   verifyWorkflowAccess: vi.fn().mockResolvedValue({
     hasAccess: true,
-    role: 'owner',
+    role: 'admin',
   }),
-  verifyOperationPermission: vi.fn().mockResolvedValue({
+  checkRolePermission: vi.fn().mockReturnValue({
     allowed: true,
   }),
 }))
@@ -212,6 +212,7 @@ describe('Socket Server Index Integration', () => {
         socketId,
         joinedAt: Date.now(),
         lastActivity: Date.now(),
+        role: 'admin',
       })
       room.activeConnections = 1
 
diff --git a/apps/sim/socket-server/middleware/permissions.ts b/apps/sim/socket-server/middleware/permissions.ts
diff --git a/apps/sim/socket-server/rooms/manager.ts b/apps/sim/socket-server/rooms/manager.ts
diff --git a/apps/sim/socket-server/validation/schemas.ts b/apps/sim/socket-server/validation/schemas.ts
