feat(workflows): lock/duplicate improvements for workflows

apps/realtime/src/handlers/operations.ts

@@ -10,6 +10,7 @@ import {
 } from '@sim/realtime-protocol/constants'
 import { WorkflowOperationSchema } from '@sim/realtime-protocol/schemas'
 import { generateId } from '@sim/utils/id'
+import { assertWorkflowMutable, WorkflowLockedError } from '@sim/workflow-authz'
 import { ZodError } from 'zod'
 import { persistWorkflowOperation } from '@/database/operations'
 import type { AuthenticatedSocket } from '@/middleware/auth'
@@ -139,6 +140,24 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
         }
       }
 
+      try {
+        await assertWorkflowMutable(workflowId)
+      } catch (error) {
+        if (error instanceof WorkflowLockedError) {
+          emitOperationError(
+            {
+              type: 'WORKFLOW_LOCKED',
+              message: error.message,
+              operation,
+              target,
+            },
+            { error: error.message, retryable: false }
+          )
+          return
+        }
+        throw error
+      }
+
       // Broadcast first for position updates to minimize latency, then persist
       // For other operations, persist first for consistency
       if (isPositionUpdate) {

apps/realtime/src/handlers/subblocks.ts

@@ -2,6 +2,7 @@ import { db } from '@sim/db'
 import { workflow, workflowBlocks } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { SUBBLOCK_OPERATIONS } from '@sim/realtime-protocol/constants'
+import { assertWorkflowMutable, WorkflowLockedError } from '@sim/workflow-authz'
 import { and, eq } from 'drizzle-orm'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import { checkRolePermission } from '@/middleware/permissions'
@@ -151,6 +152,28 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
         return
       }
 
+      try {
+        await assertWorkflowMutable(workflowId)
+      } catch (error) {
+        if (error instanceof WorkflowLockedError) {
+          socket.emit('operation-forbidden', {
+            type: 'WORKFLOW_LOCKED',
+            message: error.message,
+            operation: SUBBLOCK_OPERATIONS.UPDATE,
+            target: 'subblock',
+          })
+          if (operationId) {
+            socket.emit('operation-failed', {
+              operationId,
+              error: error.message,
+              retryable: false,
+            })
+          }
+          return
+        }
+        throw error
+      }
+
       // Update user activity
       await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
 
@@ -231,6 +254,22 @@ async function flushSubblockUpdate(
       return
     }
 
+    try {
+      await assertWorkflowMutable(workflowId)
+    } catch (error) {
+      if (error instanceof WorkflowLockedError) {
+        pending.opToSocket.forEach((socketId, opId) => {
+          io.to(socketId).emit('operation-failed', {
+            operationId: opId,
+            error: error.message,
+            retryable: false,
+          })
+        })
+        return
+      }
+      throw error
+    }
+
     let updateSuccessful = false
     let blockLocked = false
     await db.transaction(async (tx) => {

apps/realtime/src/handlers/variables.ts

@@ -2,6 +2,7 @@ import { db } from '@sim/db'
 import { workflow } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { VARIABLE_OPERATIONS } from '@sim/realtime-protocol/constants'
+import { assertWorkflowMutable, WorkflowLockedError } from '@sim/workflow-authz'
 import { eq } from 'drizzle-orm'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import { checkRolePermission } from '@/middleware/permissions'
@@ -140,6 +141,28 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager:
         return
       }
 
+      try {
+        await assertWorkflowMutable(workflowId)
+      } catch (error) {
+        if (error instanceof WorkflowLockedError) {
+          socket.emit('operation-forbidden', {
+            type: 'WORKFLOW_LOCKED',
+            message: error.message,
+            operation: VARIABLE_OPERATIONS.UPDATE,
+            target: 'variable',
+          })
+          if (operationId) {
+            socket.emit('operation-failed', {
+              operationId,
+              error: error.message,
+              retryable: false,
+            })
+          }
+          return
+        }
+        throw error
+      }
+
       // Update user activity
       await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
 
@@ -218,6 +241,22 @@ async function flushVariableUpdate(
       return
     }
 
+    try {
+      await assertWorkflowMutable(workflowId)
+    } catch (error) {
+      if (error instanceof WorkflowLockedError) {
+        pending.opToSocket.forEach((socketId, opId) => {
+          io.to(socketId).emit('operation-failed', {
+            operationId: opId,
+            error: error.message,
+            retryable: false,
+          })
+        })
+        return
+      }
+      throw error
+    }
+
     let updateSuccessful = false
     await db.transaction(async (tx) => {
       const [workflowRecord] = await tx