补强视频转码恢复与失败处理

lib/__tests__/video-transcode-queue.test.ts

@@ -0,0 +1,34 @@
+export {}
+
+describe('video-transcode-queue', () => {
+  const originalEnv = process.env
+
+  afterEach(() => {
+    process.env = originalEnv
+    jest.resetModules()
+  })
+
+  it('无参创建单例时会读取 VIDEO_TRANSCODE_MAX_CONCURRENT', async () => {
+    process.env = {
+      ...originalEnv,
+      VIDEO_TRANSCODE_MAX_CONCURRENT: '3',
+    }
+    jest.resetModules()
+
+    const { transcodeQueue } = await import('../video-transcode-queue')
+
+    expect(transcodeQueue.getStatus().maxConcurrent).toBe(3)
+  })
+
+  it('非法并发配置会回退为 1，保护低性能私有化环境', async () => {
+    process.env = {
+      ...originalEnv,
+      VIDEO_TRANSCODE_MAX_CONCURRENT: '0',
+    }
+    jest.resetModules()
+
+    const { transcodeQueue } = await import('../video-transcode-queue')
+
+    expect(transcodeQueue.getStatus().maxConcurrent).toBe(1)
+  })
+})

lib/__tests__/video-transcode-service.test.ts

@@ -35,6 +35,8 @@ describe('video-transcode-service', () => {
     jest.resetAllMocks()
     jest.useRealTimers()
     transcodeQueue.add.mockResolvedValue(undefined)
+    prisma.videoTranscodeJob.upsert.mockResolvedValue({})
+    prisma.material.update.mockResolvedValue({})
   })
 
   afterEach(() => {
@@ -105,6 +107,35 @@ describe('video-transcode-service', () => {
     expect(prisma.$queryRaw.mock.calls[0][0].join('')).toContain(`OR m."materialType" = '视频'`)
   })
 
+  it('恢复批次达到上限时会标记 hasMore 并安排继续恢复', async () => {
+    jest.useFakeTimers()
+    const { recoverStaleVideoTranscodeJobs } = require('../video-transcode-service')
+
+    prisma.videoTranscodeJob.findMany.mockResolvedValue(
+      Array.from({ length: 20 }, (_, index) => ({
+        materialId: `material-${index}`,
+        videoSourceUrl: `s3/video-${index}.mp4`,
+        material: {
+          id: `material-${index}`,
+          videoSourceUrl: `s3/video-${index}.mp4`,
+          videoCoverUrl: null,
+        },
+      }))
+    )
+    prisma.$queryRaw.mockResolvedValue([])
+
+    const result = await recoverStaleVideoTranscodeJobs()
+
+    expect(result).toEqual({ recovered: 20, hasMore: true })
+    expect(transcodeQueue.add).toHaveBeenCalledTimes(20)
+
+    prisma.videoTranscodeJob.findMany.mockResolvedValue([])
+    prisma.$queryRaw.mockResolvedValue([])
+    await jest.advanceTimersByTimeAsync(1000)
+
+    expect(prisma.videoTranscodeJob.findMany).toHaveBeenCalledTimes(2)
+  })
+
   it('任务失败但还有重试次数时，会重新写回 queued 并延迟重新入队', async () => {
     jest.useFakeTimers()
     const { enqueueVideoTranscodeJob } = require('../video-transcode-service')
@@ -154,4 +185,48 @@ describe('video-transcode-service', () => {
     expect(transcodeQueue.add).toHaveBeenCalledTimes(2)
     expect(transcodeQueue.add.mock.calls[1][0]).toBe('material-1')
   })
+
+  it('任务开始处理时同步更新素材状态为 processing', async () => {
+    const { enqueueVideoTranscodeJob } = require('../video-transcode-service')
+
+    prisma.videoTranscodeJob.update.mockResolvedValue({
+      materialId: 'material-1',
+      attempts: 1,
+      maxAttempts: 3,
+    })
+    processVideoTranscode.mockResolvedValue({
+      preview540Url: 's3/preview.mp4',
+      clipUrl: 's3/clip.mp4',
+      coverUrl: null,
+      duration: 12,
+    })
+
+    let queueRun: Promise<void> | undefined
+    transcodeQueue.add.mockImplementationOnce(async (
+      materialId: string,
+      videoSourceUrl: string,
+      needsCoverImage: boolean,
+      taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise<void>
+    ) => {
+      queueRun = taskFn(materialId, videoSourceUrl, needsCoverImage)
+      await queueRun
+    })
+
+    await enqueueVideoTranscodeJob({
+      materialId: 'material-1',
+      videoSourceUrl: 's3/video.mp4',
+      needsCoverImage: false,
+    })
+
+    await queueRun
+
+    expect(prisma.material.update).toHaveBeenCalledWith({
+      where: { id: 'material-1' },
+      data: { videoProcessingStatus: 'processing' },
+    })
+    expect(prisma.material.update).toHaveBeenLastCalledWith({
+      where: { id: 'material-1' },
+      data: expect.objectContaining({ videoProcessingStatus: 'done' }),
+    })
+  })
 })

lib/video-processor.ts

@@ -89,8 +89,8 @@ export async function checkFFmpegAvailable(): Promise<boolean> {
  * 从 S3 下载视频到临时文件
  */
 async function downloadVideoToTemp(s3Url: string, tempPath: string): Promise<void> {
-  if (!isS3Url(s3Url)) {
-    throw new Error('Invalid S3 URL')
+  if (!s3Url) {
+    throw new Error('Invalid S3 path')
   }
 
   const stream = await getS3ObjectStream(s3Url)
@@ -225,13 +225,7 @@ export async function processVideoTranscode(
   // 检查 ffmpeg 是否可用
   const ffmpegAvailable = await checkFFmpegAvailable()
   if (!ffmpegAvailable) {
-    console.warn('[VideoProcessor] FFmpeg not available, skipping transcode')
-    return {
-      preview540Url: null,
-      clipUrl: null,
-      coverUrl: null,
-      duration: null,
-    }
+    throw new Error('FFmpeg 不可用，无法执行视频转码')
   }
 
   const tempDir = tmpdir()

lib/video-transcode-queue.ts

@@ -18,10 +18,14 @@ class VideoTranscodeQueue {
   private queued: Set<string> = new Set()
   private maxConcurrent: number
 
-  constructor(maxConcurrent: number = 1) {
+  constructor(maxConcurrent?: number) {
     // 默认并发数为 1，适合低性能 NAS 设备
     // 可以通过环境变量配置：VIDEO_TRANSCODE_MAX_CONCURRENT
-    this.maxConcurrent = maxConcurrent || parseInt(process.env.VIDEO_TRANSCODE_MAX_CONCURRENT || '1', 10)
+    const configuredMaxConcurrent = parseInt(process.env.VIDEO_TRANSCODE_MAX_CONCURRENT || '1', 10)
+    const resolvedMaxConcurrent = maxConcurrent ?? configuredMaxConcurrent
+    this.maxConcurrent = Number.isFinite(resolvedMaxConcurrent) && resolvedMaxConcurrent > 0
+      ? resolvedMaxConcurrent
+      : 1
   }
 
   /**

lib/video-transcode-service.ts

@@ -7,6 +7,7 @@ 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)
@@ -45,7 +46,7 @@ async function runVideoTranscodeTask(
 ) {
   const maxAttempts = getMaxAttempts()
 
-  const job = await (prisma as any).videoTranscodeJob.update({
+    const job = await (prisma as any).videoTranscodeJob.update({
     where: { materialId },
     data: {
       status: 'processing',
@@ -61,6 +62,11 @@ async function runVideoTranscodeTask(
   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(),
@@ -211,7 +217,7 @@ export async function recoverStaleVideoTranscodeJobs() {
         },
       },
     },
-    take: 20,
+    take: RECOVERY_BATCH_SIZE,
     orderBy: { updatedAt: 'asc' },
   })
 
@@ -228,7 +234,7 @@ export async function recoverStaleVideoTranscodeJobs() {
       AND m."videoProcessingStatus" IN ('pending', 'processing')
       AND j.id IS NULL
     ORDER BY m."updatedAt" ASC
-    LIMIT 20
+    LIMIT ${RECOVERY_BATCH_SIZE}
   `
 
   let recovered = 0
@@ -272,5 +278,14 @@ export async function recoverStaleVideoTranscodeJobs() {
     console.log(`[Transcode] Recovered ${recovered} queued/stale video transcode job(s)`)
   }
 
-  return { recovered }
+  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 }
 }