Function flushVariableUpdate

apps/realtime/src/handlers/variables.ts:224–340

(
  workflowId: string,
  pending: PendingVariable,
  roomManager: IRoomManager
)

}

async function flushVariableUpdate(
  workflowId: string,
  pending: PendingVariable,
  roomManager: IRoomManager
) {
  const { variableId, field, value, timestamp } = pending.latest
  const io = roomManager.io

  try {
    const workflowExists = await db
      .select({ id: workflow.id })
      .from(workflow)
      .where(eq(workflow.id, workflowId))
      .limit(1)

    if (workflowExists.length === 0) {
      pending.opToSocket.forEach((socketId, opId) => {
        io.to(socketId).emit('operation-failed', {
          operationId: opId,
          error: 'Workflow not found',
          retryable: true,
        })
      })
      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
        .select({ variables: workflow.variables })
        .from(workflow)
        .where(eq(workflow.id, workflowId))
        .limit(1)

      if (!workflowRecord) {
        return
      }

      const variables = (workflowRecord.variables as any) || {}
      if (!variables[variableId]) {
        return
      }