498 lines
17 KiB
TypeScript
498 lines
17 KiB
TypeScript
// 视频转码任务封装，供上传接口和转码API复用
|
||
import { prisma } from '@/lib/prisma'
|
||
import { transcodeQueue } from '@/lib/video-transcode-queue'
|
||
import { processVideoTranscode, resolveLocalVideoPath } from '@/lib/video-processor'
|
||
import { access } from 'fs/promises'
|
||
|
||
export const USER_TRANSCODE_FAILURE_MESSAGE = '预览还没做好，请稍后重试'
|
||
import { deleteFileIfUnreferenced } from '@/lib/file-cleanup'
|
||
|
||
const MAX_TRANSCODE_RETRIES = 3
|
||
|
||
type CurrentLeaseResult<T> =
|
||
| { current: true; value: T }
|
||
| { current: false }
|
||
|
||
type LockedLeaseMaterial = {
|
||
videoCoverUrl: string | null
|
||
previewUrl: string
|
||
}
|
||
|
||
type TranscodeResult = Awaited<ReturnType<typeof processVideoTranscode>>
|
||
|
||
type CommittedTranscodeSnapshot = {
|
||
videoProcessingStatus: 'done'
|
||
videoPreview540Url: string
|
||
videoPreviewClipUrl: string
|
||
videoCoverUrl: string | null
|
||
previewUrl: string
|
||
videoDuration: number | null
|
||
transcodeLastError: null
|
||
transcodeAttempts: number
|
||
}
|
||
|
||
type ObservedTranscodeSnapshot = {
|
||
videoProcessingStatus: string | null
|
||
videoPreview540Url: string | null
|
||
videoPreviewClipUrl: string | null
|
||
videoCoverUrl: string | null
|
||
previewUrl: string
|
||
videoDuration: number | null
|
||
transcodeLastError: string | null
|
||
transcodeAttempts: number | null
|
||
}
|
||
|
||
type TranscodeFinalizeResult =
|
||
| { state: 'committed' }
|
||
| { state: 'not_committed'; error: unknown }
|
||
| { state: 'unknown'; error: unknown; observed?: ObservedTranscodeSnapshot | null }
|
||
|
||
/**
|
||
* 所有任务状态写入都以 PostgreSQL 的 clock_timestamp() 判定租约，不能使用
|
||
* 应用容器的本地时间。FOR UPDATE 同时把“校验 token”和后续写入收在一个事务里。
|
||
*/
|
||
async function withCurrentDatabaseLease<T>(
|
||
materialId: string,
|
||
leaseToken: string,
|
||
action: (tx: any, material: LockedLeaseMaterial) => Promise<T>,
|
||
): Promise<CurrentLeaseResult<T>> {
|
||
return prisma.$transaction(async (tx) => {
|
||
const rows = await tx.$queryRaw<Array<LockedLeaseMaterial>>`
|
||
SELECT "videoCoverUrl", "previewUrl"
|
||
FROM "materials"
|
||
WHERE "id" = ${materialId}
|
||
AND "transcode_lease_token" = ${leaseToken}
|
||
AND "transcode_lease_expires_at" > clock_timestamp()
|
||
FOR UPDATE
|
||
`
|
||
const material = rows[0]
|
||
if (!material) {
|
||
return { current: false } as const
|
||
}
|
||
return { current: true, value: await action(tx, material) } as const
|
||
})
|
||
}
|
||
|
||
class NonRetryableTranscodeError extends Error {
|
||
constructor(message: string) {
|
||
super(message)
|
||
this.name = 'NonRetryableTranscodeError'
|
||
}
|
||
}
|
||
|
||
export class TranscodeCommitStateUnknownError extends Error {
|
||
readonly originalError: unknown
|
||
|
||
constructor(materialId: string, originalError: unknown) {
|
||
super(`视频转码结果的数据库提交状态无法确认，已保留生成物等待恢复或人工对账: ${materialId}`)
|
||
this.name = 'TranscodeCommitStateUnknownError'
|
||
this.originalError = originalError
|
||
}
|
||
}
|
||
|
||
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('FFmpeg not available')
|
||
}
|
||
|
||
function generatedOutputUrls(result: TranscodeResult): string[] {
|
||
return Array.from(new Set([
|
||
result.preview540Url,
|
||
result.clipUrl,
|
||
result.coverUrl,
|
||
].filter((url): url is string => Boolean(url?.trim()))))
|
||
}
|
||
|
||
/**
|
||
* 任务拿到 processor 结果后，只有在明确未提交时才会调用这里。仍通过全库引用
|
||
* 复核再删，防止租约切换期间另一个任务或用户操作已经引用了相同 URL。
|
||
*/
|
||
export async function cleanupUncommittedTranscodeOutputs(
|
||
materialId: string,
|
||
result: TranscodeResult,
|
||
): Promise<void> {
|
||
const cleanupResults = await Promise.allSettled(
|
||
generatedOutputUrls(result).map((url) => deleteFileIfUnreferenced(url)),
|
||
)
|
||
|
||
for (const cleanupResult of cleanupResults) {
|
||
if (cleanupResult.status === 'rejected') {
|
||
console.error(
|
||
`[Transcode] Failed to clean an uncommitted output for material ${materialId}; manual reconciliation is required:`,
|
||
cleanupResult.reason,
|
||
)
|
||
} else if (!cleanupResult.value.success) {
|
||
console.error(
|
||
`[Transcode] Uncommitted output cleanup did not succeed for material ${materialId}:`,
|
||
cleanupResult.value.error,
|
||
)
|
||
}
|
||
}
|
||
}
|
||
|
||
function snapshotMatches(
|
||
observed: ObservedTranscodeSnapshot,
|
||
expected: CommittedTranscodeSnapshot,
|
||
): boolean {
|
||
return observed.videoProcessingStatus === expected.videoProcessingStatus
|
||
&& observed.videoPreview540Url === expected.videoPreview540Url
|
||
&& observed.videoPreviewClipUrl === expected.videoPreviewClipUrl
|
||
&& observed.videoCoverUrl === expected.videoCoverUrl
|
||
&& observed.previewUrl === expected.previewUrl
|
||
&& observed.videoDuration === expected.videoDuration
|
||
&& observed.transcodeLastError === expected.transcodeLastError
|
||
&& observed.transcodeAttempts === expected.transcodeAttempts
|
||
}
|
||
|
||
function observedReferencesGeneratedOutput(
|
||
observed: ObservedTranscodeSnapshot,
|
||
result: TranscodeResult,
|
||
): boolean {
|
||
const generatedUrls = new Set(generatedOutputUrls(result))
|
||
return [
|
||
observed.videoPreview540Url,
|
||
observed.videoPreviewClipUrl,
|
||
observed.videoCoverUrl,
|
||
observed.previewUrl,
|
||
].some((url) => Boolean(url && generatedUrls.has(url)))
|
||
}
|
||
|
||
async function readCommittedTranscodeSnapshot(
|
||
materialId: string,
|
||
): Promise<ObservedTranscodeSnapshot | null> {
|
||
return prisma.material.findUnique({
|
||
where: { id: materialId },
|
||
select: {
|
||
videoProcessingStatus: true,
|
||
videoPreview540Url: true,
|
||
videoPreviewClipUrl: true,
|
||
videoCoverUrl: true,
|
||
previewUrl: true,
|
||
videoDuration: true,
|
||
transcodeLastError: true,
|
||
transcodeAttempts: true,
|
||
},
|
||
})
|
||
}
|
||
|
||
async function finalizeTranscodeResult(
|
||
materialId: string,
|
||
leaseToken: string,
|
||
attempt: number,
|
||
result: TranscodeResult,
|
||
): Promise<TranscodeFinalizeResult> {
|
||
let expectedSnapshot: CommittedTranscodeSnapshot | null = null
|
||
|
||
try {
|
||
const completed = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, async (tx, lockedMaterial) => {
|
||
const currentCoverUrl = lockedMaterial.videoCoverUrl?.trim()
|
||
? lockedMaterial.videoCoverUrl
|
||
: null
|
||
const currentPreviewUrl = lockedMaterial.previewUrl?.trim()
|
||
? lockedMaterial.previewUrl
|
||
: null
|
||
const automaticCoverUrl = result.coverUrl || null
|
||
|
||
expectedSnapshot = {
|
||
videoProcessingStatus: 'done',
|
||
videoPreview540Url: result.preview540Url!,
|
||
videoPreviewClipUrl: result.clipUrl!,
|
||
videoCoverUrl: currentCoverUrl || automaticCoverUrl,
|
||
previewUrl: currentPreviewUrl || currentCoverUrl || automaticCoverUrl || '',
|
||
videoDuration: result.duration,
|
||
transcodeLastError: null,
|
||
transcodeAttempts: attempt,
|
||
}
|
||
|
||
return tx.material.updateMany({
|
||
where: { id: materialId, transcodeLeaseToken: leaseToken },
|
||
data: {
|
||
videoProcessingStatus: 'done',
|
||
videoPreview540Url: result.preview540Url,
|
||
videoPreviewClipUrl: result.clipUrl,
|
||
// 只填空值；事务锁保证判断与写入之间不会被并发更新穿透。
|
||
videoCoverUrl: currentCoverUrl ? undefined : automaticCoverUrl || undefined,
|
||
videoDuration: result.duration,
|
||
transcodeLastError: null,
|
||
transcodeAttempts: attempt,
|
||
previewUrl: currentPreviewUrl
|
||
? undefined
|
||
: currentCoverUrl || automaticCoverUrl || undefined,
|
||
},
|
||
})
|
||
})
|
||
|
||
if (!completed.current || completed.value.count !== 1) {
|
||
return {
|
||
state: 'not_committed',
|
||
error: new StaleTranscodeLeaseError(materialId),
|
||
}
|
||
}
|
||
|
||
return { state: 'committed' }
|
||
} catch (commitError) {
|
||
let observed: ObservedTranscodeSnapshot | null
|
||
try {
|
||
observed = await readCommittedTranscodeSnapshot(materialId)
|
||
} catch (verificationError) {
|
||
console.error(
|
||
`[Transcode] Cannot verify completion after database error for material ${materialId}; preserving outputs:`,
|
||
verificationError,
|
||
)
|
||
return { state: 'unknown', error: commitError }
|
||
}
|
||
|
||
if (expectedSnapshot && observed && snapshotMatches(observed, expectedSnapshot)) {
|
||
console.warn(
|
||
`[Transcode] Completion acknowledgement was lost for material ${materialId}, but the exact committed state was verified`,
|
||
)
|
||
return { state: 'committed' }
|
||
}
|
||
|
||
if (observed && observedReferencesGeneratedOutput(observed, result)) {
|
||
console.error(
|
||
`[Transcode] Completion state is partial or changed for material ${materialId}; preserving outputs for reconciliation`,
|
||
)
|
||
return { state: 'unknown', error: commitError, observed }
|
||
}
|
||
|
||
// PostgreSQL 事务已返回失败，且主库回读确认没有任何本次唯一 URL 被引用。
|
||
// 这是“明确未提交”；实际物理删除仍由 deleteFileIfUnreferenced 二次复核。
|
||
return { state: 'not_committed', error: commitError }
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 实际的转码任务处理函数
|
||
*
|
||
* 内部包含最多 3 次重试逻辑，队列外部无需处理 retry。
|
||
*/
|
||
export async function executeTranscodeTask(
|
||
materialId: string,
|
||
fallbackVideoSourceUrl: string,
|
||
needsCoverImage: boolean,
|
||
leaseToken: string,
|
||
abortSignal: AbortSignal = new AbortController().signal,
|
||
): Promise<void> {
|
||
let attempt = 0
|
||
const throwIfInterrupted = () => {
|
||
if (abortSignal.aborted) {
|
||
// 心跳、租约或 active marker 的状态无法继续证明时，当前执行结果必须
|
||
// 视为陈旧。这个分支绝不消耗重试次数，也绝不把可恢复任务写成 failed。
|
||
throw new StaleTranscodeLeaseError(materialId)
|
||
}
|
||
}
|
||
|
||
while (attempt < MAX_TRANSCODE_RETRIES) {
|
||
throwIfInterrupted()
|
||
attempt += 1
|
||
let attemptResult: TranscodeResult | null = null
|
||
let outputDisposition: 'none' | 'pending' | 'cleaned' | 'preserve' | 'committed' = 'none'
|
||
|
||
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(USER_TRANSCODE_FAILURE_MESSAGE)
|
||
}
|
||
|
||
// 入队时的 needsCoverImage 只是快照；真正开始处理时以数据库最新值为准，
|
||
// 避免排队期间用户已补手动封面却仍被自动截图覆盖。
|
||
const coverStillNeeded = !latestMaterial?.videoCoverUrl?.trim()
|
||
if (needsCoverImage !== coverStillNeeded) {
|
||
console.log(`[Transcode] Cover requirement changed while queued for material ${materialId}; using latest database value`)
|
||
}
|
||
|
||
const fileExists = await validateVideoFile(videoSourceUrl)
|
||
if (!fileExists) {
|
||
throw new Error(USER_TRANSCODE_FAILURE_MESSAGE)
|
||
}
|
||
|
||
const started = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, (tx) => tx.material.updateMany({
|
||
where: { id: materialId, transcodeLeaseToken: leaseToken },
|
||
data: { videoProcessingStatus: 'processing', transcodeLastError: null },
|
||
}))
|
||
if (!started.current || started.value.count !== 1) {
|
||
throw new StaleTranscodeLeaseError(materialId)
|
||
}
|
||
|
||
attemptResult = await processVideoTranscode({
|
||
inputPath: videoSourceUrl,
|
||
outputDir: './uploads/transcoded',
|
||
materialId,
|
||
needsCoverImage: coverStillNeeded,
|
||
abortSignal,
|
||
onProgress: (progress, stage) => {
|
||
console.log(`[Transcode] Material ${materialId} - ${stage}: ${progress}%`)
|
||
void transcodeQueue.updateProgress(materialId, leaseToken, progress, stage)
|
||
.catch((error) => console.error(`[Transcode] Failed to persist progress for ${materialId}:`, error))
|
||
},
|
||
})
|
||
outputDisposition = 'pending'
|
||
const result = attemptResult
|
||
|
||
throwIfInterrupted()
|
||
|
||
if (!result.preview540Url || !result.clipUrl || (coverStillNeeded && !result.coverUrl)) {
|
||
throw new NonRetryableTranscodeError('视频转码输出不完整，拒绝标记为完成')
|
||
}
|
||
|
||
// 锁住当前素材后只填空封面/首页预览。若事务响应丢失，必须精确回读全部
|
||
// 结果字段；只有 exact match 才当作成功，部分匹配一律保留对象等待对账。
|
||
const finalized = await finalizeTranscodeResult(materialId, leaseToken, attempt, result)
|
||
if (finalized.state === 'unknown') {
|
||
outputDisposition = 'preserve'
|
||
throw new TranscodeCommitStateUnknownError(materialId, finalized.error)
|
||
}
|
||
if (finalized.state === 'not_committed') {
|
||
await cleanupUncommittedTranscodeOutputs(materialId, result)
|
||
outputDisposition = 'cleaned'
|
||
throw finalized.error
|
||
}
|
||
outputDisposition = 'committed'
|
||
|
||
console.log(`[Transcode] Task completed for material ${materialId}`)
|
||
return
|
||
} catch (error) {
|
||
if (attemptResult && outputDisposition === 'pending') {
|
||
// processor 已成功返回，而完成事务尚未被确认；所有进入 unknown 的分支
|
||
// 都会先把 disposition 改为 preserve，因此这里仅覆盖明确未提交阶段。
|
||
await cleanupUncommittedTranscodeOutputs(materialId, attemptResult)
|
||
outputDisposition = 'cleaned'
|
||
}
|
||
if (error instanceof TranscodeCommitStateUnknownError) {
|
||
// 不能再写 failed 或重跑 FFmpeg；原事务可能已经提交。
|
||
throw error
|
||
}
|
||
if (abortSignal.aborted) {
|
||
throw new StaleTranscodeLeaseError(materialId)
|
||
}
|
||
if (error instanceof StaleTranscodeLeaseError) {
|
||
// 蓝绿切换或租约过期后，旧进程不得把成功/失败结果覆盖到新持有者。
|
||
throw 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) {
|
||
const failed = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, (tx) => tx.material.updateMany({
|
||
where: { id: materialId, transcodeLeaseToken: leaseToken },
|
||
data: {
|
||
videoProcessingStatus: 'failed',
|
||
transcodeLastError: USER_TRANSCODE_FAILURE_MESSAGE,
|
||
transcodeAttempts: attempt,
|
||
},
|
||
}))
|
||
if (!failed.current || failed.value.count !== 1) {
|
||
throw new StaleTranscodeLeaseError(materialId)
|
||
}
|
||
const finalError = error instanceof Error ? error : new Error(errorMsg)
|
||
throw finalError
|
||
}
|
||
|
||
const retryRecorded = await withCurrentDatabaseLease<{ count: number }>(materialId, leaseToken, (tx) => tx.material.updateMany({
|
||
where: { id: materialId, transcodeLeaseToken: leaseToken },
|
||
data: {
|
||
transcodeLastError: USER_TRANSCODE_FAILURE_MESSAGE,
|
||
transcodeAttempts: attempt,
|
||
},
|
||
}))
|
||
if (!retryRecorded.current || retryRecorded.value.count !== 1) {
|
||
throw new StaleTranscodeLeaseError(materialId)
|
||
}
|
||
|
||
console.warn(`[Transcode] Retrying material ${materialId} (${attempt}/${MAX_TRANSCODE_RETRIES})`)
|
||
}
|
||
}
|
||
}
|
||
|
||
export class StaleTranscodeLeaseError extends Error {
|
||
constructor(materialId: string) {
|
||
super(`视频转码租约已失效，拒绝写入旧进程结果: ${materialId}`)
|
||
this.name = 'StaleTranscodeLeaseError'
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 验证视频文件是否存在
|
||
*/
|
||
async function validateVideoFile(videoSourceUrl: string): Promise<boolean> {
|
||
try {
|
||
if (videoSourceUrl.startsWith('http://') || videoSourceUrl.startsWith('https://')) {
|
||
return true
|
||
}
|
||
|
||
const localFilePath = resolveLocalVideoPath(videoSourceUrl)
|
||
await access(localFilePath)
|
||
console.log(`[TranscodeTask] Video source is ready: ${localFilePath}`)
|
||
return true
|
||
} catch (error) {
|
||
console.error('[TranscodeTask] Video source is not readable', {
|
||
videoSourceUrl,
|
||
error,
|
||
})
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 入队一个视频转码任务
|
||
*/
|
||
export async function enqueueVideoTranscodeTask(
|
||
materialId: string,
|
||
videoSourceUrl: string,
|
||
needsCoverImage: boolean
|
||
): Promise<void> {
|
||
// 验证视频文件是否存在
|
||
const fileExists = await validateVideoFile(videoSourceUrl)
|
||
if (!fileExists) {
|
||
throw new Error(USER_TRANSCODE_FAILURE_MESSAGE)
|
||
}
|
||
|
||
console.log(`[TranscodeTask] 📝 Enqueuing transcode task: materialId=${materialId}, videoSourceUrl=${videoSourceUrl}, needsCoverImage=${needsCoverImage}`)
|
||
|
||
// 只等待数据库持久化入队，不等待 ffmpeg；失败需要反馈给调用方，不能静默丢任务。
|
||
await transcodeQueue.add(materialId, videoSourceUrl, needsCoverImage, executeTranscodeTask)
|
||
|
||
console.log(`[TranscodeTask] ✅ Transcode task enqueued successfully (fire-and-forget): materialId=${materialId}`)
|
||
}
|
||
|
||
|