import { tmpdir } from 'os'
import { prisma } from '@/lib/prisma'
import { processVideoTranscode } from '@/lib/video-processor'
import { transcodeQueue } from '@/lib/video-transcode-queue'

const DEFAULT_MAX_ATTEMPTS = 3
const DEFAULT_LOCK_TIMEOUT_MS = 30 * 60 * 1000
const DEFAULT_RETRY_DELAY_MS = 5 * 1000
const MAX_RETRY_DELAY_MS = 30 * 1000
const RECOVERY_BATCH_SIZE = 20
type VideoTranscodeJobDelegate = {
  upsert: (...args: any[]) => Promise<any>
  findMany: (...args: any[]) => Promise<any[]>
  update: (...args: any[]) => Promise<any>
}

function getVideoTranscodeJobDelegate(): VideoTranscodeJobDelegate | null {
  const delegate = (prisma as any).videoTranscodeJob
  if (
    delegate &&
    typeof delegate.upsert === 'function' &&
    typeof delegate.findMany === 'function' &&
    typeof delegate.update === 'function'
  ) {
    return delegate
  }
  return null
}

function requireVideoTranscodeJobDelegate(): VideoTranscodeJobDelegate {
  const delegate = getVideoTranscodeJobDelegate()
  if (!delegate) {
    throw new Error('Prisma Client 缺少 videoTranscodeJob，请先执行数据库迁移并重新生成 Prisma Client')
  }
  return delegate
}

function getMaxAttempts() {
  const retries = parseInt(process.env.VIDEO_TRANSCODE_MAX_RETRIES || '2', 10)
  return Math.max(1, retries + 1)
}

function getLockTimeoutMs() {
  return parseInt(
    process.env.VIDEO_TRANSCODE_LOCK_TIMEOUT_MS || String(DEFAULT_LOCK_TIMEOUT_MS),
    10
  )
}

function getRetryDelayMs(attempt: number) {
  const baseDelay = parseInt(
    process.env.VIDEO_TRANSCODE_RETRY_BASE_DELAY_MS || String(DEFAULT_RETRY_DELAY_MS),
    10
  )
  const safeBaseDelay = Number.isFinite(baseDelay) && baseDelay > 0
    ? baseDelay
    : DEFAULT_RETRY_DELAY_MS

  return Math.min(safeBaseDelay * Math.pow(2, Math.max(0, attempt - 1)), MAX_RETRY_DELAY_MS)
}

function getErrorMessage(error: unknown) {
  if (error instanceof Error) return error.message
  if (typeof error === 'string') return error
  return '未知错误'
}

function isVideoTranscodeSchemaMissing(error: unknown) {
  const code = (error as any)?.code
  const table = String((error as any)?.meta?.table || '')
  const message = getErrorMessage(error)

  return (
    (code === 'P2021' && (table.includes('video_transcode_jobs') || message.includes('video_transcode_jobs'))) ||
    (code === 'P2022' && message.includes('video_transcode_jobs')) ||
    (code === 'P2010' && message.includes('video_transcode_jobs'))
  )
}

async function runVideoTranscodeTask(
  materialId: string,
  videoSourceUrl: string,
  needsCoverImage: boolean
) {
  const maxAttempts = getMaxAttempts()
  const videoTranscodeJob = requireVideoTranscodeJobDelegate()

  const job = await videoTranscodeJob.update({
    where: { materialId },
    data: {
      status: 'processing',
      lockedAt: new Date(),
      startedAt: new Date(),
      finishedAt: null,
      errorMessage: null,
      attempts: { increment: 1 },
      maxAttempts,
    },
  })

  try {
    console.log(`[Transcode] Starting persistent task for material ${materialId} (attempt ${job.attempts}/${maxAttempts})`)

    await prisma.material.update({
      where: { id: materialId },
      data: { videoProcessingStatus: 'processing' },
    })

    const result = await processVideoTranscode({
      inputPath: videoSourceUrl,
      outputDir: tmpdir(),
      materialId,
      needsCoverImage,
    })

    const updateData: any = {
      videoProcessingStatus: 'done',
      videoPreview540Url: result.preview540Url,
      videoPreviewClipUrl: result.clipUrl,
      videoDuration: result.duration,
    }

    if (needsCoverImage && result.coverUrl) {
      updateData.videoCoverUrl = result.coverUrl
      const material = await prisma.material.findUnique({
        where: { id: materialId },
        select: { previewUrl: true },
      })
      if (material?.previewUrl?.startsWith('data:image')) {
        updateData.previewUrl = result.coverUrl
      }
    }

    await prisma.material.update({
      where: { id: materialId },
      data: updateData,
    })

    await videoTranscodeJob.update({
      where: { materialId },
      data: {
        status: 'done',
        lockedAt: null,
        finishedAt: new Date(),
        errorMessage: null,
      },
    })

    console.log(`[Transcode] Persistent task completed for material ${materialId}`)
  } catch (error) {
    const errorMessage = getErrorMessage(error)
    const nextAttempts = Number(job.attempts || 0)
    const shouldRetry = nextAttempts < maxAttempts

    console.error(`[Transcode] Persistent task failed for material ${materialId}:`, error)

    await videoTranscodeJob.update({
      where: { materialId },
      data: {
        status: shouldRetry ? 'queued' : 'failed',
        lockedAt: null,
        finishedAt: shouldRetry ? null : new Date(),
        errorMessage,
      },
    })

    await prisma.material.update({
      where: { id: materialId },
      data: {
        videoProcessingStatus: shouldRetry ? 'pending' : 'failed',
      },
    })

    if (shouldRetry) {
      const retryDelay = getRetryDelayMs(nextAttempts)
      console.warn(`[Transcode] Task re-queued for material ${materialId} (${nextAttempts}/${maxAttempts}), retrying in ${retryDelay}ms`)
      setTimeout(() => {
        transcodeQueue.add(materialId, videoSourceUrl, needsCoverImage, runVideoTranscodeTask).catch(
          (retryError) => {
            console.error(`[Transcode] Retried task failed for material ${materialId}:`, retryError)
          }
        )
      }, retryDelay)
    }

    throw error
  }
}

export async function enqueueVideoTranscodeJob(params: {
  materialId: string
  videoSourceUrl: string
  needsCoverImage: boolean
  forceRetry?: boolean
}) {
  const { materialId, videoSourceUrl, needsCoverImage, forceRetry = false } = params
  const maxAttempts = getMaxAttempts()
  const videoTranscodeJob = requireVideoTranscodeJobDelegate()

  await videoTranscodeJob.upsert({
    where: { materialId },
    create: {
      materialId,
      videoSourceUrl,
      needsCoverImage,
      status: 'queued',
      attempts: 0,
      maxAttempts,
      errorMessage: null,
    },
    update: {
      videoSourceUrl,
      needsCoverImage,
      status: 'queued',
      lockedAt: null,
      startedAt: forceRetry ? null : undefined,
      finishedAt: null,
      errorMessage: null,
      attempts: forceRetry ? 0 : undefined,
      maxAttempts,
    },
  })

  await prisma.material.update({
    where: { id: materialId },
    data: { videoProcessingStatus: 'pending' },
  })

  transcodeQueue.add(materialId, videoSourceUrl, needsCoverImage, runVideoTranscodeTask).catch(
    (error) => {
      console.error(`[Transcode] Queued task failed for material ${materialId}:`, error)
    }
  )
}

export async function recoverStaleVideoTranscodeJobs() {
  const staleBefore = new Date(Date.now() - getLockTimeoutMs())
  const videoTranscodeJob = getVideoTranscodeJobDelegate()
  if (!videoTranscodeJob) {
    console.warn('[Transcode] 跳过启动恢复：Prisma Client 缺少 videoTranscodeJob，请执行数据库迁移并重新生成 Prisma Client')
    return { recovered: 0, hasMore: false, skipped: true }
  }

  let staleJobs: any[] = []
  let orphanMaterials: Array<{
    id: string
    videoSourceUrl: string | null
    videoCoverUrl: string | null
  }> = []

  try {
    staleJobs = await videoTranscodeJob.findMany({
      where: {
        OR: [
          { status: 'queued' },
          { status: 'processing', lockedAt: null },
          { status: 'processing', lockedAt: { lt: staleBefore } },
        ],
        material: {
          videoSourceUrl: { not: null },
          videoProcessingStatus: { in: ['pending', 'processing'] },
        },
      },
      include: {
        material: {
          select: {
            id: true,
            videoSourceUrl: true,
            videoCoverUrl: true,
          },
        },
      },
      take: RECOVERY_BATCH_SIZE,
      orderBy: { updatedAt: 'asc' },
    })

    orphanMaterials = await prisma.$queryRaw<Array<{
      id: string
      videoSourceUrl: string | null
      videoCoverUrl: string | null
    }>>`
      SELECT m.id, m."videoSourceUrl", m."videoCoverUrl"
      FROM "materials" m
      LEFT JOIN "video_transcode_jobs" j ON j."materialId" = m.id
      WHERE (m."materialKind" = 'VIDEO' OR m."materialType" = '视频')
        AND m."videoSourceUrl" IS NOT NULL
        AND m."videoProcessingStatus" IN ('pending', 'processing')
        AND j.id IS NULL
      ORDER BY m."updatedAt" ASC
      LIMIT ${RECOVERY_BATCH_SIZE}
    `
  } catch (error) {
    if (isVideoTranscodeSchemaMissing(error)) {
      console.warn('[Transcode] 跳过启动恢复：数据库缺少 video_transcode_jobs 表，请执行数据库迁移')
      return { recovered: 0, hasMore: false, skipped: true }
    }

    throw error
  }

  let recovered = 0

  for (const job of staleJobs) {
    const sourceUrl = job.material?.videoSourceUrl || job.videoSourceUrl
    if (!sourceUrl) continue

    await videoTranscodeJob.update({
      where: { materialId: job.materialId },
      data: {
        status: 'queued',
        lockedAt: null,
        errorMessage: null,
      },
    })
    await prisma.material.update({
      where: { id: job.materialId },
      data: { videoProcessingStatus: 'pending' },
    })
    await enqueueVideoTranscodeJob({
      materialId: job.materialId,
      videoSourceUrl: sourceUrl,
      needsCoverImage: !job.material?.videoCoverUrl,
    })
    recovered++
  }

  for (const material of orphanMaterials) {
    if (!material.videoSourceUrl) continue

    await enqueueVideoTranscodeJob({
      materialId: material.id,
      videoSourceUrl: material.videoSourceUrl,
      needsCoverImage: !material.videoCoverUrl,
    })
    recovered++
  }

  if (recovered > 0) {
    console.log(`[Transcode] Recovered ${recovered} queued/stale video transcode job(s)`)
  }

  const hasMore = staleJobs.length >= RECOVERY_BATCH_SIZE || orphanMaterials.length >= RECOVERY_BATCH_SIZE
  if (hasMore) {
    setTimeout(() => {
      recoverStaleVideoTranscodeJobs().catch((error) => {
        console.error('[Transcode] Continued recovery failed:', error)
      })
    }, 1000)
  }

  return { recovered, hasMore }
}