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

This commit is contained in:
Vikhyath Mondreti
2025-10-07 17:31:02 -07:00
committed by GitHub
parent 715f42c1a6
commit 8e43774b5e
11 changed files with 162 additions and 123 deletions
@@ -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 {
+20 -5
View File
@@ -600,22 +600,37 @@ 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) {
@@ -623,7 +638,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
pendingPositionUpdates.current.delete(blockId)
}
positionUpdateTimeouts.current.delete(blockId)
}, 33) // 30fps - good balance between smoothness and DB performance
}, 33)
positionUpdateTimeouts.current.set(blockId, timeoutId)
}
+11 -5
View File
@@ -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(
+2
View File
@@ -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,
},
@@ -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({
+65 -46
View File
@@ -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,16 +110,29 @@ 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,
@@ -111,19 +140,9 @@ export function setupOperationsHandlers(
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)) {
@@ -46,6 +46,7 @@ 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) {
@@ -53,6 +54,7 @@ export function setupWorkflowHandlers(
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)
+3 -2
View File
@@ -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
@@ -6,6 +6,56 @@ import { getUserEntityPermissions } from '@/lib/permissions/utils'
const logger = createLogger('SocketPermissions')
// Define operation permissions based on role
const ROLE_PERMISSIONS: Record<string, string[]> = {
admin: [
'add',
'remove',
'update',
'update-position',
'update-name',
'toggle-enabled',
'update-parent',
'update-wide',
'update-advanced-mode',
'update-trigger-mode',
'toggle-handles',
'duplicate',
],
write: [
'add',
'remove',
'update',
'update-position',
'update-name',
'toggle-enabled',
'update-parent',
'update-wide',
'update-advanced-mode',
'update-trigger-mode',
'toggle-handles',
'duplicate',
],
read: ['update-position'],
}
// Check if a role allows a specific operation (no DB query, pure logic)
export function checkRolePermission(
role: string,
operation: string
): { allowed: boolean; reason?: string } {
const allowedOperations = ROLE_PERMISSIONS[role] || []
if (!allowedOperations.includes(operation)) {
return {
allowed: false,
reason: `Role '${role}' not permitted to perform '${operation}'`,
}
}
return { allowed: true }
}
export async function verifyWorkspaceMembership(
userId: string,
workspaceId: string
@@ -75,66 +125,3 @@ export async function verifyWorkflowAccess(
return { hasAccess: false }
}
}
// Enhanced authorization for specific operations
export async function verifyOperationPermission(
userId: string,
workflowId: string,
operation: string,
target: string
): Promise<{ allowed: boolean; reason?: string }> {
try {
const accessInfo = await verifyWorkflowAccess(userId, workflowId)
if (!accessInfo.hasAccess) {
return { allowed: false, reason: 'No access to workflow' }
}
// Define operation permissions based on role
const rolePermissions = {
admin: [
'add',
'remove',
'update',
'update-position',
'update-name',
'toggle-enabled',
'update-parent',
'update-wide',
'update-advanced-mode',
'update-trigger-mode',
'toggle-handles',
'duplicate',
],
write: [
'add',
'remove',
'update',
'update-position',
'update-name',
'toggle-enabled',
'update-parent',
'update-wide',
'update-advanced-mode',
'update-trigger-mode',
'toggle-handles',
'duplicate',
],
read: ['update-position'], // Read-only users can only move things around
}
const allowedOperations = rolePermissions[accessInfo.role as keyof typeof rolePermissions] || []
if (!allowedOperations.includes(operation)) {
return {
allowed: false,
reason: `Role '${accessInfo.role}' not permitted to perform '${operation}' on '${target}'`,
}
}
return { allowed: true }
} catch (error) {
logger.error(`Error verifying operation permission:`, error)
return { allowed: false, reason: 'Permission check failed' }
}
}
+1
View File
@@ -28,6 +28,7 @@ export interface UserPresence {
socketId: string
joinedAt: number
lastActivity: number
role: string
cursor?: { x: number; y: number }
selection?: { type: 'block' | 'edge' | 'none'; id?: string }
}
@@ -36,6 +36,7 @@ export const BlockOperationSchema = z.object({
type: z.string().optional(),
name: z.string().optional(),
position: PositionSchema.optional(),
commit: z.boolean().optional(),
data: z.record(z.any()).optional(),
subBlocks: z.record(z.any()).optional(),
outputs: z.record(z.any()).optional(),