4e89b57798
- Expand batch upload list area and optimize preview image layout - Improve upload speed tracking and display (server-side & client-side) - Reduce network detection log noise - Optimize Docker build process (caching, env validation) - Add shadow environment deployment support
597 lines
20 KiB
TypeScript
597 lines
20 KiB
TypeScript
/**
|
||
* 视频转码任务队列
|
||
*
|
||
* 用于控制转码任务的并发数，充分利用多核 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%）
|
||
*/
|
||
|
||
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
|
||
}
|
||
|
||
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,
|
||
status: { in: ['completed', 'failed'] }
|
||
}
|
||
})
|
||
} 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, retryCount?: number) => 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, retryCount?: number) => 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)`)
|
||
|
||
// 执行任务（传递 retryCount=0 作为初始重试次数）
|
||
// 注意：这里不 await，让任务异步执行，队列继续处理下一个任务
|
||
taskFn(task.materialId, task.videoSourceUrl, task.needsCoverImage, 0)
|
||
.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, retryCount?: number) => 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') {
|
||
this.progress.set(materialId, progress)
|
||
if (stage) {
|
||
this.stages.set(materialId, stage)
|
||
// R1修复：异步保存到数据库（不阻塞）
|
||
this.saveStageProgress(materialId, stage, 'processing', progress)
|
||
.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, retryCount?: number) => Promise<void>,
|
||
cleanupFn: (material: { id: string; videoPreview540Url: string | null; videoPreviewClipUrl: string | null; videoCoverUrl: string | null }) => Promise<void>
|
||
): Promise<{ recovered: number; requeued: number }> {
|
||
const { prisma } = await import('@/lib/prisma')
|
||
|
||
let recovered = 0
|
||
let requeued = 0
|
||
|
||
try {
|
||
// 1. 查询所有 processing 状态的视频（部署中断时留下的）
|
||
const processingMaterials = await prisma.material.findMany({
|
||
where: {
|
||
videoProcessingStatus: 'processing',
|
||
videoSourceUrl: { not: null },
|
||
deletedAt: null,
|
||
} as any,
|
||
select: {
|
||
id: true,
|
||
videoPreview540Url: true,
|
||
videoPreviewClipUrl: true,
|
||
videoCoverUrl: true,
|
||
videoSourceUrl: true,
|
||
} as any,
|
||
})
|
||
|
||
console.log(`[TranscodeQueue] Found ${processingMaterials.length} interrupted tasks (processing status)`)
|
||
|
||
// 2. 清理残留文件并重置状态为 pending
|
||
for (const material of processingMaterials) {
|
||
try {
|
||
// 提取字段以避免类型错误
|
||
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
|
||
|
||
// 清理残留文件（类型转换以匹配 cleanupFn 的期望）
|
||
await cleanupFn({
|
||
id: materialId,
|
||
videoPreview540Url: preview540Url,
|
||
videoPreviewClipUrl: previewClipUrl,
|
||
videoCoverUrl: coverUrl,
|
||
} as any)
|
||
|
||
// 重置状态为 pending，清除错误信息
|
||
await prisma.material.update({
|
||
where: { id: materialId },
|
||
data: {
|
||
videoProcessingStatus: 'pending',
|
||
transcodeLastError: null,
|
||
} as any,
|
||
})
|
||
|
||
recovered++
|
||
console.log(`[TranscodeQueue] Recovered task: ${material.id}`)
|
||
} catch (error) {
|
||
console.error(`[TranscodeQueue] Failed to recover task ${material.id}:`, error)
|
||
}
|
||
}
|
||
|
||
// 3. 查询所有 pending 状态的视频，按时间排序
|
||
const pendingMaterials = await prisma.material.findMany({
|
||
where: {
|
||
videoProcessingStatus: 'pending',
|
||
videoSourceUrl: { not: null },
|
||
deletedAt: null,
|
||
} as any,
|
||
orderBy: { createdAt: 'asc' }, // 按创建时间排序，保持原始队列顺序
|
||
select: {
|
||
id: true,
|
||
videoSourceUrl: true,
|
||
videoCoverUrl: true,
|
||
} as any,
|
||
})
|
||
|
||
console.log(`[TranscodeQueue] Found ${pendingMaterials.length} pending tasks to requeue`)
|
||
|
||
// 4. 按顺序重新入队（但不立即执行，等待队列调度）
|
||
for (const material of pendingMaterials) {
|
||
try {
|
||
// 提取字段以避免类型错误
|
||
const materialId = String(material.id)
|
||
const videoSourceUrl = material.videoSourceUrl ? String(material.videoSourceUrl) : null
|
||
const needsCoverImage = !material.videoCoverUrl
|
||
|
||
if (!videoSourceUrl) {
|
||
console.warn(`[TranscodeQueue] Skipping material ${materialId}: no videoSourceUrl`)
|
||
continue
|
||
}
|
||
|
||
// 重新加入队列（队列会自动调度执行）
|
||
await this.add(
|
||
materialId,
|
||
videoSourceUrl,
|
||
needsCoverImage,
|
||
taskFn
|
||
)
|
||
requeued++
|
||
} catch (error) {
|
||
console.error(`[TranscodeQueue] Failed to requeue task ${material.id}:`, error)
|
||
}
|
||
}
|
||
|
||
console.log(`[TranscodeQueue] Recovery completed: recovered ${recovered} tasks, requeued ${requeued} 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()
|
||
|