gongxiangp2/lib/video-transcode-queue.ts
2026-05-07 16:37:22 +08:00

635 lines
21 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

/**
* 视频转码任务队列
*
* 用于控制转码任务的并发数,充分利用多核 CPU
*
* 优化说明:
* - 默认并发数:2(充分利用多核 CPU)
* - 可以通过环境变量配置:VIDEO_TRANSCODE_MAX_CONCURRENT
* - 支持队列管理,避免系统过载
* - 支持部署中断后自动恢复转码任务
*
* 并发数建议:
* - H.265 编码:2 并发(CPU 占用 60-80%)
* - H.265 编码:3-4 并发(高性能服务器,CPU 占用 80-100%)
* - H.264 编码:3-4 并发(CPU 占用 60-80%)
*/
import { LEGACY_COMPATIBLE_VIDEO_MATERIAL_TYPES } from './video-material-types'
export const VIDEO_MATERIAL_TYPES = [...LEGACY_COMPATIBLE_VIDEO_MATERIAL_TYPES]
interface QueuedTask {
materialId: string
videoSourceUrl: string
needsCoverImage: boolean
resolve: () => void
reject: (error: Error) => void
}
// R1修复:每个阶段的进度接口
interface StageProgress {
materialId: string
videoSourceUrl: string
needsCoverImage: boolean
resolve: () => void
reject: (error: Error) => void
}
// R1修复:素材的所有阶段任务
interface MaterialTasks {
materialId: string
videoSourceUrl: string
needsCoverImage: boolean
stages: Array<{
name: 'preview' | 'clip' | 'cover'
promise: Promise<void>
}>
resolve: () => void
reject: (error: Error) => void
}
interface RecoverInterruptedTasksOptions {
requeuePending?: boolean
limit?: number
videoTypes?: string[]
}
class VideoTranscodeQueue {
private queue: QueuedTask[] = []
private processing: Set<string> = new Set()
private maxConcurrent: number
private progress: Map<string, number> = new Map() // 转码进度:materialId -> 0-100
private stages: Map<string, 'preview' | 'clip' | 'cover'> = new Map() // 转码阶段:materialId -> stage
private prisma: any = null // R1修复:Prisma 客户端延迟初始化
constructor(maxConcurrent?: number) {
// 默认并发数计算逻辑:
// 1. 优先使用环境变量配置
// 2. 如果没有配置,使用默认值 2(充分利用多核 CPU)
// 3. 可以根据实际负载情况调整到 3-4(高性能服务器)
const envConcurrent = process.env.VIDEO_TRANSCODE_MAX_CONCURRENT
if (envConcurrent) {
this.maxConcurrent = parseInt(envConcurrent, 10) || 2
} else if (maxConcurrent !== undefined) {
this.maxConcurrent = maxConcurrent
} else {
// 默认值:2 并发(充分利用多核 CPU,提升 CPU 使用率)
// 可以根据实际负载情况调整到 3-4(高性能服务器)
// 注意:使用 H.265 (libx265) 时,2 并发可以达到 60-80% CPU 使用率
this.maxConcurrent = 2
}
console.log(`[TranscodeQueue] Initialized with maxConcurrent: ${this.maxConcurrent}`)
}
/**
* R1修复:延迟初始化 Prisma 客户端(避免循环依赖)
*/
private async getPrisma() {
if (!this.prisma) {
const { prisma } = await import('@/lib/prisma')
this.prisma = prisma
}
return this.prisma
}
/**
* R1修复:保存阶段进度到数据库
*/
private async saveStageProgress(
materialId: string,
stage: 'preview' | 'clip' | 'cover',
status: 'waiting' | 'processing' | 'completed' | 'failed',
progress: number = 0,
errorMessage: string | null = null
): Promise<void> {
const prisma = await this.getPrisma()
try {
await prisma.videoTranscodeProgress.upsert({
where: {
materialId_stage: { materialId, stage }
},
create: {
materialId,
stage,
status,
progress,
errorMessage,
lastHeartbeat: new Date(),
},
update: {
status,
progress,
errorMessage,
lastHeartbeat: status === 'processing' ? new Date() : undefined,
completedAt: status === 'completed' || status === 'failed' ? new Date() : undefined,
}
})
} catch (error) {
console.error(`[TranscodeQueue] Failed to save progress for ${materialId}/${stage}:`, error)
}
}
/**
* R1修复:从数据库加载等待中的任务
*/
async loadWaitingTasks(): Promise<Array<{ materialId: string; videoSourceUrl: string; needsCoverImage: boolean }>> {
const prisma = await this.getPrisma()
try {
// 查询所有 waiting 或 processing 状态但超时的任务
const timeoutThreshold = new Date(Date.now() - 10 * 60 * 1000) // 10分钟心跳超时
const tasks = await prisma.videoTranscodeProgress.findMany({
where: {
OR: [
{ status: 'waiting' },
{
AND: [
{ status: 'processing' },
{ lastHeartbeat: { lt: timeoutThreshold } }
]
}
]
},
include: {
material: {
select: {
videoSourceUrl: true,
videoCoverUrl: true,
}
}
},
orderBy: [
{ priority: 'desc' },
{ createdAt: 'asc' }
]
})
// 按素材分组(去重)
const materialTasks = new Map<string, { videoSourceUrl: string; needsCoverImage: boolean }>()
for (const task of tasks) {
if (!materialTasks.has(task.materialId)) {
materialTasks.set(task.materialId, {
videoSourceUrl: (task.material as any).videoSourceUrl,
needsCoverImage: !(task.material as any).videoCoverUrl
})
}
}
console.log(`[TranscodeQueue] Loaded ${materialTasks.size} tasks from database`)
return Array.from(materialTasks.entries()).map(([materialId, data]) => ({
materialId,
videoSourceUrl: data.videoSourceUrl,
needsCoverImage: data.needsCoverImage
}))
} catch (error) {
console.error('[TranscodeQueue] Failed to load waiting tasks:', error)
return []
}
}
/**
* R1修复:清理已完成的阶段记录
*/
private async cleanupCompletedStages(materialId: string): Promise<void> {
const prisma = await this.getPrisma()
try {
await prisma.videoTranscodeProgress.deleteMany({
where: {
materialId,
}
})
} catch (error) {
console.error(`[TranscodeQueue] Failed to cleanup stages for ${materialId}:`, error)
}
}
/**
* 添加转码任务到队列
*/
async add(
materialId: string,
videoSourceUrl: string,
needsCoverImage: boolean,
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise<void>
): Promise<void> {
// 如果已经在处理中,直接返回
if (this.processing.has(materialId)) {
console.log(`[TranscodeQueue] Material ${materialId} is already processing, skipping`)
return
}
// 检查是否已在队列中等待
const alreadyInQueue = this.queue.some(t => t.materialId === materialId)
if (alreadyInQueue) {
console.log(`[TranscodeQueue] Material ${materialId} is already in queue, skipping`)
return
}
console.log(`[TranscodeQueue] Adding task to queue: materialId=${materialId}, videoSourceUrl=${videoSourceUrl}, needsCoverImage=${needsCoverImage}`)
// R1修复:保存到数据库(持久化队列状态)
const stages: Array<'preview' | 'clip' | 'cover'> = ['preview', 'clip']
if (needsCoverImage) {
stages.push('cover')
}
for (const stage of stages) {
await this.saveStageProgress(materialId, stage, 'waiting', 0)
}
return new Promise((resolve, reject) => {
this.queue.push({
materialId,
videoSourceUrl,
needsCoverImage,
resolve,
reject,
})
console.log(`[TranscodeQueue] Task added to queue. Queue length: ${this.queue.length}, Processing: ${this.processing.size}/${this.maxConcurrent}`)
// 立即尝试处理队列
this.process(taskFn).catch((error) => {
console.error(`[TranscodeQueue] Error in process() for material ${materialId}:`, error)
// 如果 process 本身出错,从队列中移除任务并拒绝 Promise
const taskIndex = this.queue.findIndex(t => t.materialId === materialId)
if (taskIndex >= 0) {
const task = this.queue[taskIndex]
this.queue.splice(taskIndex, 1)
task.reject(error)
}
})
})
}
/**
* 处理队列中的任务
*/
private async process(
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise<void>
): Promise<void> {
try {
// 如果已达到最大并发数,等待
if (this.processing.size >= this.maxConcurrent) {
return
}
// 如果队列为空,返回
if (this.queue.length === 0) {
return
}
// 取出队列中的第一个任务
const task = this.queue.shift()
if (!task) {
return
}
// 验证任务参数
if (!task.materialId || !task.videoSourceUrl) {
console.error(`[TranscodeQueue] Invalid task parameters:`, task)
task.reject(new Error('Invalid task parameters'))
// 继续处理下一个任务
this.process(taskFn).catch(err => console.error('[TranscodeQueue] Error processing next task:', err))
return
}
// 标记为处理中
this.processing.add(task.materialId)
console.log(`[TranscodeQueue] 🚀 Starting task for material ${task.materialId} (queue: ${this.queue.length} waiting, ${this.processing.size}/${this.maxConcurrent} processing)`)
// 执行任务(内部自行处理重试逻辑)
// 注意:这里不 await,让任务异步执行,队列继续处理下一个任务
taskFn(task.materialId, task.videoSourceUrl, task.needsCoverImage)
.then(() => {
// 转码成功,清除进度信息
this.clearProgress(task.materialId)
console.log(`[TranscodeQueue] ✅ Task completed for material ${task.materialId}`)
task.resolve()
})
.catch((error) => {
// 转码失败,清除进度信息
this.clearProgress(task.materialId)
const errorMsg = error?.message || String(error)
console.error(`[TranscodeQueue] ❌ Task failed for material ${task.materialId}:`, errorMsg)
task.reject(error)
})
.finally(() => {
// 从处理中移除
this.processing.delete(task.materialId)
console.log(`[TranscodeQueue] Task removed from processing. Remaining: ${this.processing.size}/${this.maxConcurrent}`)
// 继续处理下一个任务
this.process(taskFn).catch(err => {
console.error('[TranscodeQueue] Error processing next task after completion:', err)
})
})
} catch (error) {
console.error('[TranscodeQueue] Error in process() method:', error)
// 即使 process 方法本身出错,也要继续处理队列
// 避免整个队列卡死
setTimeout(() => {
this.process(taskFn).catch(err => {
console.error('[TranscodeQueue] Error in retry process():', err)
})
}, 1000)
}
}
/**
* 获取队列状态
*/
getStatus() {
return {
queueLength: this.queue.length,
processingCount: this.processing.size,
processingIds: Array.from(this.processing),
maxConcurrent: this.maxConcurrent,
progress: this.progress,
stages: this.stages,
waiting: this.queue.map(t => t.materialId),
}
}
/**
* R1修复:启动时恢复队列(从数据库加载未完成任务)
* @returns 恢复的任务数量
*/
async recoverOnStartup(
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise<void>
): Promise<number> {
console.log('[TranscodeQueue] Starting recovery from database...')
const tasks = await this.loadWaitingTasks()
let recovered = 0
for (const task of tasks) {
try {
// 重新入队(但不重复保存数据库记录)
this.queue.push({
materialId: task.materialId,
videoSourceUrl: task.videoSourceUrl,
needsCoverImage: task.needsCoverImage,
resolve: () => {}, // 启动恢复不需要 resolve/reject
reject: () => {},
})
recovered++
} catch (error) {
console.error(`[TranscodeQueue] Failed to recover task ${task.materialId}:`, error)
}
}
console.log(`[TranscodeQueue] Recovered ${recovered} tasks from database`)
// 开始处理队列
if (recovered > 0) {
this.process(taskFn).catch(err => {
console.error('[TranscodeQueue] Error processing recovered tasks:', err)
})
}
return recovered
}
/**
* 更新转码进度
*/
updateProgress(materialId: string, progress: number, stage?: 'preview' | 'clip' | 'cover') {
const normalized = Math.min(100, Math.max(0, progress))
this.progress.set(materialId, normalized)
if (stage) {
this.stages.set(materialId, stage)
const status: 'processing' | 'completed' = normalized >= 100 ? 'completed' : 'processing'
this.saveStageProgress(materialId, stage, status, normalized)
.catch(err => console.error('[TranscodeQueue] Failed to save progress:', err))
}
}
/**
* 清除转码进度(转码完成或失败时)
*/
clearProgress(materialId: string) {
this.progress.delete(materialId)
this.stages.delete(materialId)
// R1修复:异步清理数据库记录(不阻塞)
this.cleanupCompletedStages(materialId)
.catch(err => console.error('[TranscodeQueue] Failed to cleanup stages:', err))
}
/**
* 获取队列详情(包含等待中的素材ID列表,方便前端展示"排队位置")
*/
getDetailedStatus() {
return {
queue: this.queue.map((t) => t.materialId),
processing: Array.from(this.processing),
maxConcurrent: this.maxConcurrent,
}
}
/**
* 恢复中断的转码任务
*
* 部署中断后的恢复流程:
* 1. 查询所有 processing 状态的视频,清理残留文件并重置为 pending
* 2. 查询所有 pending 状态的视频,按时间顺序重新入队
*
* @param taskFn 转码任务执行函数
* @param cleanupFn 清理残留文件的函数
*/
async recoverInterruptedTasks(
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise<void>,
cleanupFn: (material: { id: string; videoPreview540Url: string | null; videoPreviewClipUrl: string | null; videoCoverUrl: string | null }) => Promise<void>,
options: RecoverInterruptedTasksOptions = {}
): Promise<{ recovered: number; requeued: number }> {
const prisma = await this.getPrisma()
const { requeuePending = false, limit, videoTypes = VIDEO_MATERIAL_TYPES } = options
const materialTypesFilter = videoTypes.length > 0 ? videoTypes : undefined
let recovered = 0
let requeued = 0
try {
const processingWhere: Record<string, unknown> = {
videoProcessingStatus: 'processing',
videoSourceUrl: { not: null },
deletedAt: null,
}
if (materialTypesFilter) {
processingWhere.materialType = { in: materialTypesFilter }
}
const processingMaterials = await prisma.material.findMany({
where: processingWhere,
select: {
id: true,
videoPreview540Url: true,
videoPreviewClipUrl: true,
videoCoverUrl: true,
videoSourceUrl: true,
},
})
console.log(`[TranscodeQueue] Found ${processingMaterials.length} interrupted tasks (processing status)`)
const recoveredPayloads: Array<{ materialId: string; videoSourceUrl: string; needsCoverImage: boolean }> = []
for (const material of processingMaterials) {
const materialId = String(material.id)
const preview540Url = material.videoPreview540Url ? String(material.videoPreview540Url) : null
const previewClipUrl = material.videoPreviewClipUrl ? String(material.videoPreviewClipUrl) : null
const coverUrl = material.videoCoverUrl ? String(material.videoCoverUrl) : null
const sourceUrl = material.videoSourceUrl ? String(material.videoSourceUrl) : ''
const needsCoverImage = !material.videoCoverUrl
try {
await cleanupFn({
id: materialId,
videoPreview540Url: preview540Url,
videoPreviewClipUrl: previewClipUrl,
videoCoverUrl: coverUrl,
})
await prisma.material.update({
where: { id: materialId },
data: {
videoProcessingStatus: 'pending',
transcodeLastError: null,
},
})
recovered++
if (sourceUrl) {
recoveredPayloads.push({
materialId,
videoSourceUrl: sourceUrl,
needsCoverImage,
})
} else {
console.warn(`[TranscodeQueue] Skipping recovered task ${materialId}: missing videoSourceUrl`)
}
} catch (error) {
console.error(`[TranscodeQueue] Failed to recover task ${materialId}:`, error)
}
}
for (const payload of recoveredPayloads) {
try {
await this.add(payload.materialId, payload.videoSourceUrl, payload.needsCoverImage, taskFn)
} catch (error) {
console.error(`[TranscodeQueue] Failed to requeue recovered task ${payload.materialId}:`, error)
}
}
if (requeuePending) {
const pendingWhere: Record<string, unknown> = {
videoProcessingStatus: 'pending',
videoSourceUrl: { not: null },
deletedAt: null,
}
if (materialTypesFilter) {
pendingWhere.materialType = { in: materialTypesFilter }
}
const pendingQueryOptions: Record<string, unknown> = {
where: pendingWhere,
orderBy: { createdAt: 'asc' },
select: {
id: true,
videoSourceUrl: true,
videoCoverUrl: true,
},
}
if (typeof limit === 'number' && limit > 0) {
pendingQueryOptions.take = limit
}
const pendingMaterials = await prisma.material.findMany(pendingQueryOptions)
console.log(`[TranscodeQueue] Found ${pendingMaterials.length} pending tasks to requeue (option enabled)`)
for (const material of pendingMaterials) {
const materialId = String(material.id)
const sourceUrl = material.videoSourceUrl ? String(material.videoSourceUrl) : ''
if (!sourceUrl) {
console.warn(`[TranscodeQueue] Skipping pending material ${materialId}: no videoSourceUrl`)
continue
}
try {
await this.add(materialId, sourceUrl, !material.videoCoverUrl, taskFn)
requeued++
} catch (error) {
console.error(`[TranscodeQueue] Failed to requeue pending task ${materialId}:`, error)
}
}
} else {
console.log('[TranscodeQueue] Pending task requeue disabled; skipped scanning pending list')
}
console.log(`[TranscodeQueue] Recovery completed: recovered ${recovered} tasks, requeued ${requeued} pending tasks`)
} catch (error) {
console.error('[TranscodeQueue] Recovery failed:', error)
}
return { recovered, requeued }
}
}
/**
* 清理转码残留文件(S3上的预览视频、片段、自动生成的封面)
*
* @param material 素材信息(包含转码生成的文件URL)
*/
export async function cleanupTranscodeResidues(material: {
id: string
videoPreview540Url: string | null
videoPreviewClipUrl: string | null
videoCoverUrl: string | null
}): Promise<void> {
const { deleteFromS3, isS3PublicUrl } = await import('@/lib/s3')
const filesToDelete: string[] = []
// 1. 预览视频(540P)- 如果存在且是S3 URL,删除
if (material.videoPreview540Url && isS3PublicUrl(material.videoPreview540Url)) {
filesToDelete.push(material.videoPreview540Url)
}
// 2. 预览片段(10秒)- 如果存在且是S3 URL,删除
if (material.videoPreviewClipUrl && isS3PublicUrl(material.videoPreviewClipUrl)) {
filesToDelete.push(material.videoPreviewClipUrl)
}
// 3. 封面图 - 仅当是自动生成的(文件名包含 "video-cover-")才删除
if (material.videoCoverUrl && isS3PublicUrl(material.videoCoverUrl)) {
const fileName = material.videoCoverUrl.split('/').pop() || ''
if (fileName.includes('video-cover-')) {
filesToDelete.push(material.videoCoverUrl)
} else {
console.log(`[TranscodeQueue] Skipping user-uploaded cover: ${material.videoCoverUrl}`)
}
}
// 批量删除文件
if (filesToDelete.length > 0) {
console.log(`[TranscodeQueue] Cleaning up ${filesToDelete.length} residue files for material ${material.id}`)
for (const url of filesToDelete) {
try {
await deleteFromS3(url)
console.log(`[TranscodeQueue] Deleted: ${url}`)
} catch (error) {
console.error(`[TranscodeQueue] Failed to delete ${url}:`, error)
}
}
}
}
// 单例模式
export const transcodeQueue = new VideoTranscodeQueue()