From 45d1ad3adf1fb8769fb33bff55584f12f5b9c5cc Mon Sep 17 00:00:00 2001 From: WuYongJie Date: Mon, 25 May 2026 16:05:01 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8C=81=E4=B9=85=E5=8C=96=E8=A7=86=E9=A2=91?= =?UTF-8?q?=E8=BD=AC=E7=A0=81=E4=BB=BB=E5=8A=A1=E5=B9=B6=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../transcode/__tests__/route.test.ts | 211 +++++++++++ app/api/materials/transcode/route.ts | 171 ++++----- lib/__tests__/s3.test.ts | 339 ++++++++++++++++++ lib/__tests__/video-transcode-service.test.ts | 157 ++++++++ lib/s3.ts | 150 +++++++- lib/video-processor.ts | 42 ++- lib/video-transcode-queue.ts | 8 + lib/video-transcode-service.ts | 276 ++++++++++++++ .../migration.sql | 32 ++ prisma/schema.prisma | 22 ++ 10 files changed, 1289 insertions(+), 119 deletions(-) create mode 100644 app/api/materials/transcode/__tests__/route.test.ts create mode 100644 lib/__tests__/s3.test.ts create mode 100644 lib/__tests__/video-transcode-service.test.ts create mode 100644 lib/video-transcode-service.ts create mode 100644 prisma/migrations/20260525020000_add_video_transcode_jobs/migration.sql diff --git a/app/api/materials/transcode/__tests__/route.test.ts b/app/api/materials/transcode/__tests__/route.test.ts new file mode 100644 index 0000000..eaf7503 --- /dev/null +++ b/app/api/materials/transcode/__tests__/route.test.ts @@ -0,0 +1,211 @@ +import { NextRequest } from 'next/server' +import { GET, PATCH, POST } from '../route' + +jest.mock('@/lib/auth', () => ({ + authOptions: {}, +})) + +jest.mock('next-auth', () => ({ + getServerSession: jest.fn(), +})) + +jest.mock('@/lib/prisma', () => ({ + prisma: { + material: { + findUnique: jest.fn(), + update: jest.fn(), + }, + }, +})) + +jest.mock('@/lib/admin-permissions', () => ({ + isCurrentSuperAdmin: jest.fn(), +})) + +jest.mock('@/lib/video-transcode-service', () => ({ + enqueueVideoTranscodeJob: jest.fn(), + recoverStaleVideoTranscodeJobs: jest.fn(), +})) + +const { getServerSession } = require('next-auth') +const { prisma } = require('@/lib/prisma') +const { isCurrentSuperAdmin } = require('@/lib/admin-permissions') +const { enqueueVideoTranscodeJob, recoverStaleVideoTranscodeJobs } = require('@/lib/video-transcode-service') + +describe('/api/materials/transcode 权限边界', () => { + beforeEach(() => { + jest.resetAllMocks() + }) + + it('未登录时拒绝查询转码状态', async () => { + getServerSession.mockResolvedValue(null) + + const response = await GET( + new NextRequest('http://localhost/api/materials/transcode?materialId=material-1') + ) + const data = await response.json() + + expect(response.status).toBe(401) + expect(data.error).toBe('请先登录') + expect(prisma.material.findUnique).not.toHaveBeenCalled() + }) + + it('普通用户不能查询别人素材的转码状态', async () => { + getServerSession.mockResolvedValue({ user: { id: 'user-1', role: 'USER' } }) + prisma.material.findUnique.mockResolvedValue({ + id: 'material-1', + userId: 'user-2', + createdByUserId: 'user-2', + videoProcessingStatus: 'pending', + videoPreview540Url: null, + videoPreviewClipUrl: null, + videoCoverUrl: null, + videoDuration: null, + }) + + const response = await GET( + new NextRequest('http://localhost/api/materials/transcode?materialId=material-1') + ) + const data = await response.json() + + expect(response.status).toBe(403) + expect(data.error).toBe('无权查看该素材的转码状态') + }) + + it('素材上传者可以查询自己的转码状态', async () => { + getServerSession.mockResolvedValue({ user: { id: 'user-1', role: 'USER' } }) + prisma.material.findUnique.mockResolvedValue({ + id: 'material-1', + userId: 'user-2', + createdByUserId: 'user-1', + videoProcessingStatus: 'done', + videoPreview540Url: 'preview.mp4', + videoPreviewClipUrl: null, + videoCoverUrl: 'cover.jpg', + videoDuration: 60, + }) + + const response = await GET( + new NextRequest('http://localhost/api/materials/transcode?materialId=material-1') + ) + const data = await response.json() + + expect(response.status).toBe(200) + expect(data.status).toBe('done') + expect(data.hasPreview540).toBe(true) + }) + + it('未登录时拒绝触发转码任务', async () => { + getServerSession.mockResolvedValue(null) + + const response = await POST( + new NextRequest('http://localhost/api/materials/transcode', { + method: 'POST', + body: JSON.stringify({ materialId: 'material-1' }), + }) + ) + const data = await response.json() + + expect(response.status).toBe(401) + expect(data.error).toBe('请先登录') + expect(prisma.material.update).not.toHaveBeenCalled() + expect(enqueueVideoTranscodeJob).not.toHaveBeenCalled() + }) + + it('普通用户不能触发别人素材的转码任务', async () => { + getServerSession.mockResolvedValue({ user: { id: 'user-1', role: 'USER' } }) + prisma.material.findUnique.mockResolvedValue({ + id: 'material-1', + userId: 'user-2', + createdByUserId: 'user-2', + materialKind: 'VIDEO', + videoSourceUrl: 'source.mp4', + videoCoverUrl: null, + videoProcessingStatus: 'pending', + materialType: '视频', + }) + + const response = await POST( + new NextRequest('http://localhost/api/materials/transcode', { + method: 'POST', + body: JSON.stringify({ materialId: 'material-1' }), + }) + ) + const data = await response.json() + + expect(response.status).toBe(403) + expect(data.error).toBe('无权触发该素材的转码任务') + expect(prisma.material.update).not.toHaveBeenCalled() + expect(enqueueVideoTranscodeJob).not.toHaveBeenCalled() + }) + + it('管理员可以触发转码任务', async () => { + getServerSession.mockResolvedValue({ user: { id: 'admin-1', role: 'ADMIN' } }) + prisma.material.findUnique.mockResolvedValue({ + id: 'material-1', + userId: 'user-2', + createdByUserId: 'user-2', + materialKind: 'VIDEO', + videoSourceUrl: 'source.mp4', + videoCoverUrl: null, + videoProcessingStatus: 'pending', + materialType: '视频', + }) + prisma.material.update.mockResolvedValue({}) + enqueueVideoTranscodeJob.mockResolvedValue(undefined) + + const response = await POST( + new NextRequest('http://localhost/api/materials/transcode', { + method: 'POST', + body: JSON.stringify({ materialId: 'material-1' }), + }) + ) + const data = await response.json() + + expect(response.status).toBe(200) + expect(data.status).toBe('pending') + expect(enqueueVideoTranscodeJob).toHaveBeenCalledWith({ + materialId: 'material-1', + videoSourceUrl: 'source.mp4', + needsCoverImage: true, + forceRetry: false, + }) + }) + + it('普通管理员不能触发卡住转码任务恢复', async () => { + getServerSession.mockResolvedValue({ user: { id: 'admin-1', role: 'ADMIN' } }) + isCurrentSuperAdmin.mockResolvedValue(false) + + const response = await PATCH( + new NextRequest('http://localhost/api/materials/transcode', { + method: 'PATCH', + body: JSON.stringify({ action: 'recover' }), + }) + ) + const data = await response.json() + + expect(response.status).toBe(403) + expect(data.error).toBe('无权限') + expect(isCurrentSuperAdmin).toHaveBeenCalledWith({ user: { id: 'admin-1', role: 'ADMIN' } }) + expect(recoverStaleVideoTranscodeJobs).not.toHaveBeenCalled() + }) + + it('最高权限账号可以触发卡住转码任务恢复', async () => { + getServerSession.mockResolvedValue({ user: { id: 'super-1', role: 'ADMIN', isSuperAdmin: true } }) + isCurrentSuperAdmin.mockResolvedValue(true) + recoverStaleVideoTranscodeJobs.mockResolvedValue({ recovered: 2 }) + + const response = await PATCH( + new NextRequest('http://localhost/api/materials/transcode', { + method: 'PATCH', + body: JSON.stringify({ action: 'recover' }), + }) + ) + const data = await response.json() + + expect(response.status).toBe(200) + expect(data.recovered).toBe(2) + expect(isCurrentSuperAdmin).toHaveBeenCalledWith({ user: { id: 'super-1', role: 'ADMIN', isSuperAdmin: true } }) + expect(recoverStaleVideoTranscodeJobs).toHaveBeenCalled() + }) +}) diff --git a/app/api/materials/transcode/route.ts b/app/api/materials/transcode/route.ts index 01e5a1a..f5ca792 100644 --- a/app/api/materials/transcode/route.ts +++ b/app/api/materials/transcode/route.ts @@ -8,21 +8,40 @@ */ import { NextRequest, NextResponse } from 'next/server' +import { getServerSession } from 'next-auth' +import { authOptions } from '@/lib/auth' +import { isCurrentSuperAdmin } from '@/lib/admin-permissions' import { prisma } from '@/lib/prisma' -import { processVideoTranscode } from '@/lib/video-processor' -import { transcodeQueue } from '@/lib/video-transcode-queue' -import { tmpdir } from 'os' -import { join } from 'path' +import { enqueueVideoTranscodeJob, recoverStaleVideoTranscodeJobs } from '@/lib/video-transcode-service' export const dynamic = 'force-dynamic' export const maxDuration = 3600 // 1 hour timeout for transcode +function getSessionUser(session: unknown) { + return (session as any)?.user as { id?: string; role?: string } | undefined +} + +function canManageTranscode(session: unknown, material: { userId?: string | null; createdByUserId?: string | null }) { + const user = getSessionUser(session) + if (!user?.id) return false + if (user.role === 'ADMIN') return true + return material.userId === user.id || material.createdByUserId === user.id +} + /** * GET /api/materials/transcode?materialId=xxx * 查询视频转码状态 */ export async function GET(req: NextRequest) { try { + const session = await getServerSession(authOptions) + if (!getSessionUser(session)?.id) { + return NextResponse.json( + { error: '请先登录' }, + { status: 401 } + ) + } + const { searchParams } = new URL(req.url) const materialId = searchParams.get('materialId') @@ -37,6 +56,8 @@ export async function GET(req: NextRequest) { where: { id: materialId }, select: { id: true, + userId: true, + createdByUserId: true, videoProcessingStatus: true, videoPreview540Url: true, videoPreviewClipUrl: true, @@ -52,6 +73,13 @@ export async function GET(req: NextRequest) { ) } + if (!canManageTranscode(session, material)) { + return NextResponse.json( + { error: '无权查看该素材的转码状态' }, + { status: 403 } + ) + } + return NextResponse.json({ materialId: material.id, status: material.videoProcessingStatus || 'pending', @@ -75,6 +103,14 @@ export async function GET(req: NextRequest) { */ export async function POST(req: NextRequest) { try { + const session = await getServerSession(authOptions) + if (!getSessionUser(session)?.id) { + return NextResponse.json( + { error: '请先登录' }, + { status: 401 } + ) + } + const body = await req.json().catch(() => ({})) const { materialId } = body @@ -90,7 +126,8 @@ export async function POST(req: NextRequest) { where: { id: materialId }, select: { id: true, - // @ts-ignore - materialKind 字段已在 schema 中定义 + userId: true, + createdByUserId: true, materialKind: true, videoSourceUrl: true, videoCoverUrl: true, @@ -106,6 +143,13 @@ export async function POST(req: NextRequest) { ) } + if (!canManageTranscode(session, material)) { + return NextResponse.json( + { error: '无权触发该素材的转码任务' }, + { status: 403 } + ) + } + // 检查是否为视频素材 const materialKind = (material as any).materialKind || '' const isVideo = materialKind.toUpperCase() === 'VIDEO' || material.materialType === '视频' @@ -146,40 +190,17 @@ export async function POST(req: NextRequest) { console.log(`[Transcode] Force retry requested for material ${materialId}`) } - // 更新状态为 processing - await prisma.material.update({ - where: { id: materialId }, - data: { - videoProcessingStatus: 'processing', - }, + await enqueueVideoTranscodeJob({ + materialId, + videoSourceUrl: material.videoSourceUrl, + needsCoverImage: !material.videoCoverUrl, + forceRetry, }) - // 异步执行转码任务（不阻塞响应） - // needsCoverImage: 如果用户已经上传了封面图，则不需要从视频中截取 - const needsCoverImage = !material.videoCoverUrl - - // 使用队列管理转码任务，避免并发过多导致系统过载 - transcodeQueue.add(materialId, material.videoSourceUrl, needsCoverImage, processVideoTranscodeTask).catch( - (error) => { - console.error(`[Transcode] Task failed for material ${materialId}:`, error) - // 更新状态为 failed - prisma.material - .update({ - where: { id: materialId }, - data: { - videoProcessingStatus: 'failed', - }, - }) - .catch((dbError) => { - console.error(`[Transcode] Failed to update status to failed:`, dbError) - }) - } - ) - return NextResponse.json({ message: '转码任务已启动', materialId, - status: 'processing', + status: 'pending', }) } catch (error) { console.error('[Transcode] API error:', error) @@ -190,72 +211,34 @@ export async function POST(req: NextRequest) { } } -/** - * 执行转码任务（异步，带重试机制） - */ -async function processVideoTranscodeTask( - materialId: string, - videoSourceUrl: string, - needsCoverImage: boolean, - retryCount: number = 0 -): Promise { - const MAX_RETRIES = parseInt(process.env.VIDEO_TRANSCODE_MAX_RETRIES || '2', 10) - +export async function PATCH(req: NextRequest) { try { - console.log(`[Transcode] Starting transcode task for material ${materialId}${retryCount > 0 ? ` (retry ${retryCount}/${MAX_RETRIES})` : ''}...`) - - const outputDir = tmpdir() - const result = await processVideoTranscode({ - inputPath: videoSourceUrl, - outputDir, - materialId, - needsCoverImage, - }) - - // 更新数据库 - const updateData: any = { - videoProcessingStatus: 'done', - videoPreview540Url: result.preview540Url, - videoPreviewClipUrl: result.clipUrl, - videoDuration: result.duration, + const session = await getServerSession(authOptions) + if (!(await isCurrentSuperAdmin(session))) { + return NextResponse.json( + { error: '无权限' }, + { status: 403 } + ) } - // 如果用户没有上传封面图，使用自动截取的封面图 - if (needsCoverImage && result.coverUrl) { - updateData.videoCoverUrl = result.coverUrl - // 同时更新 previewUrl（如果当前是占位符） - const material = await prisma.material.findUnique({ - where: { id: materialId }, - select: { previewUrl: true }, - }) - if (material?.previewUrl?.startsWith('data:image')) { - updateData.previewUrl = result.coverUrl - } + const body = await req.json().catch(() => ({})) + if (body.action !== 'recover') { + return NextResponse.json( + { error: '不支持的操作' }, + { status: 400 } + ) } - await prisma.material.update({ - where: { id: materialId }, - data: updateData, + const result = await recoverStaleVideoTranscodeJobs() + return NextResponse.json({ + message: '已触发转码任务恢复', + ...result, }) - - console.log(`[Transcode] Transcode task completed for material ${materialId}`) } catch (error) { - console.error(`[Transcode] Transcode task failed for material ${materialId}:`, error) - - // 重试机制 - if (retryCount < MAX_RETRIES) { - const delay = Math.min(1000 * Math.pow(2, retryCount), 30000) // 指数退避，最多30秒 - console.log(`[Transcode] Retrying transcode task for material ${materialId} after ${delay}ms...`) - - await new Promise((resolve) => setTimeout(resolve, delay)) - - // 递归重试 - return processVideoTranscodeTask(materialId, videoSourceUrl, needsCoverImage, retryCount + 1) - } else { - // 达到最大重试次数，标记为失败 - console.error(`[Transcode] Max retries (${MAX_RETRIES}) reached for material ${materialId}, marking as failed`) - throw error - } + console.error('[Transcode] Recovery error:', error) + return NextResponse.json( + { error: '转码任务恢复失败' }, + { status: 500 } + ) } } - diff --git a/lib/__tests__/s3.test.ts b/lib/__tests__/s3.test.ts new file mode 100644 index 0000000..ed6fe70 --- /dev/null +++ b/lib/__tests__/s3.test.ts @@ -0,0 +1,339 @@ +/** + * S3 服务模块测试 + * 测试 lib/s3.ts 中的核心功能 + */ + +export {} + +// Mock AWS SDK +const mockSend = jest.fn(); +const mockGetSignedUrl = jest.fn(); + +jest.mock('@aws-sdk/client-s3', () => ({ + S3Client: jest.fn().mockImplementation(() => ({ + send: mockSend, + })), + PutObjectCommand: jest.fn().mockImplementation((params) => ({ ...params, _type: 'PutObject' })), + DeleteObjectCommand: jest.fn().mockImplementation((params) => ({ ...params, _type: 'DeleteObject' })), + GetObjectCommand: jest.fn().mockImplementation((params) => ({ ...params, _type: 'GetObject' })), + CreateBucketCommand: jest.fn().mockImplementation((params) => ({ ...params, _type: 'CreateBucket' })), + HeadBucketCommand: jest.fn().mockImplementation((params) => ({ ...params, _type: 'HeadBucket' })), + PutBucketPolicyCommand: jest.fn().mockImplementation((params) => ({ ...params, _type: 'PutBucketPolicy' })), +})); + +jest.mock('@aws-sdk/s3-request-presigner', () => ({ + getSignedUrl: mockGetSignedUrl, +})); + +// 保存原始环境变量 +const originalEnv = process.env; + +describe('s3', () => { + beforeEach(() => { + jest.clearAllMocks(); + // 重置模块缓存以重新加载环境变量 + jest.resetModules(); + // 设置测试环境变量 + process.env = { + ...originalEnv, + S3_ENDPOINT: 'http://localhost:9000', + S3_REGION: 'us-east-1', + S3_ACCESS_KEY_ID: 'test-access-key', + S3_SECRET_ACCESS_KEY: 'test-secret-key', + S3_BUCKET: 'test-bucket', + S3_PUBLIC_DOMAIN: 'http://localhost:9000/test-bucket', + }; + }); + + afterEach(() => { + process.env = originalEnv; + }); + + describe('isS3Url', () => { + it('应该正确识别 HTTP URL', async () => { + const { isS3Url } = await import('../s3'); + expect(isS3Url('http://example.com/file.jpg')).toBe(true); + expect(isS3Url('https://example.com/file.jpg')).toBe(true); + }); + + it('应该正确识别非 URL 字符串', async () => { + const { isS3Url } = await import('../s3'); + expect(isS3Url('path/to/file.jpg')).toBe(false); + expect(isS3Url('/absolute/path.jpg')).toBe(false); + }); + + it('应该处理空值', async () => { + const { isS3Url } = await import('../s3'); + expect(isS3Url(null)).toBe(false); + expect(isS3Url(undefined)).toBe(false); + expect(isS3Url('')).toBe(false); + }); + }); + + describe('extractS3Key', () => { + it('应该从完整 URL 提取 key', async () => { + const { extractS3Key } = await import('../s3'); + const key = extractS3Key('http://localhost:9000/test-bucket/path/to/file.jpg'); + expect(key).toBe('test-bucket/path/to/file.jpg'); + }); + + it('应该处理相对路径', async () => { + const { extractS3Key } = await import('../s3'); + expect(extractS3Key('path/to/file.jpg')).toBe('path/to/file.jpg'); + }); + + it('应该移除前导斜杠', async () => { + const { extractS3Key } = await import('../s3'); + expect(extractS3Key('/path/to/file.jpg')).toBe('path/to/file.jpg'); + expect(extractS3Key('///path/to/file.jpg')).toBe('path/to/file.jpg'); + }); + + it('应该抛出错误当路径为空', async () => { + const { extractS3Key } = await import('../s3'); + expect(() => extractS3Key('')).toThrow('Invalid S3 path'); + }); + }); + + describe('uploadToS3', () => { + beforeEach(() => { + mockSend.mockResolvedValue({}); + }); + + it('应该成功上传文件', async () => { + const { uploadToS3 } = await import('../s3'); + const buffer = Buffer.from('test content'); + + const result = await uploadToS3(buffer, 'test-file.txt', 'text/plain'); + + expect(result).toContain('test-file.txt'); + expect(mockSend).toHaveBeenCalled(); + }); + + it('默认上传时不应该设置 public-read ACL', async () => { + const { uploadToS3 } = await import('../s3'); + const buffer = Buffer.from('test content'); + + await uploadToS3(buffer, 'private-file.txt', 'text/plain'); + + const putObjectCall = mockSend.mock.calls.find((call) => call[0]?._type === 'PutObject'); + expect(putObjectCall?.[0]).not.toHaveProperty('ACL'); + }); + + it('显式开启 S3_SET_ACL 时才设置 public-read ACL', async () => { + process.env.S3_SET_ACL = '1'; + const { uploadToS3 } = await import('../s3'); + const buffer = Buffer.from('test content'); + + await uploadToS3(buffer, 'public-file.txt', 'text/plain'); + + const putObjectCall = mockSend.mock.calls.find((call) => call[0]?._type === 'PutObject'); + expect(putObjectCall?.[0].ACL).toBe('public-read'); + }); + + it('上传失败时应该抛出错误', async () => { + jest.resetModules(); + // 先让 ensureBucketExists 成功，然后让 PutObject 失败 + mockSend + .mockResolvedValueOnce({}) // HeadBucket 成功 + .mockRejectedValueOnce(new Error('Upload failed')); // PutObject 失败 + + const { uploadToS3 } = await import('../s3'); + const buffer = Buffer.from('test content'); + + await expect(uploadToS3(buffer, 'test-file.txt', 'text/plain')) + .rejects.toThrow('S3上传失败'); + }); + }); + + describe('uploadStreamToS3', () => { + beforeEach(() => { + mockSend.mockResolvedValue({}); + }); + + it('应该支持流式上传并传递 ContentLength', async () => { + const { Readable } = await import('stream'); + const { uploadStreamToS3 } = await import('../s3'); + const stream = Readable.from(['stream content']); + + const result = await uploadStreamToS3(stream, 'stream-file.mp4', 'video/mp4', 14); + + expect(result).toContain('stream-file.mp4'); + const putObjectCall = mockSend.mock.calls.find((call) => call[0]?._type === 'PutObject'); + expect(putObjectCall?.[0].Body).toBe(stream); + expect(putObjectCall?.[0].ContentLength).toBe(14); + expect(putObjectCall?.[0].ContentType).toBe('video/mp4'); + }); + }); + + describe('deleteFromS3', () => { + it('应该成功删除文件', async () => { + mockSend.mockResolvedValue({}); + + const { deleteFromS3 } = await import('../s3'); + await deleteFromS3('http://localhost:9000/test-bucket/path/to/file.jpg'); + + expect(mockSend).toHaveBeenCalled(); + }); + + it('删除失败时应该抛出错误', async () => { + mockSend.mockRejectedValue(new Error('Delete failed')); + + const { deleteFromS3 } = await import('../s3'); + + await expect(deleteFromS3('path/to/file.jpg')) + .rejects.toThrow('Delete failed'); + }); + }); + + describe('ensureBucketExists', () => { + it('bucket 存在时不应该创建', async () => { + mockSend.mockResolvedValue({}); + jest.resetModules(); + + const { ensureBucketExists } = await import('../s3'); + await ensureBucketExists(); + + // HeadBucket 调用 + expect(mockSend).toHaveBeenCalledTimes(1); + }); + + it('bucket 不存在时默认只创建私有 bucket', async () => { + const notFoundError = { name: 'NotFound', $metadata: { httpStatusCode: 404 } }; + mockSend + .mockRejectedValueOnce(notFoundError) // HeadBucket 失败 + .mockResolvedValueOnce({}); // CreateBucket 成功 + + jest.resetModules(); + const { ensureBucketExists } = await import('../s3'); + await ensureBucketExists(); + + expect(mockSend).toHaveBeenCalledTimes(2); + expect(mockSend.mock.calls.some((call) => call[0]?._type === 'PutBucketPolicy')).toBe(false); + }); + + it('bucket 不存在且显式开启 S3_SET_ACL 时才设置公开读取策略', async () => { + process.env.S3_SET_ACL = '1'; + const notFoundError = { name: 'NotFound', $metadata: { httpStatusCode: 404 } }; + mockSend + .mockRejectedValueOnce(notFoundError) + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({}); + + jest.resetModules(); + const { ensureBucketExists } = await import('../s3'); + await ensureBucketExists(); + + expect(mockSend).toHaveBeenCalledTimes(3); + expect(mockSend.mock.calls.some((call) => call[0]?._type === 'PutBucketPolicy')).toBe(true); + }); + + it('没有配置 S3_BUCKET 时应该跳过', async () => { + delete process.env.S3_BUCKET; + jest.resetModules(); + + const { ensureBucketExists } = await import('../s3'); + await ensureBucketExists(); + + expect(mockSend).not.toHaveBeenCalled(); + }); + }); + + describe('createPresignedDownloadUrl', () => { + it('应该生成预签名 URL', async () => { + mockGetSignedUrl.mockResolvedValue('https://signed-url.example.com/file.jpg'); + + const { createPresignedDownloadUrl } = await import('../s3'); + const url = await createPresignedDownloadUrl('path/to/file.jpg'); + + expect(url).toBe('https://signed-url.example.com/file.jpg'); + expect(mockGetSignedUrl).toHaveBeenCalled(); + }); + + it('应该支持自定义过期时间', async () => { + mockGetSignedUrl.mockResolvedValue('https://signed-url.example.com/file.jpg'); + + const { createPresignedDownloadUrl } = await import('../s3'); + await createPresignedDownloadUrl('path/to/file.jpg', { expiresIn: 3600 }); + + expect(mockGetSignedUrl).toHaveBeenCalledWith( + expect.anything(), + expect.anything(), + { expiresIn: 3600 } + ); + }); + }); + + describe('getS3ObjectStream', () => { + it('应该返回对象流信息', async () => { + mockSend.mockResolvedValue({ + Body: 'mock-stream', + ContentLength: 1024, + ContentType: 'image/jpeg', + AcceptRanges: 'bytes', + }); + + const { getS3ObjectStream } = await import('../s3'); + const result = await getS3ObjectStream('path/to/file.jpg'); + + expect(result.body).toBe('mock-stream'); + expect(result.contentLength).toBe(1024); + expect(result.contentType).toBe('image/jpeg'); + }); + + it('应该支持 Range 请求', async () => { + mockSend.mockResolvedValue({ + Body: 'mock-stream', + ContentLength: 512, + ContentRange: 'bytes 0-511/1024', + }); + + const { getS3ObjectStream } = await import('../s3'); + const result = await getS3ObjectStream('path/to/file.jpg', 'bytes=0-511'); + + expect(result.contentRange).toBe('bytes 0-511/1024'); + }); + }); +}); + +describe('S3 环境变量处理', () => { + beforeEach(() => { + jest.clearAllMocks(); + jest.resetModules(); + }); + + afterEach(() => { + process.env = originalEnv; + }); + + it('应该处理 localhost 域名替换', async () => { + process.env = { + ...originalEnv, + S3_ENDPOINT: 'http://localhost:9000', + S3_REGION: 'us-east-1', + S3_ACCESS_KEY_ID: 'test-key', + S3_SECRET_ACCESS_KEY: 'test-secret', + S3_BUCKET: 'test-bucket', + S3_PUBLIC_DOMAIN: 'http://127.0.0.1:9000/test-bucket', + SERVER_IP: '192.168.1.100', + }; + + const { extractS3Key } = await import('../s3'); + // 测试 extractS3Key 能正常工作 + expect(extractS3Key('path/to/file.jpg')).toBe('path/to/file.jpg'); + }); + + it('应该处理空的 S3_PUBLIC_DOMAIN', async () => { + process.env = { + ...originalEnv, + S3_ENDPOINT: 'http://localhost:9000', + S3_REGION: 'us-east-1', + S3_ACCESS_KEY_ID: 'test-key', + S3_SECRET_ACCESS_KEY: 'test-secret', + S3_BUCKET: 'test-bucket', + S3_PUBLIC_DOMAIN: '', + }; + + const { extractS3Key } = await import('../s3'); + expect(extractS3Key('path/to/file.jpg')).toBe('path/to/file.jpg'); + }); +}); diff --git a/lib/__tests__/video-transcode-service.test.ts b/lib/__tests__/video-transcode-service.test.ts new file mode 100644 index 0000000..b9b3562 --- /dev/null +++ b/lib/__tests__/video-transcode-service.test.ts @@ -0,0 +1,157 @@ +export {} + +jest.mock('@/lib/prisma', () => ({ + prisma: { + $queryRaw: jest.fn(), + material: { + findMany: jest.fn(), + findUnique: jest.fn(), + update: jest.fn(), + }, + videoTranscodeJob: { + upsert: jest.fn(), + findMany: jest.fn(), + update: jest.fn(), + }, + }, +})) + +jest.mock('@/lib/video-transcode-queue', () => ({ + transcodeQueue: { + add: jest.fn(), + }, +})) + +jest.mock('@/lib/video-processor', () => ({ + processVideoTranscode: jest.fn(), +})) + +const { prisma } = require('@/lib/prisma') +const { transcodeQueue } = require('@/lib/video-transcode-queue') +const { processVideoTranscode } = require('@/lib/video-processor') + +describe('video-transcode-service', () => { + beforeEach(() => { + jest.resetAllMocks() + jest.useRealTimers() + transcodeQueue.add.mockResolvedValue(undefined) + }) + + afterEach(() => { + jest.useRealTimers() + delete process.env.VIDEO_TRANSCODE_RETRY_BASE_DELAY_MS + }) + + it('enqueueVideoTranscodeJob 会先落持久化任务再加入内存队列', async () => { + const { enqueueVideoTranscodeJob } = require('../video-transcode-service') + + await enqueueVideoTranscodeJob({ + materialId: 'material-1', + videoSourceUrl: 's3/video.mp4', + needsCoverImage: true, + }) + + expect(prisma.videoTranscodeJob.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { materialId: 'material-1' }, + create: expect.objectContaining({ + materialId: 'material-1', + status: 'queued', + videoSourceUrl: 's3/video.mp4', + }), + }) + ) + expect(prisma.material.update).toHaveBeenCalledWith({ + where: { id: 'material-1' }, + data: { videoProcessingStatus: 'pending' }, + }) + expect(transcodeQueue.add).toHaveBeenCalled() + }) + + it('recoverStaleVideoTranscodeJobs 会恢复卡住的 processing 任务和孤儿 pending 视频', async () => { + const { recoverStaleVideoTranscodeJobs } = require('../video-transcode-service') + + prisma.videoTranscodeJob.findMany.mockResolvedValue([ + { + materialId: 'material-1', + videoSourceUrl: 's3/old.mp4', + material: { + id: 'material-1', + videoSourceUrl: 's3/old.mp4', + videoCoverUrl: null, + }, + }, + ]) + prisma.$queryRaw.mockResolvedValue([ + { + id: 'material-2', + videoSourceUrl: 's3/orphan.mp4', + videoCoverUrl: 's3/cover.jpg', + }, + ]) + + const result = await recoverStaleVideoTranscodeJobs() + + expect(result.recovered).toBe(2) + expect(prisma.videoTranscodeJob.update).toHaveBeenCalledWith({ + where: { materialId: 'material-1' }, + data: expect.objectContaining({ + status: 'queued', + lockedAt: null, + }), + }) + expect(prisma.videoTranscodeJob.upsert).toHaveBeenCalledTimes(2) + expect(transcodeQueue.add).toHaveBeenCalledTimes(2) + expect(prisma.$queryRaw.mock.calls[0][0].join('')).toContain(`OR m."materialType" = '视频'`) + }) + + it('任务失败但还有重试次数时，会重新写回 queued 并延迟重新入队', async () => { + jest.useFakeTimers() + const { enqueueVideoTranscodeJob } = require('../video-transcode-service') + process.env.VIDEO_TRANSCODE_RETRY_BASE_DELAY_MS = '10' + + prisma.videoTranscodeJob.upsert.mockResolvedValue({}) + prisma.material.update.mockResolvedValue({}) + prisma.videoTranscodeJob.update.mockResolvedValue({ + materialId: 'material-1', + attempts: 1, + maxAttempts: 3, + }) + processVideoTranscode.mockRejectedValue(new Error('ffmpeg failed')) + + let firstQueueRun: Promise | undefined + transcodeQueue.add.mockImplementationOnce(async ( + materialId: string, + videoSourceUrl: string, + needsCoverImage: boolean, + taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise + ) => { + firstQueueRun = expect(taskFn(materialId, videoSourceUrl, needsCoverImage)).rejects.toThrow('ffmpeg failed') + await firstQueueRun + }) + transcodeQueue.add.mockResolvedValueOnce(undefined) + + await enqueueVideoTranscodeJob({ + materialId: 'material-1', + videoSourceUrl: 's3/video.mp4', + needsCoverImage: true, + }) + + await firstQueueRun + + expect(prisma.videoTranscodeJob.update).toHaveBeenLastCalledWith({ + where: { materialId: 'material-1' }, + data: expect.objectContaining({ + status: 'queued', + lockedAt: null, + finishedAt: null, + errorMessage: 'ffmpeg failed', + }), + }) + + await jest.advanceTimersByTimeAsync(10) + + expect(transcodeQueue.add).toHaveBeenCalledTimes(2) + expect(transcodeQueue.add.mock.calls[1][0]).toBe('material-1') + }) +}) diff --git a/lib/s3.ts b/lib/s3.ts index 2f16a9b..ff0f356 100644 --- a/lib/s3.ts +++ b/lib/s3.ts @@ -1,5 +1,6 @@ -import { S3Client, PutObjectCommand, DeleteObjectCommand, GetObjectCommand } from '@aws-sdk/client-s3' +import { S3Client, PutObjectCommand, DeleteObjectCommand, GetObjectCommand, CreateBucketCommand, HeadBucketCommand, PutBucketPolicyCommand } from '@aws-sdk/client-s3' import { getSignedUrl } from '@aws-sdk/s3-request-presigner' +import type { Readable } from 'stream' const s3Client = new S3Client({ endpoint: process.env.S3_ENDPOINT, @@ -17,6 +18,17 @@ const s3Client = new S3Client({ maxAttempts: 10, // 失败重试10次，确保上传成功 }) +// Docker 内部网络 IP 段（需要重写为宿主机 IP） +const DOCKER_INTERNAL_PREFIXES = ['172.17.', '172.18.', '172.19.', '172.20.', '172.21.', '172.22.', '172.23.', '172.24.', '172.25.', '172.26.', '172.27.', '172.28.', '172.29.', '172.30.', '172.31.'] + +const isDockerInternalIP = (hostname: string): boolean => { + return DOCKER_INTERNAL_PREFIXES.some(prefix => hostname.startsWith(prefix)) +} + +const isLocalOrDockerIP = (hostname: string): boolean => { + return hostname === '127.0.0.1' || hostname === 'localhost' || isDockerInternalIP(hostname) +} + const normalizePublicDomain = () => { const raw = (process.env.S3_PUBLIC_DOMAIN || '').trim() if (!raw) return '' @@ -24,20 +36,22 @@ const normalizePublicDomain = () => { try { const url = new URL(raw) const host = url.hostname - if (host === '127.0.0.1' || host === 'localhost') { - // 尝试使用 SERVER_IP 或 APP_PUBLIC_URL 的主机名替换，避免浏览器访问本机导致 404 + if (isLocalOrDockerIP(host)) { + // 尝试使用 SERVER_IP 或 APP_PUBLIC_URL 的主机名替换，避免浏览器访问本机或 Docker 内部 IP 导致 404 const fallbackHost = process.env.SERVER_IP || (() => { const appUrl = process.env.APP_PUBLIC_URL || process.env.NEXTAUTH_URL || '' if (!appUrl) return '' try { - return new URL(appUrl).hostname + const appHostname = new URL(appUrl).hostname + // 确保 fallback 不是 Docker 内部 IP + return isLocalOrDockerIP(appHostname) ? '' : appHostname } catch { return '' } })() - if (fallbackHost) { + if (fallbackHost && !isLocalOrDockerIP(fallbackHost)) { url.hostname = fallbackHost } } @@ -64,6 +78,11 @@ const buildKeyPrefix = () => { const sanitizeKey = (value: string) => value.replace(/^\/+/, '') +function shouldUsePublicReadAcl() { + const value = (process.env.S3_SET_ACL || process.env.S3_PUBLIC_READ || '').trim().toLowerCase() + return value === '1' || value === 'true' || value === 'yes' +} + export function isS3Url(value?: string | null) { if (!value) return false return /^https?:\/\//.test(value) @@ -96,16 +115,30 @@ export async function uploadToS3( fileName: string, contentType: string ): Promise { + // 确保 bucket 存在 + await ensureBucketExists() + const keyPrefix = buildKeyPrefix() const key = `${keyPrefix}${fileName}` - const command = new PutObjectCommand({ + const putObjectInput: { + Bucket: string + Key: string + Body: Buffer + ContentType: string + ACL?: 'public-read' + } = { Bucket: process.env.S3_BUCKET!, Key: key, Body: file, ContentType: contentType, - ACL: 'public-read', - }) + } + + if (shouldUsePublicReadAcl()) { + putObjectInput.ACL = 'public-read' + } + + const command = new PutObjectCommand(putObjectInput) try { await s3Client.send(command) @@ -119,6 +152,51 @@ export async function uploadToS3( } } +export async function uploadStreamToS3( + stream: Readable, + fileName: string, + contentType: string, + contentLength?: number +): Promise { + await ensureBucketExists() + + const keyPrefix = buildKeyPrefix() + const key = `${keyPrefix}${fileName}` + + const putObjectInput: { + Bucket: string + Key: string + Body: Readable + ContentType: string + ContentLength?: number + ACL?: 'public-read' + } = { + Bucket: process.env.S3_BUCKET!, + Key: key, + Body: stream, + ContentType: contentType, + } + + if (Number.isFinite(contentLength) && contentLength && contentLength > 0) { + putObjectInput.ContentLength = contentLength + } + + if (shouldUsePublicReadAcl()) { + putObjectInput.ACL = 'public-read' + } + + const command = new PutObjectCommand(putObjectInput) + + try { + await s3Client.send(command) + const publicDomain = normalizePublicDomain() + return publicDomain ? `${publicDomain}/${key}` : key + } catch (error) { + console.error('[S3] Stream upload error (after all retries):', error) + throw new Error('S3上传失败，请稍后重试') + } +} + export async function deleteFromS3(fileUrl: string): Promise { try { const key = extractS3Key(fileUrl) @@ -137,6 +215,61 @@ export async function deleteFromS3(fileUrl: string): Promise { export { s3Client } +// 确保 bucket 存在，如果不存在则创建 +let bucketEnsured = false +export async function ensureBucketExists(): Promise { + if (bucketEnsured) return + + const bucket = process.env.S3_BUCKET + if (!bucket) { + console.warn('[S3] S3_BUCKET not configured, skipping bucket check') + return + } + + try { + // 检查 bucket 是否存在 + await s3Client.send(new HeadBucketCommand({ Bucket: bucket })) + console.log(`[S3] Bucket "${bucket}" exists`) + bucketEnsured = true + } catch (error: unknown) { + const err = error as { name?: string; $metadata?: { httpStatusCode?: number } } + // 如果 bucket 不存在，创建它 + if (err.name === 'NotFound' || err.$metadata?.httpStatusCode === 404) { + console.log(`[S3] Bucket "${bucket}" not found, creating...`) + try { + await s3Client.send(new CreateBucketCommand({ Bucket: bucket })) + console.log(`[S3] Bucket "${bucket}" created successfully`) + + if (shouldUsePublicReadAcl()) { + const policy = { + Version: '2012-10-17', + Statement: [ + { + Effect: 'Allow', + Principal: '*', + Action: ['s3:GetObject'], + Resource: [`arn:aws:s3:::${bucket}/*`] + } + ] + } + await s3Client.send(new PutBucketPolicyCommand({ + Bucket: bucket, + Policy: JSON.stringify(policy) + })) + console.log(`[S3] Bucket "${bucket}" policy set to public-read`) + } else { + console.log(`[S3] Bucket "${bucket}" created with private access`) + } + bucketEnsured = true + } catch (createError) { + console.error('[S3] Failed to create bucket:', createError) + } + } else { + console.error('[S3] Error checking bucket:', error) + } + } +} + type SignedUrlOptions = { expiresIn?: number responseContentDisposition?: string @@ -181,4 +314,3 @@ export async function getS3ObjectStream( acceptRanges: response.AcceptRanges, } } - diff --git a/lib/video-processor.ts b/lib/video-processor.ts index 5b15f16..2695d82 100644 --- a/lib/video-processor.ts +++ b/lib/video-processor.ts @@ -7,10 +7,12 @@ import { exec } from 'child_process' import { promisify } from 'util' -import { readFile, unlink, writeFile, stat } from 'fs/promises' +import { createReadStream, createWriteStream } from 'fs' +import { unlink, stat } from 'fs/promises' import { tmpdir } from 'os' import { join, resolve, isAbsolute } from 'path' -import { uploadToS3, getS3ObjectStream, isS3Url } from './s3' +import { pipeline } from 'stream/promises' +import { uploadStreamToS3, getS3ObjectStream, isS3Url } from './s3' import { Readable } from 'stream' const execAsync = promisify(exec) @@ -96,14 +98,8 @@ async function downloadVideoToTemp(s3Url: string, tempPath: string): Promise = new Set() + private queued: Set = new Set() private maxConcurrent: number constructor(maxConcurrent: number = 1) { @@ -37,6 +38,10 @@ class VideoTranscodeQueue { console.log(`[TranscodeQueue] Material ${materialId} is already processing, skipping`) return } + if (this.queued.has(materialId)) { + console.log(`[TranscodeQueue] Material ${materialId} is already queued, skipping`) + return + } return new Promise((resolve, reject) => { this.queue.push({ @@ -46,6 +51,7 @@ class VideoTranscodeQueue { resolve, reject, }) + this.queued.add(materialId) // 立即尝试处理队列 this.process(taskFn) @@ -74,6 +80,7 @@ class VideoTranscodeQueue { if (!task) { return } + this.queued.delete(task.materialId) console.log(`[TranscodeQueue] 开始处理任务: ${task.materialId} (队列剩余: ${this.queue.length})`) @@ -108,6 +115,7 @@ class VideoTranscodeQueue { queueLength: this.queue.length, processingCount: this.processing.size, processingIds: Array.from(this.processing), + queuedIds: Array.from(this.queued), maxConcurrent: this.maxConcurrent, } } diff --git a/lib/video-transcode-service.ts b/lib/video-transcode-service.ts new file mode 100644 index 0000000..84189af --- /dev/null +++ b/lib/video-transcode-service.ts @@ -0,0 +1,276 @@ +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 + +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 '未知错误' +} + +async function runVideoTranscodeTask( + materialId: string, + videoSourceUrl: string, + needsCoverImage: boolean +) { + const maxAttempts = getMaxAttempts() + + const job = await (prisma as any).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})`) + + 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 (prisma as any).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 (prisma as any).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() + + await (prisma as any).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 staleJobs = await (prisma as any).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: 20, + orderBy: { updatedAt: 'asc' }, + }) + + const orphanMaterials = await prisma.$queryRaw>` + 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 20 + ` + + let recovered = 0 + + for (const job of staleJobs) { + const sourceUrl = job.material?.videoSourceUrl || job.videoSourceUrl + if (!sourceUrl) continue + + await (prisma as any).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)`) + } + + return { recovered } +} diff --git a/prisma/migrations/20260525020000_add_video_transcode_jobs/migration.sql b/prisma/migrations/20260525020000_add_video_transcode_jobs/migration.sql new file mode 100644 index 0000000..353eda6 --- /dev/null +++ b/prisma/migrations/20260525020000_add_video_transcode_jobs/migration.sql @@ -0,0 +1,32 @@ +CREATE TABLE IF NOT EXISTS "video_transcode_jobs" ( + "id" TEXT NOT NULL, + "materialId" TEXT NOT NULL, + "videoSourceUrl" TEXT NOT NULL, + "status" TEXT NOT NULL DEFAULT 'queued', + "attempts" INTEGER NOT NULL DEFAULT 0, + "maxAttempts" INTEGER NOT NULL DEFAULT 3, + "needsCoverImage" BOOLEAN NOT NULL DEFAULT true, + "errorMessage" TEXT, + "lockedAt" TIMESTAMP(3), + "startedAt" TIMESTAMP(3), + "finishedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT "video_transcode_jobs_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX IF NOT EXISTS "video_transcode_jobs_materialId_key" +ON "video_transcode_jobs"("materialId"); + +CREATE INDEX IF NOT EXISTS "video_transcode_jobs_status_updatedAt_idx" +ON "video_transcode_jobs"("status", "updatedAt"); + +CREATE INDEX IF NOT EXISTS "video_transcode_jobs_lockedAt_idx" +ON "video_transcode_jobs"("lockedAt"); + +ALTER TABLE "video_transcode_jobs" +DROP CONSTRAINT IF EXISTS "video_transcode_jobs_materialId_fkey"; + +ALTER TABLE "video_transcode_jobs" +ADD CONSTRAINT "video_transcode_jobs_materialId_fkey" +FOREIGN KEY ("materialId") REFERENCES "materials"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 1c2c81d..b4d2d7c 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -224,6 +224,7 @@ model Material { logs MaterialLog[] browseHistory BrowseHistory[] shareTokens MaterialShareToken[] + videoTranscodeJob VideoTranscodeJob? template UploadTemplate? @relation(fields: [templateId], references: [id]) user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@ -241,6 +242,27 @@ model Material { @@map("materials") } +model VideoTranscodeJob { + id String @id @default(uuid()) + materialId String @unique + videoSourceUrl String + status String @default("queued") // queued / processing / done / failed + attempts Int @default(0) + maxAttempts Int @default(3) + needsCoverImage Boolean @default(true) + errorMessage String? + lockedAt DateTime? + startedAt DateTime? + finishedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + material Material @relation(fields: [materialId], references: [id], onDelete: Cascade) + + @@index([status, updatedAt]) + @@index([lockedAt]) + @@map("video_transcode_jobs") +} + model MaterialLike { id String @id @default(uuid()) materialId String