Function persistWorkflowOperation

apps/realtime/src/database/operations.ts:213–295

(workflowId: string, operation: any)

}

export async function persistWorkflowOperation(workflowId: string, operation: any) {
  const startTime = Date.now()
  try {
    const { operation: op, target, payload, timestamp, userId } = operation

    const activeWorkflow = await getActiveWorkflowContext(workflowId)
    if (!activeWorkflow) {
      throw new Error(`Workflow ${workflowId} is archived or unavailable`)
    }

    if (op === BLOCK_OPERATIONS.UPDATE_POSITION && randomFloat() < 0.01) {
      logger.debug('Socket DB operation sample:', {
        operation: op,
        target,
        workflowId: `${workflowId.substring(0, 8)}...`,
      })
    }

    await db.transaction(async (tx) => {
      await tx
        .update(workflow)
        .set({ updatedAt: new Date(timestamp) })
        .where(eq(workflow.id, workflowId))

      switch (target) {
        case OPERATION_TARGETS.BLOCK:
          await handleBlockOperationTx(tx, workflowId, op, payload)
          break
        case OPERATION_TARGETS.BLOCKS:
          await handleBlocksOperationTx(tx, workflowId, op, payload)
          break
        case OPERATION_TARGETS.EDGE:
          await handleEdgeOperationTx(tx, workflowId, op, payload)
          break
        case OPERATION_TARGETS.EDGES:
          await handleEdgesOperationTx(tx, workflowId, op, payload)
          break
        case OPERATION_TARGETS.SUBFLOW:
          await handleSubflowOperationTx(tx, workflowId, op, payload)
          break
        case OPERATION_TARGETS.SUBBLOCK:
          await handleSubblockOperationTx(tx, workflowId, op, payload)
          break
        case OPERATION_TARGETS.VARIABLE:
          await handleVariableOperationTx(tx, workflowId, op, payload)
          break
        case OPERATION_TARGETS.WORKFLOW:
          await handleWorkflowOperationTx(tx, workflowId, op, payload)
          break
        default:
          throw new Error(`Unknown operation target: ${target}`)
      }
    })

    // Audit workflow-level lock/unlock operations
    if (
      target === OPERATION_TARGETS.BLOCKS &&
      op === BLOCKS_OPERATIONS.BATCH_TOGGLE_LOCKED &&