持久化视频转码任务并支持恢复
This commit is contained in:
WuYongJie
parent
bd50dbd6e0
commit
45d1ad3adf
211
app/api/materials/transcode/__tests__/route.test.ts
Normal file
211
app/api/materials/transcode/__tests__/route.test.ts
Normal file
@ -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()
|
||||
})
|
||||
})
|
||||
@ -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<void> {
|
||||
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 }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
339
lib/__tests__/s3.test.ts
Normal file
339
lib/__tests__/s3.test.ts
Normal file
@ -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');
|
||||
});
|
||||
});
|
||||
157
lib/__tests__/video-transcode-service.test.ts
Normal file
157
lib/__tests__/video-transcode-service.test.ts
Normal file
@ -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<void> | undefined
|
||||
transcodeQueue.add.mockImplementationOnce(async (
|
||||
materialId: string,
|
||||
videoSourceUrl: string,
|
||||
needsCoverImage: boolean,
|
||||
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise<void>
|
||||
) => {
|
||||
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')
|
||||
})
|
||||
})
|
||||
150
lib/s3.ts
150
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<string> {
|
||||
// 确保 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<string> {
|
||||
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<void> {
|
||||
try {
|
||||
const key = extractS3Key(fileUrl)
|
||||
@ -137,6 +215,61 @@ export async function deleteFromS3(fileUrl: string): Promise<void> {
|
||||
|
||||
export { s3Client }
|
||||
|
||||
// 确保 bucket 存在，如果不存在则创建
|
||||
let bucketEnsured = false
|
||||
export async function ensureBucketExists(): Promise<void> {
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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<voi
|
||||
throw new Error('Failed to get video stream from S3')
|
||||
}
|
||||
|
||||
const chunks: Buffer[] = []
|
||||
const readable = stream.body as Readable
|
||||
|
||||
for await (const chunk of readable) {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
|
||||
}
|
||||
|
||||
await writeFile(tempPath, Buffer.concat(chunks))
|
||||
await pipeline(readable, createWriteStream(tempPath))
|
||||
}
|
||||
|
||||
/**
|
||||
@ -256,17 +252,27 @@ export async function processVideoTranscode(
|
||||
// 3. 转码为预览视频（分辨率可配置）
|
||||
console.log(`[VideoProcessor] Transcoding to ${TRANSCODE_CONFIG.previewResolution}P preview...`)
|
||||
await transcodeToPreview(tempInputPath, temp540Path)
|
||||
const preview540Buffer = await readFile(temp540Path)
|
||||
const preview540Stat = await stat(temp540Path)
|
||||
const preview540FileName = `video-preview-${TRANSCODE_CONFIG.previewResolution}p-${materialId}-${Date.now()}.mp4`
|
||||
const preview540Url = await uploadToS3(preview540Buffer, preview540FileName, 'video/mp4')
|
||||
const preview540Url = await uploadStreamToS3(
|
||||
createReadStream(temp540Path),
|
||||
preview540FileName,
|
||||
'video/mp4',
|
||||
preview540Stat.size
|
||||
)
|
||||
console.log(`[VideoProcessor] ${TRANSCODE_CONFIG.previewResolution}P preview uploaded:`, preview540Url)
|
||||
|
||||
// 4. 生成预览片段（时长可配置）
|
||||
console.log(`[VideoProcessor] Generating ${TRANSCODE_CONFIG.clipDuration}s clip...`)
|
||||
await generateClip(tempInputPath, tempClipPath)
|
||||
const clipBuffer = await readFile(tempClipPath)
|
||||
const clipStat = await stat(tempClipPath)
|
||||
const clipFileName = `video-clip-${TRANSCODE_CONFIG.clipDuration}s-${materialId}-${Date.now()}.mp4`
|
||||
const clipUrl = await uploadToS3(clipBuffer, clipFileName, 'video/mp4')
|
||||
const clipUrl = await uploadStreamToS3(
|
||||
createReadStream(tempClipPath),
|
||||
clipFileName,
|
||||
'video/mp4',
|
||||
clipStat.size
|
||||
)
|
||||
console.log(`[VideoProcessor] ${TRANSCODE_CONFIG.clipDuration}s clip uploaded:`, clipUrl)
|
||||
|
||||
// 5. 截取封面图（仅在需要时）
|
||||
@ -274,9 +280,14 @@ export async function processVideoTranscode(
|
||||
if (needsCoverImage) {
|
||||
console.log('[VideoProcessor] Extracting cover image...')
|
||||
await extractCoverImage(tempInputPath, tempCoverPath)
|
||||
const coverBuffer = await readFile(tempCoverPath)
|
||||
const coverStat = await stat(tempCoverPath)
|
||||
const coverFileName = `video-cover-${materialId}-${Date.now()}.jpg`
|
||||
coverUrl = await uploadToS3(coverBuffer, coverFileName, 'image/jpeg')
|
||||
coverUrl = await uploadStreamToS3(
|
||||
createReadStream(tempCoverPath),
|
||||
coverFileName,
|
||||
'image/jpeg',
|
||||
coverStat.size
|
||||
)
|
||||
console.log('[VideoProcessor] Cover image uploaded:', coverUrl)
|
||||
} else {
|
||||
console.log('[VideoProcessor] Skipping cover image extraction (user already uploaded)')
|
||||
@ -307,4 +318,3 @@ export async function processVideoTranscode(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -15,6 +15,7 @@ interface QueuedTask {
|
||||
class VideoTranscodeQueue {
|
||||
private queue: QueuedTask[] = []
|
||||
private processing: Set<string> = new Set()
|
||||
private queued: Set<string> = 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,
|
||||
}
|
||||
}
|
||||
|
||||
276
lib/video-transcode-service.ts
Normal file
276
lib/video-transcode-service.ts
Normal file
@ -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<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 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 }
|
||||
}
|
||||
@ -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;
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user