gongxiangp2/lib/video-transcode-task.ts

// 视频转码任务封装，供上传接口和转码API复用
import { prisma } from '@/lib/prisma'
import { transcodeQueue } from '@/lib/video-transcode-queue'
import { processVideoTranscode } from '@/lib/video-processor'

const MAX_TRANSCODE_RETRIES = 3

class NonRetryableTranscodeError extends Error {
  constructor(message: string) {
    super(message)
    this.name = 'NonRetryableTranscodeError'
  }
}

function extractErrorMessage(error: unknown): string {
  if (error instanceof Error) {
    return error.message
  }

  if (typeof error === 'string') {
    return error
  }

  try {
    return JSON.stringify(error)
  } catch {
    return String(error)
  }
}

function isNonRetryableTranscodeError(error: unknown): boolean {
  if (error instanceof NonRetryableTranscodeError) {
    return true
  }

  const code = (error as { code?: string } | undefined)?.code
  if (code && code.toUpperCase() === 'ENOENT') {
    return true
  }

  const message = extractErrorMessage(error)
  return message.includes('视频文件不存在') || message.includes('ENOENT')
}

/**
 * 实际的转码任务处理函数
 * 
 * 内部包含最多 3 次重试逻辑，队列外部无需处理 retry。
 */
export async function executeTranscodeTask(
  materialId: string,
  fallbackVideoSourceUrl: string,
  needsCoverImage: boolean
): Promise<void> {
  let attempt = 0

  while (attempt < MAX_TRANSCODE_RETRIES) {
    attempt += 1

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

      const latestMaterial = await prisma.material.findUnique({
        where: { id: materialId },
        select: {
          videoSourceUrl: true,
          videoCoverUrl: true,
          previewUrl: true,
        },
      })

      const videoSourceUrl = latestMaterial?.videoSourceUrl || fallbackVideoSourceUrl
      if (!videoSourceUrl) {
        throw new NonRetryableTranscodeError('视频文件不存在')
      }

      const coverStillNeeded = needsCoverImage || !latestMaterial?.videoCoverUrl

      const fileExists = await validateVideoFile(videoSourceUrl)
      if (!fileExists) {
        throw new NonRetryableTranscodeError(`视频文件不存在: ${videoSourceUrl}`)
      }

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

      const result = await processVideoTranscode({
        inputPath: videoSourceUrl,
        outputDir: './uploads/transcoded',
        materialId,
        needsCoverImage: coverStillNeeded,
        onProgress: (progress, stage) => {
          console.log(`[Transcode] Material ${materialId} - ${stage}: ${progress}%`)
          transcodeQueue.updateProgress(materialId, progress, stage)
        },
      })

      const nextPreviewUrl = latestMaterial?.previewUrl && latestMaterial.previewUrl.trim() !== ''
        ? latestMaterial.previewUrl
        : result.coverUrl || null

      await prisma.material.update({
        where: { id: materialId },
        data: {
          videoProcessingStatus: 'done',
          videoPreview540Url: result.preview540Url,
          videoPreviewClipUrl: result.clipUrl,
          videoCoverUrl: result.coverUrl || undefined,
          videoDuration: result.duration,
          transcodeLastError: null,
          transcodeAttempts: attempt,
          previewUrl: nextPreviewUrl || undefined,
        },
      })

      console.log(`[Transcode] Task completed for material ${materialId}`)
      return
    } catch (error) {
      const errorMsg = extractErrorMessage(error)
      console.error(`[Transcode] Task attempt ${attempt} failed for material ${materialId}:`, errorMsg)

      const shouldRetry = attempt < MAX_TRANSCODE_RETRIES && !isNonRetryableTranscodeError(error)

      if (!shouldRetry) {
        await prisma.material.update({
          where: { id: materialId },
          data: {
            videoProcessingStatus: 'failed',
            transcodeLastError: errorMsg.substring(0, 500),
            transcodeAttempts: attempt,
          },
        })
        const finalError = error instanceof Error ? error : new Error(errorMsg)
        throw finalError
      }

      await prisma.material.update({
        where: { id: materialId },
        data: {
          transcodeLastError: errorMsg.substring(0, 500),
          transcodeAttempts: attempt,
        },
      })

      console.warn(`[Transcode] Retrying material ${materialId} (${attempt}/${MAX_TRANSCODE_RETRIES})`)
    }
  }
}

/**
 * 验证视频文件是否存在
 */
async function validateVideoFile(videoSourceUrl: string): Promise<boolean> {
  try {
    // 如果是 HTTP URL，跳过本地文件检查
    if (videoSourceUrl.startsWith('http://') || videoSourceUrl.startsWith('https://')) {
      return true
    }

    const path = await import('path')
    const { access, constants } = await import('fs/promises')
    
    let localFilePath = videoSourceUrl
    
    // 如果是相对路径（以 /uploads/ 开头），转换为绝对路径
    if (videoSourceUrl.startsWith('/uploads/')) {
      const uploadDir = process.env.UPLOAD_DIR || './uploads'
      const absoluteUploadDir = path.isAbsolute(uploadDir) ? uploadDir : path.join(process.cwd(), uploadDir)
      localFilePath = path.join(absoluteUploadDir, videoSourceUrl.replace('/uploads/', ''))
    } else if (!path.isAbsolute(videoSourceUrl)) {
      // 相对路径：基于当前工作目录
      localFilePath = path.join(process.cwd(), videoSourceUrl)
    }
    
    // 检查文件是否存在
    await access(localFilePath, constants.F_OK)
    console.log(`[TranscodeTask] ✅ Video file exists: ${localFilePath}`)
    return true
  } catch (error) {
    console.error(`[TranscodeTask] ❌ Video file not found: ${videoSourceUrl}`, error)
    return false
  }
}

/**
 * 入队一个视频转码任务
 */
export async function enqueueVideoTranscodeTask(
  materialId: string,
  videoSourceUrl: string,
  needsCoverImage: boolean
): Promise<void> {
  // 验证视频文件是否存在
  const fileExists = await validateVideoFile(videoSourceUrl)
  if (!fileExists) {
    throw new Error(`视频文件不存在: ${videoSourceUrl}`)
  }

  console.log(`[TranscodeTask] 📝 Enqueuing transcode task: materialId=${materialId}, videoSourceUrl=${videoSourceUrl}, needsCoverImage=${needsCoverImage}`)

  // 注意：这里不等待整个转码流程完成，只负责把任务丢进队列即可
  // 队列内部会在任务开始/结束时更新数据库状态和进度
  transcodeQueue
    .add(materialId, videoSourceUrl, needsCoverImage, executeTranscodeTask)
    .then(() => {
      console.log(`[TranscodeTask] ✅ Transcode task finished in queue: materialId=${materialId}`)
    })
    .catch((error) => {
      // 这里一般不会抛出未处理错误，因为 executeTranscodeTask 内部已经负责更新失败状态
      console.error(`[TranscodeTask] ❌ Transcode task error in queue for material ${materialId}:`, error)
    })

  console.log(`[TranscodeTask] ✅ Transcode task enqueued successfully (fire-and-forget): materialId=${materialId}`)
}