import {
  db,
  runOutsideTransactionContext,
  workflow,
  workflowDeploymentOperation,
  workflowDeploymentVersion,
} from '@sim/db'
import { credential } from '@sim/db/schema'
import { createLogger } from '@sim/logger'
import { getActiveWorkflowContext } from '@sim/platform-authz/workflow'
import { getErrorMessage } from '@sim/utils/errors'
import { generateId } from '@sim/utils/id'
import {
  loadWorkflowFromNormalizedTablesRaw,
  persistMigratedBlocks,
} from '@sim/workflow-persistence/load'
import { saveWorkflowToNormalizedTables as saveWorkflowToNormalizedTablesRaw } from '@sim/workflow-persistence/save'
import type { DbOrTx, NormalizedWorkflowData } from '@sim/workflow-persistence/types'
import type { BlockState, Loop, Parallel, WorkflowState } from '@sim/workflow-types/workflow'
import type { InferSelectModel } from 'drizzle-orm'
import { and, desc, eq, inArray, lt, sql } from 'drizzle-orm'
import { LRUCache } from 'lru-cache'
import type { Edge } from 'reactflow'
import { releaseWebhookPathClaims } from '@/lib/webhooks/path-claims'
import { remapConditionBlockIds, remapConditionEdgeHandle } from '@/lib/workflows/condition-ids'
import { isDynamicHandleSubblock } from '@/lib/workflows/dynamic-handle-topology'
import {
  backfillCanonicalModes,
  migrateSubblockIds,
} from '@/lib/workflows/migrations/subblock-migrations'
import { backfillWhatsAppInteractiveType } from '@/lib/workflows/migrations/whatsapp-interactive-type'
import { supersedeInFlightDeploymentOperations } from '@/lib/workflows/persistence/deployment-operations'
import { sanitizeAgentToolsInBlocks } from '@/lib/workflows/sanitization/validation'

const logger = createLogger('WorkflowDBHelpers')

export type { DbOrTx, NormalizedWorkflowData } from '@sim/workflow-persistence/types'

export type WorkflowDeploymentVersion = InferSelectModel<typeof workflowDeploymentVersion>

export class NoActiveDeploymentError extends Error {
  constructor(workflowId: string) {
    super(`Workflow ${workflowId} has no active deployment`)
    this.name = 'NoActiveDeploymentError'
  }
}

function hasReturnedRows(result: unknown): boolean {
  if (Array.isArray(result)) return result.length > 0

  if (result && typeof result === 'object') {
    const rows = 'rows' in result ? result.rows : undefined
    if (Array.isArray(rows)) return rows.length > 0
  }

  return Boolean(result)
}

async function lockWorkflowForUpdate(tx: DbOrTx, workflowId: string): Promise<boolean> {
  const query = tx.select({ id: workflow.id }).from(workflow).where(eq(workflow.id, workflowId))

  if ('limit' in query && typeof query.limit === 'function') {
    const limited = query.limit(1)
    const rows =
      'for' in limited && typeof limited.for === 'function'
        ? await limited.for('update')
        : await limited
    return hasReturnedRows(rows)
  }

  const rows = await query

  return hasReturnedRows(rows)
}

export interface WorkflowDeploymentVersionResponse {
  id: string
  version: number
  name?: string | null
  description?: string | null
  isActive: boolean
  createdAt: string
  createdBy?: string | null
  deployedBy?: string | null
  latestOperationStatus?: 'preparing' | 'activating' | 'active' | 'failed' | 'superseded' | null
}

export interface DeployedWorkflowData extends NormalizedWorkflowData {
  deploymentVersionId: string
  variables?: Record<string, unknown>
}

export async function blockExistsInDeployment(
  workflowId: string,
  blockId: string
): Promise<boolean> {
  try {
    const [result] = await db
      .select({ state: workflowDeploymentVersion.state })
      .from(workflowDeploymentVersion)
      .where(
        and(
          eq(workflowDeploymentVersion.workflowId, workflowId),
          eq(workflowDeploymentVersion.isActive, true)
        )
      )
      .limit(1)

    if (!result?.state) {
      return false
    }

    const state = result.state as WorkflowState
    return !!state.blocks?.[blockId]
  } catch (error) {
    logger.error(`Error checking block ${blockId} in deployment for workflow ${workflowId}:`, error)
    return false
  }
}

const DEPLOYED_STATE_CACHE_MAX_ENTRIES = 500
const DEPLOYED_STATE_CACHE_TTL_MS = 5 * 60 * 1000

/**
 * Caches post-migration deployed state by the immutable `deploymentVersionId`, so
 * a redeploy/rollback (which changes the active id) self-invalidates. The TTL is
 * absolute on purpose — it bounds the one non-immutable part, the live credential
 * remap in `applyBlockMigrations` — so credential changes still propagate.
 */
const deployedStateCache = new LRUCache<string, DeployedWorkflowData>({
  max: DEPLOYED_STATE_CACHE_MAX_ENTRIES,
  ttl: DEPLOYED_STATE_CACHE_TTL_MS,
})

/** Evicts one deployed-state entry, or clears the cache when no id is given. */
export function invalidateDeployedStateCache(deploymentVersionId?: string): void {
  if (deploymentVersionId) {
    deployedStateCache.delete(deploymentVersionId)
    return
  }
  deployedStateCache.clear()
}

interface DeploymentStateRow {
  id: string
  state: unknown
}

async function materializeDeploymentState(
  workflowId: string,
  version: DeploymentStateRow,
  providedWorkspaceId?: string
): Promise<DeployedWorkflowData> {
  const cached = deployedStateCache.get(version.id)
  if (cached) {
    return structuredClone(cached)
  }

  const state = version.state as WorkflowState & { variables?: Record<string, unknown> }
  let resolvedWorkspaceId = providedWorkspaceId
  if (!resolvedWorkspaceId) {
    const workflowContext = await getActiveWorkflowContext(workflowId)
    resolvedWorkspaceId = workflowContext?.workspaceId
  }

  if (!resolvedWorkspaceId) {
    throw new Error(`Workflow ${workflowId} has no workspace`)
  }

  const { blocks: migratedBlocks } = await applyBlockMigrations(
    state.blocks || {},
    resolvedWorkspaceId
  )
  const deployedState: DeployedWorkflowData = {
    blocks: migratedBlocks,
    edges: state.edges || [],
    loops: state.loops || {},
    parallels: state.parallels || {},
    variables: state.variables || {},
    isFromNormalizedTables: false,
    deploymentVersionId: version.id,
  }

  deployedStateCache.set(version.id, deployedState)
  return structuredClone(deployedState)
}

export async function loadDeployedWorkflowState(
  workflowId: string,
  providedWorkspaceId?: string
): Promise<DeployedWorkflowData> {
  try {
    const [active] = await db
      .select({
        id: workflowDeploymentVersion.id,
        state: workflowDeploymentVersion.state,
        createdAt: workflowDeploymentVersion.createdAt,
      })
      .from(workflowDeploymentVersion)
      .where(
        and(
          eq(workflowDeploymentVersion.workflowId, workflowId),
          eq(workflowDeploymentVersion.isActive, true)
        )
      )
      .orderBy(desc(workflowDeploymentVersion.createdAt))
      .limit(1)

    if (!active?.state) {
      throw new NoActiveDeploymentError(workflowId)
    }

    return materializeDeploymentState(workflowId, active, providedWorkspaceId)
  } catch (error) {
    logger.error(`Error loading deployed workflow state ${workflowId}:`, error)
    throw error
  }
}

/**
 * Loads an immutable deployment snapshot by ID for work admitted before a later cutover.
 */
export async function loadWorkflowDeploymentVersionState(
  workflowId: string,
  deploymentVersionId: string,
  providedWorkspaceId?: string
): Promise<DeployedWorkflowData> {
  const [version] = await db
    .select({
      id: workflowDeploymentVersion.id,
      state: workflowDeploymentVersion.state,
    })
    .from(workflowDeploymentVersion)
    .where(
      and(
        eq(workflowDeploymentVersion.workflowId, workflowId),
        eq(workflowDeploymentVersion.id, deploymentVersionId)
      )
    )
    .limit(1)

  if (!version?.state) {
    throw new Error(`Deployment ${deploymentVersionId} was not found for workflow ${workflowId}`)
  }

  return materializeDeploymentState(workflowId, version, providedWorkspaceId)
}

interface MigrationContext {
  blocks: Record<string, BlockState>
  workspaceId: string
  executor: DbOrTx
  migrated: boolean
}

type BlockMigration = (ctx: MigrationContext) => MigrationContext | Promise<MigrationContext>

function createMigrationPipeline(migrations: BlockMigration[]) {
  return async (
    blocks: Record<string, BlockState>,
    workspaceId: string,
    executor: DbOrTx = db
  ): Promise<{ blocks: Record<string, BlockState>; migrated: boolean }> => {
    let ctx: MigrationContext = { blocks, workspaceId, executor, migrated: false }
    for (const migration of migrations) {
      ctx = await migration(ctx)
    }
    return { blocks: ctx.blocks, migrated: ctx.migrated }
  }
}

const applyBlockMigrations = createMigrationPipeline([
  (ctx) => {
    const { blocks } = sanitizeAgentToolsInBlocks(ctx.blocks)
    return { ...ctx, blocks }
  },

  (ctx) => ({
    ...ctx,
    blocks: migrateAgentBlocksToMessagesFormat(ctx.blocks),
  }),

  (ctx) => {
    const { blocks, migrated } = migrateSubblockIds(ctx.blocks)
    return { ...ctx, blocks, migrated: ctx.migrated || migrated }
  },

  (ctx) => {
    const { blocks, migrated } = backfillWhatsAppInteractiveType(ctx.blocks)
    return { ...ctx, blocks, migrated: ctx.migrated || migrated }
  },

  async (ctx) => {
    const { blocks, migrated } = await migrateCredentialIds(
      ctx.blocks,
      ctx.workspaceId,
      ctx.executor
    )
    return { ...ctx, blocks, migrated: ctx.migrated || migrated }
  },

  (ctx) => {
    const { blocks, migrated } = backfillCanonicalModes(ctx.blocks)
    return { ...ctx, blocks, migrated: ctx.migrated || migrated }
  },
])

/**
 * Migrates agent blocks from old format (systemPrompt/userPrompt) to new format (messages array)
 */
export function migrateAgentBlocksToMessagesFormat(
  blocks: Record<string, BlockState>
): Record<string, BlockState> {
  return Object.fromEntries(
    Object.entries(blocks).map(([id, block]) => {
      if (block.type === 'agent') {
        const systemPrompt = block.subBlocks.systemPrompt?.value
        const userPrompt = block.subBlocks.userPrompt?.value
        const messages = block.subBlocks.messages?.value

        if ((systemPrompt || userPrompt) && !messages) {
          const newMessages: Array<{ role: string; content: string }> = []

          if (systemPrompt) {
            newMessages.push({
              role: 'system',
              content: typeof systemPrompt === 'string' ? systemPrompt : String(systemPrompt),
            })
          }

          if (userPrompt) {
            let userContent = userPrompt

            if (typeof userContent === 'object' && userContent !== null) {
              if ('input' in userContent) {
                userContent = (userContent as any).input
              } else {
                userContent = JSON.stringify(userContent)
              }
            }

            newMessages.push({
              role: 'user',
              content: String(userContent),
            })
          }

          return [
            id,
            {
              ...block,
              subBlocks: {
                ...block.subBlocks,
                messages: {
                  id: 'messages',
                  type: 'messages-input',
                  value: newMessages,
                },
              },
            },
          ]
        }
      }
      return [id, block]
    })
  )
}

export const CREDENTIAL_SUBBLOCK_IDS = new Set([
  'credential',
  'manualCredential',
  'triggerCredentials',
  'customBotCredential',
  'manualBotCredential',
])

async function migrateCredentialIds(
  blocks: Record<string, BlockState>,
  workspaceId: string,
  executor: DbOrTx
): Promise<{ blocks: Record<string, BlockState>; migrated: boolean }> {
  const potentialLegacyIds = new Set<string>()

  for (const block of Object.values(blocks)) {
    for (const [subBlockId, subBlock] of Object.entries(block.subBlocks || {})) {
      if (!subBlock || typeof subBlock !== 'object') continue
      const value = (subBlock as { value?: unknown }).value
      if (
        CREDENTIAL_SUBBLOCK_IDS.has(subBlockId) &&
        typeof value === 'string' &&
        value &&
        !value.startsWith('cred_')
      ) {
        potentialLegacyIds.add(value)
      }

      if (subBlockId === 'tools' && Array.isArray(value)) {
        for (const tool of value) {
          const credParam = tool?.params?.credential
          if (typeof credParam === 'string' && credParam && !credParam.startsWith('cred_')) {
            potentialLegacyIds.add(credParam)
          }
        }
      }
    }
  }

  if (potentialLegacyIds.size === 0) {
    return { blocks, migrated: false }
  }

  const rows = await executor
    .select({ id: credential.id, accountId: credential.accountId })
    .from(credential)
    .where(
      and(
        inArray(credential.accountId, [...potentialLegacyIds]),
        eq(credential.workspaceId, workspaceId)
      )
    )

  if (rows.length === 0) {
    return { blocks, migrated: false }
  }

  const accountToCredential = new Map(rows.map((r) => [r.accountId!, r.id]))

  const migratedBlocks = Object.fromEntries(
    Object.entries(blocks).map(([blockId, block]) => {
      let blockChanged = false
      const newSubBlocks = { ...block.subBlocks }

      for (const [subBlockId, subBlock] of Object.entries(newSubBlocks)) {
        if (CREDENTIAL_SUBBLOCK_IDS.has(subBlockId) && typeof subBlock.value === 'string') {
          const newId = accountToCredential.get(subBlock.value)
          if (newId) {
            newSubBlocks[subBlockId] = { ...subBlock, value: newId }
            blockChanged = true
          }
        }

        if (subBlockId === 'tools' && Array.isArray(subBlock.value)) {
          let toolsChanged = false
          const newTools = (subBlock.value as any[]).map((tool: any) => {
            const credParam = tool?.params?.credential
            if (typeof credParam === 'string') {
              const newId = accountToCredential.get(credParam)
              if (newId) {
                toolsChanged = true
                return { ...tool, params: { ...tool.params, credential: newId } }
              }
            }
            return tool
          })
          if (toolsChanged) {
            newSubBlocks[subBlockId] = { ...subBlock, value: newTools as any }
            blockChanged = true
          }
        }
      }

      return [blockId, blockChanged ? { ...block, subBlocks: newSubBlocks } : block]
    })
  )

  const anyBlockChanged = Object.keys(migratedBlocks).some(
    (id) => migratedBlocks[id] !== blocks[id]
  )

  return { blocks: migratedBlocks, migrated: anyBlockChanged }
}

/**
 * Load workflow from normalized tables and apply all block migrations
 * (credential ID rewrites, agent message migration, subblock ID migrations,
 * WhatsApp interactive-type backfill, canonical-mode backfill, tool
 * sanitization). Returns null if the workflow has not been migrated to
 * normalized tables yet.
 */
export async function loadWorkflowFromNormalizedTables(
  workflowId: string,
  externalTx?: DbOrTx
): Promise<NormalizedWorkflowData | null> {
  const raw = await loadWorkflowFromNormalizedTablesRaw(workflowId, externalTx)
  if (!raw) return null

  const { blocks: finalBlocks, migrated } = await applyBlockMigrations(
    raw.blocks,
    raw.workspaceId,
    externalTx ?? db
  )

  if (migrated) {
    // Deliberate fire-and-forget persistence on the global pool: it must not
    // join (or block) a read transaction this load may be running inside, so
    // it escapes the transaction context instead of tripping the wire.
    runOutsideTransactionContext(() => {
      Promise.resolve().then(() =>
        persistMigratedBlocks(workflowId, raw.blocks, finalBlocks, raw.blockUpdatedAtById)
      )
    })
  }

  const patchedLoops: Record<string, Loop> = { ...raw.loops }
  const patchedParallels: Record<string, Parallel> = { ...raw.parallels }

  for (const id of Object.keys(raw.loops)) {
    if (finalBlocks[id]) {
      patchedLoops[id] = { ...raw.loops[id], enabled: finalBlocks[id].enabled ?? true }
    }
  }
  for (const id of Object.keys(raw.parallels)) {
    if (finalBlocks[id]) {
      patchedParallels[id] = {
        ...raw.parallels[id],
        enabled: finalBlocks[id].enabled ?? true,
      }
    }
  }

  return {
    blocks: finalBlocks,
    edges: raw.edges,
    loops: patchedLoops,
    parallels: patchedParallels,
    isFromNormalizedTables: true,
  }
}

export async function loadWorkflowDeploymentSnapshot(
  workflowId: string,
  externalTx?: DbOrTx
): Promise<WorkflowState | null> {
  const loadSnapshot = async (tx: DbOrTx) => {
    const [normalizedData, [workflowRecord]] = await Promise.all([
      loadWorkflowFromNormalizedTables(workflowId, tx),
      tx
        .select({ variables: workflow.variables })
        .from(workflow)
        .where(eq(workflow.id, workflowId))
        .limit(1),
    ])

    if (!normalizedData) return null

    return buildWorkflowDeploymentSnapshot(normalizedData, workflowRecord?.variables)
  }

  if (externalTx) {
    return loadSnapshot(externalTx)
  }

  return db.transaction(async (tx) => {
    await tx.execute(sql`SET TRANSACTION ISOLATION LEVEL REPEATABLE READ`)
    return loadSnapshot(tx)
  })
}

export function buildWorkflowDeploymentSnapshot(
  normalizedData: NormalizedWorkflowData,
  variables: unknown
): WorkflowState {
  return {
    blocks: normalizedData.blocks,
    edges: normalizedData.edges,
    loops: normalizedData.loops,
    parallels: normalizedData.parallels,
    variables: (variables as WorkflowState['variables']) || {},
    lastSaved: Date.now(),
  }
}

export async function saveWorkflowToNormalizedTables(
  workflowId: string,
  state: WorkflowState,
  externalTx?: DbOrTx
): Promise<{ success: boolean; error?: string }> {
  if (externalTx) {
    return saveWorkflowToNormalizedTablesRaw(workflowId, state, externalTx)
  }

  try {
    return await db.transaction(async (tx) => {
      await lockWorkflowForUpdate(tx, workflowId)
      return saveWorkflowToNormalizedTablesRaw(workflowId, state, tx)
    })
  } catch (error) {
    const message = getErrorMessage(error, 'Failed to save workflow state')
    logger.error(`Error saving workflow ${workflowId} to normalized tables:`, error)
    return { success: false, error: message }
  }
}

export async function workflowExistsInNormalizedTables(workflowId: string): Promise<boolean> {
  try {
    const { workflowBlocks } = await import('@sim/db')
    const blocks = await db
      .select({ id: workflowBlocks.id })
      .from(workflowBlocks)
      .where(eq(workflowBlocks.workflowId, workflowId))
      .limit(1)

    return blocks.length > 0
  } catch (error) {
    logger.error(`Error checking if workflow ${workflowId} exists in normalized tables:`, error)
    return false
  }
}

/**
 * Update the name and/or description metadata of an existing deployment version.
 * Shared by the workflow deployment-version PATCH route and the copilot
 * `update_deployment_version` tool so both behave identically. Returns the
 * resulting name/description, or null if the version does not exist.
 */
export async function updateDeploymentVersionMetadata(params: {
  workflowId: string
  version: number
  name?: string | null
  description?: string | null
}): Promise<{ name: string | null; description: string | null } | null> {
  const updateData: { name?: string | null; description?: string | null } = {}
  if (params.name !== undefined) updateData.name = params.name
  if (params.description !== undefined) updateData.description = params.description

  if (Object.keys(updateData).length === 0) {
    const [row] = await db
      .select({
        name: workflowDeploymentVersion.name,
        description: workflowDeploymentVersion.description,
      })
      .from(workflowDeploymentVersion)
      .where(
        and(
          eq(workflowDeploymentVersion.workflowId, params.workflowId),
          eq(workflowDeploymentVersion.version, params.version)
        )
      )
      .limit(1)
    return row ?? null
  }

  const [updated] = await db
    .update(workflowDeploymentVersion)
    .set(updateData)
    .where(
      and(
        eq(workflowDeploymentVersion.workflowId, params.workflowId),
        eq(workflowDeploymentVersion.version, params.version)
      )
    )
    .returning({
      name: workflowDeploymentVersion.name,
      description: workflowDeploymentVersion.description,
    })
  return updated ?? null
}

export interface RegenerateStateInput {
  blocks?: Record<string, BlockState>
  edges?: Edge[]
  loops?: Record<string, Loop>
  parallels?: Record<string, Parallel>
  lastSaved?: number
  variables?: Record<string, unknown>
  metadata?: Record<string, unknown>
}

interface RegenerateStateOutput {
  blocks: Record<string, BlockState>
  edges: Edge[]
  loops: Record<string, Loop>
  parallels: Record<string, Parallel>
  lastSaved: number
  variables?: Record<string, unknown>
  metadata?: Record<string, unknown>
}

export function regenerateWorkflowStateIds(state: RegenerateStateInput): RegenerateStateOutput {
  const blockIdMapping = new Map<string, string>()
  const edgeIdMapping = new Map<string, string>()
  const loopIdMapping = new Map<string, string>()
  const parallelIdMapping = new Map<string, string>()

  Object.keys(state.blocks || {}).forEach((oldId) => {
    blockIdMapping.set(oldId, generateId())
  })

  ;(state.edges || []).forEach((edge: Edge) => {
    edgeIdMapping.set(edge.id, generateId())
  })

  Object.keys(state.loops || {}).forEach((oldId) => {
    loopIdMapping.set(oldId, generateId())
  })

  Object.keys(state.parallels || {}).forEach((oldId) => {
    parallelIdMapping.set(oldId, generateId())
  })

  const newBlocks: Record<string, BlockState> = {}
  const newEdges: Edge[] = []
  const newLoops: Record<string, Loop> = {}
  const newParallels: Record<string, Parallel> = {}

  Object.entries(state.blocks || {}).forEach(([oldId, block]) => {
    const newId = blockIdMapping.get(oldId)!
    const newBlock: BlockState = {
      ...block,
      id: newId,
      subBlocks: structuredClone(block.subBlocks),
      locked: false,
    }

    if (newBlock.data?.parentId) {
      const newParentId = blockIdMapping.get(newBlock.data.parentId)
      if (newParentId) {
        newBlock.data = { ...newBlock.data, parentId: newParentId }
      }
    }

    if (newBlock.subBlocks) {
      const updatedSubBlocks: Record<string, BlockState['subBlocks'][string]> = {}
      Object.entries(newBlock.subBlocks).forEach(([subId, subBlock]) => {
        const updatedSubBlock = { ...subBlock }

        if (
          typeof updatedSubBlock.value === 'string' &&
          blockIdMapping.has(updatedSubBlock.value)
        ) {
          updatedSubBlock.value = blockIdMapping.get(updatedSubBlock.value) ?? updatedSubBlock.value
        }

        if (
          isDynamicHandleSubblock(block.type, subId) &&
          typeof updatedSubBlock.value === 'string'
        ) {
          try {
            const parsed = JSON.parse(updatedSubBlock.value)
            if (Array.isArray(parsed) && remapConditionBlockIds(parsed, oldId, newId)) {
              updatedSubBlock.value = JSON.stringify(parsed)
            }
          } catch {}
        }

        updatedSubBlocks[subId] = updatedSubBlock
      })
      newBlock.subBlocks = updatedSubBlocks
    }

    newBlocks[newId] = newBlock
  })

  ;(state.edges || []).forEach((edge: Edge) => {
    const newId = edgeIdMapping.get(edge.id)!
    const newSource = blockIdMapping.get(edge.source) || edge.source
    const newTarget = blockIdMapping.get(edge.target) || edge.target
    const newSourceHandle =
      edge.sourceHandle && blockIdMapping.has(edge.source)
        ? remapConditionEdgeHandle(edge.sourceHandle, edge.source, newSource)
        : edge.sourceHandle

    newEdges.push({
      ...edge,
      id: newId,
      source: newSource,
      target: newTarget,
      sourceHandle: newSourceHandle,
    })
  })

  Object.entries(state.loops || {}).forEach(([oldId, loop]) => {
    const newId = loopIdMapping.get(oldId)!
    const newLoop: Loop = { ...loop, id: newId }

    if (newLoop.nodes) {
      newLoop.nodes = newLoop.nodes.map((nodeId: string) => blockIdMapping.get(nodeId) || nodeId)
    }

    newLoops[newId] = newLoop
  })

  Object.entries(state.parallels || {}).forEach(([oldId, parallel]) => {
    const newId = parallelIdMapping.get(oldId)!
    const newParallel: Parallel = { ...parallel, id: newId }

    if (newParallel.nodes) {
      newParallel.nodes = newParallel.nodes.map(
        (nodeId: string) => blockIdMapping.get(nodeId) || nodeId
      )
    }

    newParallels[newId] = newParallel
  })

  return {
    blocks: newBlocks,
    edges: newEdges,
    loops: newLoops,
    parallels: newParallels,
    lastSaved: state.lastSaved || Date.now(),
    ...(state.variables && { variables: state.variables }),
    ...(state.metadata && { metadata: state.metadata }),
  }
}

export async function undeployWorkflow(params: {
  workflowId: string
  tx?: DbOrTx
  onUndeployTransaction?: (tx: DbOrTx, result: { deploymentVersionIds: string[] }) => Promise<void>
}): Promise<{
  success: boolean
  error?: string
}> {
  const { workflowId, tx } = params

  const executeUndeploy = async (dbCtx: DbOrTx) => {
    if (!(await lockWorkflowForUpdate(dbCtx, workflowId))) {
      throw new Error('Workflow not found')
    }

    const deploymentVersions = await dbCtx
      .select({ id: workflowDeploymentVersion.id })
      .from(workflowDeploymentVersion)
      .where(eq(workflowDeploymentVersion.workflowId, workflowId))
    const deploymentVersionIds = deploymentVersions.map((version) => version.id)

    await supersedeInFlightDeploymentOperations(dbCtx, workflowId)
    const { deleteSchedulesForWorkflow } = await import('@/lib/workflows/schedules/deploy')
    await deleteSchedulesForWorkflow(workflowId, dbCtx)
    await releaseWebhookPathClaims(dbCtx, workflowId)

    await dbCtx
      .update(workflowDeploymentVersion)
      .set({ isActive: false })
      .where(eq(workflowDeploymentVersion.workflowId, workflowId))

    await dbCtx
      .update(workflow)
      .set({ isDeployed: false, deployedAt: null })
      .where(eq(workflow.id, workflowId))

    await params.onUndeployTransaction?.(dbCtx, { deploymentVersionIds })
  }

  try {
    if (tx) {
      await executeUndeploy(tx)
    } else {
      await db.transaction(async (txn) => {
        await executeUndeploy(txn)
      })
    }

    logger.info(`Undeployed workflow ${workflowId}`)
    return { success: true }
  } catch (error) {
    logger.error(`Error undeploying workflow ${workflowId}:`, error)
    return {
      success: false,
      error: getErrorMessage(error, 'Failed to undeploy workflow'),
    }
  }
}

/**
 * Resolves the deployment version that precedes the currently active one —
 * the default rollback target when no explicit version is given.
 */
export async function findPreviousDeploymentVersion(
  workflowId: string
): Promise<
  { ok: true; version: number } | { ok: false; reason: 'no_active_version' | 'no_previous_version' }
> {
  const [activeRow] = await db
    .select({ version: workflowDeploymentVersion.version })
    .from(workflowDeploymentVersion)
    .where(
      and(
        eq(workflowDeploymentVersion.workflowId, workflowId),
        eq(workflowDeploymentVersion.isActive, true)
      )
    )
    .limit(1)

  if (!activeRow) {
    return { ok: false, reason: 'no_active_version' }
  }

  const [previousRow] = await db
    .select({ version: workflowDeploymentVersion.version })
    .from(workflowDeploymentVersion)
    .where(
      and(
        eq(workflowDeploymentVersion.workflowId, workflowId),
        lt(workflowDeploymentVersion.version, activeRow.version)
      )
    )
    .orderBy(desc(workflowDeploymentVersion.version))
    .limit(1)

  if (!previousRow) {
    return { ok: false, reason: 'no_previous_version' }
  }

  return { ok: true, version: previousRow.version }
}

/**
 * Fetches a single deployment version of a workflow, including its state
 * snapshot. Returns null when the version does not exist.
 */
export async function getWorkflowDeploymentVersion(
  workflowId: string,
  version: number
): Promise<{
  id: string
  version: number
  name: string | null
  description: string | null
  isActive: boolean
  createdAt: Date
  state: unknown
} | null> {
  const [row] = await db
    .select({
      id: workflowDeploymentVersion.id,
      version: workflowDeploymentVersion.version,
      name: workflowDeploymentVersion.name,
      description: workflowDeploymentVersion.description,
      isActive: workflowDeploymentVersion.isActive,
      createdAt: workflowDeploymentVersion.createdAt,
      state: workflowDeploymentVersion.state,
    })
    .from(workflowDeploymentVersion)
    .where(
      and(
        eq(workflowDeploymentVersion.workflowId, workflowId),
        eq(workflowDeploymentVersion.version, version)
      )
    )
    .limit(1)

  return row ?? null
}

export async function listWorkflowVersions(workflowId: string): Promise<{
  versions: Array<{
    id: string
    version: number
    name: string | null
    description: string | null
    isActive: boolean
    createdAt: Date
    createdBy: string | null
    deployedByName: string | null
    latestOperationStatus: string | null
  }>
}> {
  const { user } = await import('@sim/db')

  const [rows, [currentOperation]] = await Promise.all([
    db
      .select({
        id: workflowDeploymentVersion.id,
        version: workflowDeploymentVersion.version,
        name: workflowDeploymentVersion.name,
        description: workflowDeploymentVersion.description,
        isActive: workflowDeploymentVersion.isActive,
        createdAt: workflowDeploymentVersion.createdAt,
        createdBy: workflowDeploymentVersion.createdBy,
        deployedByName: user.name,
      })
      .from(workflowDeploymentVersion)
      .leftJoin(user, eq(workflowDeploymentVersion.createdBy, user.id))
      .where(eq(workflowDeploymentVersion.workflowId, workflowId))
      .orderBy(desc(workflowDeploymentVersion.version)),
    /**
     * Only the workflow's current (latest-generation) operation carries a
     * status marker: a failed or in-flight attempt is live information until
     * the next deploy action supersedes it, at which point it is history and
     * the marker clears rather than sticking to old versions forever.
     */
    db
      .select({
        deploymentVersionId: workflowDeploymentOperation.deploymentVersionId,
        status: workflowDeploymentOperation.status,
      })
      .from(workflowDeploymentOperation)
      .where(eq(workflowDeploymentOperation.workflowId, workflowId))
      .orderBy(desc(workflowDeploymentOperation.generation))
      .limit(1),
  ])

  return {
    versions: rows.map((row) => ({
      ...row,
      deployedByName: row.deployedByName ?? (row.createdBy === 'admin-api' ? 'Admin' : null),
      latestOperationStatus:
        currentOperation && currentOperation.deploymentVersionId === row.id
          ? currentOperation.status
          : null,
    })),
  }
}