// 视频转码任务封装,供上传接口和转码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 { 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 { 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 { // 验证视频文件是否存在 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}`) }