353 lines
10 KiB
TypeScript
353 lines
10 KiB
TypeScript
import { tmpdir } from 'os'
|
||
import { prisma } from '@/lib/prisma'
|
||
import { processVideoTranscode } from '@/lib/video-processor'
|
||
import { transcodeQueue } from '@/lib/video-transcode-queue'
|
||
|
||
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
|
||
type VideoTranscodeJobDelegate = {
|
||
upsert: (...args: any[]) => Promise<any>
|
||
findMany: (...args: any[]) => Promise<any[]>
|
||
update: (...args: any[]) => Promise<any>
|
||
}
|
||
|
||
function getVideoTranscodeJobDelegate(): VideoTranscodeJobDelegate | null {
|
||
const delegate = (prisma as any).videoTranscodeJob
|
||
if (
|
||
delegate &&
|
||
typeof delegate.upsert === 'function' &&
|
||
typeof delegate.findMany === 'function' &&
|
||
typeof delegate.update === 'function'
|
||
) {
|
||
return delegate
|
||
}
|
||
return null
|
||
}
|
||
|
||
function requireVideoTranscodeJobDelegate(): VideoTranscodeJobDelegate {
|
||
const delegate = getVideoTranscodeJobDelegate()
|
||
if (!delegate) {
|
||
throw new Error('Prisma Client 缺少 videoTranscodeJob，请先执行数据库迁移并重新生成 Prisma Client')
|
||
}
|
||
return delegate
|
||
}
|
||
|
||
function getMaxAttempts() {
|
||
const retries = parseInt(process.env.VIDEO_TRANSCODE_MAX_RETRIES || '2', 10)
|
||
return Math.max(1, retries + 1)
|
||
}
|
||
|
||
function getLockTimeoutMs() {
|
||
return parseInt(
|
||
process.env.VIDEO_TRANSCODE_LOCK_TIMEOUT_MS || String(DEFAULT_LOCK_TIMEOUT_MS),
|
||
10
|
||
)
|
||
}
|
||
|
||
function getRetryDelayMs(attempt: number) {
|
||
const baseDelay = parseInt(
|
||
process.env.VIDEO_TRANSCODE_RETRY_BASE_DELAY_MS || String(DEFAULT_RETRY_DELAY_MS),
|
||
10
|
||
)
|
||
const safeBaseDelay = Number.isFinite(baseDelay) && baseDelay > 0
|
||
? baseDelay
|
||
: DEFAULT_RETRY_DELAY_MS
|
||
|
||
return Math.min(safeBaseDelay * Math.pow(2, Math.max(0, attempt - 1)), MAX_RETRY_DELAY_MS)
|
||
}
|
||
|
||
function getErrorMessage(error: unknown) {
|
||
if (error instanceof Error) return error.message
|
||
if (typeof error === 'string') return error
|
||
return '未知错误'
|
||
}
|
||
|
||
function isVideoTranscodeSchemaMissing(error: unknown) {
|
||
const code = (error as any)?.code
|
||
const table = String((error as any)?.meta?.table || '')
|
||
const message = getErrorMessage(error)
|
||
|
||
return (
|
||
(code === 'P2021' && (table.includes('video_transcode_jobs') || message.includes('video_transcode_jobs'))) ||
|
||
(code === 'P2022' && message.includes('video_transcode_jobs')) ||
|
||
(code === 'P2010' && message.includes('video_transcode_jobs'))
|
||
)
|
||
}
|
||
|
||
async function runVideoTranscodeTask(
|
||
materialId: string,
|
||
videoSourceUrl: string,
|
||
needsCoverImage: boolean
|
||
) {
|
||
const maxAttempts = getMaxAttempts()
|
||
const videoTranscodeJob = requireVideoTranscodeJobDelegate()
|
||
|
||
const job = await videoTranscodeJob.update({
|
||
where: { materialId },
|
||
data: {
|
||
status: 'processing',
|
||
lockedAt: new Date(),
|
||
startedAt: new Date(),
|
||
finishedAt: null,
|
||
errorMessage: null,
|
||
attempts: { increment: 1 },
|
||
maxAttempts,
|
||
},
|
||
})
|
||
|
||
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(),
|
||
materialId,
|
||
needsCoverImage,
|
||
})
|
||
|
||
const updateData: any = {
|
||
videoProcessingStatus: 'done',
|
||
videoPreview540Url: result.preview540Url,
|
||
videoPreviewClipUrl: result.clipUrl,
|
||
videoDuration: result.duration,
|
||
}
|
||
|
||
if (needsCoverImage && result.coverUrl) {
|
||
updateData.videoCoverUrl = result.coverUrl
|
||
const material = await prisma.material.findUnique({
|
||
where: { id: materialId },
|
||
select: { previewUrl: true },
|
||
})
|
||
if (material?.previewUrl?.startsWith('data:image')) {
|
||
updateData.previewUrl = result.coverUrl
|
||
}
|
||
}
|
||
|
||
await prisma.material.update({
|
||
where: { id: materialId },
|
||
data: updateData,
|
||
})
|
||
|
||
await videoTranscodeJob.update({
|
||
where: { materialId },
|
||
data: {
|
||
status: 'done',
|
||
lockedAt: null,
|
||
finishedAt: new Date(),
|
||
errorMessage: null,
|
||
},
|
||
})
|
||
|
||
console.log(`[Transcode] Persistent task completed for material ${materialId}`)
|
||
} catch (error) {
|
||
const errorMessage = getErrorMessage(error)
|
||
const nextAttempts = Number(job.attempts || 0)
|
||
const shouldRetry = nextAttempts < maxAttempts
|
||
|
||
console.error(`[Transcode] Persistent task failed for material ${materialId}:`, error)
|
||
|
||
await videoTranscodeJob.update({
|
||
where: { materialId },
|
||
data: {
|
||
status: shouldRetry ? 'queued' : 'failed',
|
||
lockedAt: null,
|
||
finishedAt: shouldRetry ? null : new Date(),
|
||
errorMessage,
|
||
},
|
||
})
|
||
|
||
await prisma.material.update({
|
||
where: { id: materialId },
|
||
data: {
|
||
videoProcessingStatus: shouldRetry ? 'pending' : 'failed',
|
||
},
|
||
})
|
||
|
||
if (shouldRetry) {
|
||
const retryDelay = getRetryDelayMs(nextAttempts)
|
||
console.warn(`[Transcode] Task re-queued for material ${materialId} (${nextAttempts}/${maxAttempts}), retrying in ${retryDelay}ms`)
|
||
setTimeout(() => {
|
||
transcodeQueue.add(materialId, videoSourceUrl, needsCoverImage, runVideoTranscodeTask).catch(
|
||
(retryError) => {
|
||
console.error(`[Transcode] Retried task failed for material ${materialId}:`, retryError)
|
||
}
|
||
)
|
||
}, retryDelay)
|
||
}
|
||
|
||
throw error
|
||
}
|
||
}
|
||
|
||
export async function enqueueVideoTranscodeJob(params: {
|
||
materialId: string
|
||
videoSourceUrl: string
|
||
needsCoverImage: boolean
|
||
forceRetry?: boolean
|
||
}) {
|
||
const { materialId, videoSourceUrl, needsCoverImage, forceRetry = false } = params
|
||
const maxAttempts = getMaxAttempts()
|
||
const videoTranscodeJob = requireVideoTranscodeJobDelegate()
|
||
|
||
await videoTranscodeJob.upsert({
|
||
where: { materialId },
|
||
create: {
|
||
materialId,
|
||
videoSourceUrl,
|
||
needsCoverImage,
|
||
status: 'queued',
|
||
attempts: 0,
|
||
maxAttempts,
|
||
errorMessage: null,
|
||
},
|
||
update: {
|
||
videoSourceUrl,
|
||
needsCoverImage,
|
||
status: 'queued',
|
||
lockedAt: null,
|
||
startedAt: forceRetry ? null : undefined,
|
||
finishedAt: null,
|
||
errorMessage: null,
|
||
attempts: forceRetry ? 0 : undefined,
|
||
maxAttempts,
|
||
},
|
||
})
|
||
|
||
await prisma.material.update({
|
||
where: { id: materialId },
|
||
data: { videoProcessingStatus: 'pending' },
|
||
})
|
||
|
||
transcodeQueue.add(materialId, videoSourceUrl, needsCoverImage, runVideoTranscodeTask).catch(
|
||
(error) => {
|
||
console.error(`[Transcode] Queued task failed for material ${materialId}:`, error)
|
||
}
|
||
)
|
||
}
|
||
|
||
export async function recoverStaleVideoTranscodeJobs() {
|
||
const staleBefore = new Date(Date.now() - getLockTimeoutMs())
|
||
const videoTranscodeJob = getVideoTranscodeJobDelegate()
|
||
if (!videoTranscodeJob) {
|
||
console.warn('[Transcode] 跳过启动恢复：Prisma Client 缺少 videoTranscodeJob，请执行数据库迁移并重新生成 Prisma Client')
|
||
return { recovered: 0, hasMore: false, skipped: true }
|
||
}
|
||
|
||
let staleJobs: any[] = []
|
||
let orphanMaterials: Array<{
|
||
id: string
|
||
videoSourceUrl: string | null
|
||
videoCoverUrl: string | null
|
||
}> = []
|
||
|
||
try {
|
||
staleJobs = await videoTranscodeJob.findMany({
|
||
where: {
|
||
OR: [
|
||
{ status: 'queued' },
|
||
{ status: 'processing', lockedAt: null },
|
||
{ status: 'processing', lockedAt: { lt: staleBefore } },
|
||
],
|
||
material: {
|
||
videoSourceUrl: { not: null },
|
||
videoProcessingStatus: { in: ['pending', 'processing'] },
|
||
},
|
||
},
|
||
include: {
|
||
material: {
|
||
select: {
|
||
id: true,
|
||
videoSourceUrl: true,
|
||
videoCoverUrl: true,
|
||
},
|
||
},
|
||
},
|
||
take: RECOVERY_BATCH_SIZE,
|
||
orderBy: { updatedAt: 'asc' },
|
||
})
|
||
|
||
orphanMaterials = await prisma.$queryRaw<Array<{
|
||
id: string
|
||
videoSourceUrl: string | null
|
||
videoCoverUrl: string | null
|
||
}>>`
|
||
SELECT m.id, m."videoSourceUrl", m."videoCoverUrl"
|
||
FROM "materials" m
|
||
LEFT JOIN "video_transcode_jobs" j ON j."materialId" = m.id
|
||
WHERE (m."materialKind" = 'VIDEO' OR m."materialType" = '视频')
|
||
AND m."videoSourceUrl" IS NOT NULL
|
||
AND m."videoProcessingStatus" IN ('pending', 'processing')
|
||
AND j.id IS NULL
|
||
ORDER BY m."updatedAt" ASC
|
||
LIMIT ${RECOVERY_BATCH_SIZE}
|
||
`
|
||
} catch (error) {
|
||
if (isVideoTranscodeSchemaMissing(error)) {
|
||
console.warn('[Transcode] 跳过启动恢复：数据库缺少 video_transcode_jobs 表，请执行数据库迁移')
|
||
return { recovered: 0, hasMore: false, skipped: true }
|
||
}
|
||
|
||
throw error
|
||
}
|
||
|
||
let recovered = 0
|
||
|
||
for (const job of staleJobs) {
|
||
const sourceUrl = job.material?.videoSourceUrl || job.videoSourceUrl
|
||
if (!sourceUrl) continue
|
||
|
||
await videoTranscodeJob.update({
|
||
where: { materialId: job.materialId },
|
||
data: {
|
||
status: 'queued',
|
||
lockedAt: null,
|
||
errorMessage: null,
|
||
},
|
||
})
|
||
await prisma.material.update({
|
||
where: { id: job.materialId },
|
||
data: { videoProcessingStatus: 'pending' },
|
||
})
|
||
await enqueueVideoTranscodeJob({
|
||
materialId: job.materialId,
|
||
videoSourceUrl: sourceUrl,
|
||
needsCoverImage: !job.material?.videoCoverUrl,
|
||
})
|
||
recovered++
|
||
}
|
||
|
||
for (const material of orphanMaterials) {
|
||
if (!material.videoSourceUrl) continue
|
||
|
||
await enqueueVideoTranscodeJob({
|
||
materialId: material.id,
|
||
videoSourceUrl: material.videoSourceUrl,
|
||
needsCoverImage: !material.videoCoverUrl,
|
||
})
|
||
recovered++
|
||
}
|
||
|
||
if (recovered > 0) {
|
||
console.log(`[Transcode] Recovered ${recovered} queued/stale video transcode job(s)`)
|
||
}
|
||
|
||
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 }
|
||
}
|