Function flushSubblockUpdate

apps/realtime/src/handlers/subblocks.ts:237–374

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

}

async function flushSubblockUpdate(
  workflowId: string,
  pending: PendingSubblock,
  roomManager: IRoomManager
) {
  const { blockId, subblockId, value, timestamp } = pending.latest
  const io = roomManager.io

  try {
    // Verify workflow still exists
    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
    let blockLocked = false
    await db.transaction(async (tx) => {
      const allBlocks = await tx
        .select({
          id: workflowBlocks.id,
          subBlocks: workflowBlocks.subBlocks,
          locked: workflowBlocks.locked,
          data: workflowBlocks.data,
        })
        .from(workflowBlocks)
        .where(eq(workflowBlocks.workflowId, workflowId))

      type SubblockUpdateBlockRecord = (typeof allBlocks)[number]
      const blocksById: Record<string, SubblockUpdateBlockRecord> = Object.fromEntries(