flnxi-Wu/lib/video-transcode-service.ts

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

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 '未知错误'
}

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

    const job = await (prisma as any).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 (prisma as any).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 (prisma as any).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()

  await (prisma as any).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 staleJobs = await (prisma as any).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' },
  })

  const 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}
  `

  let recovered = 0

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

    await (prisma as any).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 }
}