Juno ChowandClaude Fable 5.1
f665082dad
前端（Vite + React）、API/worker（Fastify + BullMQ）、api-contracts 契约包、 迁移脚本、测试与自托管 docker-compose 示例。 开源准备：运维脚本与机器信息已拆到私有仓库；公司名、备案号、隐私邮箱改读 publicCompliance.local.json（仓库只保留占位模板）；站点地址与客服二维码改为 VITE_ 环境变量；fixtures 中的真实用户标识已匿名化；三个 package.json 标注 Apache-2.0。 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
634 lines
28 KiB
TypeScript
634 lines
28 KiB
TypeScript
import { createHash } from 'node:crypto';
|
|
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
|
|
|
const mocks = vi.hoisted(() => ({
|
|
poolQuery: vi.fn(),
|
|
withTransaction: vi.fn(),
|
|
createUploadSignature: vi.fn(),
|
|
copyStoredObject: vi.fn(),
|
|
readStoredObjectSize: vi.fn(),
|
|
readUploadedMediaMetadata: vi.fn(),
|
|
validateUploadedMediaDecoding: vi.fn(),
|
|
createUploadedImageThumbnail: vi.fn(),
|
|
createUploadedVideoPreview: vi.fn(),
|
|
enqueueUploadSession: vi.fn(),
|
|
resolveAssetFolderId: vi.fn(),
|
|
scheduleObjectDeletion: vi.fn(),
|
|
mapAssetRow: vi.fn(),
|
|
}));
|
|
|
|
vi.mock('../src/db/pool.js', () => ({
|
|
pool: { query: mocks.poolQuery },
|
|
withTransaction: mocks.withTransaction,
|
|
}));
|
|
vi.mock('../src/storage/storageService.js', () => ({
|
|
UPLOAD_MAX_BYTES: 1024 * 1024 * 1024,
|
|
buildPublicUrl: (key: string) => `https://storage.example.com/${key}`,
|
|
buildUploadStagingObjectKey: ({ userId, sessionId, filename }: Record<string, string>) => `users/${userId}/upload-staging/${sessionId}/${filename}`,
|
|
buildUploadFinalObjectKey: ({ userId, sessionId, filename }: Record<string, string>) => `users/${userId}/uploads/${sessionId}/${filename}`,
|
|
createUploadSignature: mocks.createUploadSignature,
|
|
copyStoredObject: mocks.copyStoredObject,
|
|
readStoredObjectSize: mocks.readStoredObjectSize,
|
|
}));
|
|
vi.mock('../src/domain/uploadedVideoPreview.js', () => ({
|
|
buildUploadedImageThumbnailKey: (_userId: string, sessionId: string) => `thumb/${sessionId}.webp`,
|
|
buildUploadedVideoThumbnailKey: (_userId: string, sessionId: string) => `thumb/${sessionId}.jpg`,
|
|
createUploadedImageThumbnail: mocks.createUploadedImageThumbnail,
|
|
createUploadedVideoPreview: mocks.createUploadedVideoPreview,
|
|
readUploadedMediaMetadata: mocks.readUploadedMediaMetadata,
|
|
validateUploadedMediaDecoding: mocks.validateUploadedMediaDecoding,
|
|
}));
|
|
vi.mock('../src/queues/uploadQueue.js', () => ({ enqueueUploadSession: mocks.enqueueUploadSession }));
|
|
vi.mock('../src/domain/assetFolders.js', () => ({ resolveAssetFolderId: mocks.resolveAssetFolderId }));
|
|
vi.mock('../src/domain/assetDeletionService.js', () => ({ scheduleObjectDeletion: mocks.scheduleObjectDeletion }));
|
|
vi.mock('../src/domain/assets.js', () => ({ mapAssetRow: mocks.mapAssetRow }));
|
|
|
|
import {
|
|
createOrReplayUploadSession,
|
|
enqueueRecoverableUploadSessions,
|
|
processUploadSession,
|
|
queueUploadCompletion,
|
|
} from '../src/domain/uploadSessionService.js';
|
|
|
|
const input = {
|
|
filename: 'clip.mp4',
|
|
contentType: 'video/mp4',
|
|
sizeBytes: 1024,
|
|
source: 'uploaded-video' as const,
|
|
folderId: null,
|
|
idempotencyKey: 'upload-session-stable-key',
|
|
};
|
|
|
|
describe('uploadSessionService', () => {
|
|
beforeEach(() => {
|
|
Object.values(mocks).forEach(mock => mock.mockReset());
|
|
mocks.withTransaction.mockImplementation(async callback => callback({ query: mocks.poolQuery }));
|
|
mocks.createUploadSignature.mockImplementation(async (request: { objectKey: string }) => ({
|
|
driver: 'local', method: 'PUT', uploadUrl: 'http://localhost/upload', headers: { 'content-type': 'video/mp4' },
|
|
objectKey: request.objectKey,
|
|
expiresAt: '2099-01-01T00:00:00.000Z', maxSizeBytes: 1024,
|
|
}));
|
|
mocks.enqueueUploadSession.mockResolvedValue(undefined);
|
|
mocks.validateUploadedMediaDecoding.mockResolvedValue(undefined);
|
|
mocks.createUploadedImageThumbnail.mockImplementation(async ({ thumbnailKey }) => thumbnailKey);
|
|
mocks.createUploadedVideoPreview.mockImplementation(async ({ thumbnailKey }) => ({
|
|
thumbnailKey, width: 16, height: 16, durationSeconds: 1,
|
|
}));
|
|
});
|
|
|
|
it('同键同请求在 waiting-upload 时重签同一 staging key，不创建新会话', async () => {
|
|
const existing = sessionRow({ request_hash: requestHash(input) });
|
|
mocks.poolQuery
|
|
.mockResolvedValueOnce(queryResult([existing]))
|
|
.mockResolvedValueOnce(queryResult([{ ...existing, expires_at: '2099-01-01T00:00:00.000Z' }]));
|
|
|
|
const result = await createOrReplayUploadSession('user-1', input);
|
|
|
|
expect(result.signature).toEqual(expect.objectContaining({
|
|
objectKey: existing.object_key,
|
|
sessionId: existing.id,
|
|
}));
|
|
expect(mocks.createUploadSignature).toHaveBeenCalledWith(expect.objectContaining({ objectKey: existing.object_key }));
|
|
expect(mocks.resolveAssetFolderId).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it.each(['processing', 'completed'] as const)('%s 会话重放只返回 session，不重新签发写权限', async status => {
|
|
const existing = sessionRow({
|
|
request_hash: requestHash(input),
|
|
status,
|
|
processing_stage: status === 'completed' ? 'completed' : 'probing',
|
|
});
|
|
mocks.poolQuery.mockResolvedValueOnce(queryResult([existing]));
|
|
|
|
await expect(createOrReplayUploadSession('user-1', input)).resolves.toMatchObject({ signature: null });
|
|
expect(mocks.createUploadSignature).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('同幂等键请求事实变化时返回明确 409', async () => {
|
|
mocks.poolQuery.mockResolvedValueOnce(queryResult([sessionRow({ request_hash: 'different' })]));
|
|
await expect(createOrReplayUploadSession('user-1', input)).rejects.toMatchObject({
|
|
statusCode: 409,
|
|
code: 'UPLOAD_IDEMPOTENCY_CONFLICT',
|
|
});
|
|
});
|
|
|
|
it('complete 只把 waiting-upload 原子推进 queued，并用下一个 attempt 的稳定 job id 入队', async () => {
|
|
const pending = sessionRow();
|
|
mocks.poolQuery
|
|
.mockResolvedValueOnce(queryResult([pending]))
|
|
.mockResolvedValueOnce(queryResult([{ ...pending, processing_stage: 'queued' }]));
|
|
|
|
const result = await queueUploadCompletion('user-1', pending.id);
|
|
|
|
expect(result).toMatchObject({ status: 'pending', processingStage: 'queued' });
|
|
expect(mocks.enqueueUploadSession).toHaveBeenCalledWith(pending.id, 1);
|
|
});
|
|
|
|
it('上游交接使用同一事务，租约检查失败时不会修改上传状态或发布队列', async () => {
|
|
const leaseLost = Object.assign(new Error('导入租约失效'), { code: 'MCP_FILE_TIMEOUT' });
|
|
const beforeQueue = vi.fn(async (client) => {
|
|
expect(client.query).toBe(mocks.poolQuery);
|
|
throw leaseLost;
|
|
});
|
|
|
|
await expect(queueUploadCompletion('user-1', 'session-1', { beforeQueue })).rejects.toBe(leaseLost);
|
|
|
|
expect(beforeQueue).toHaveBeenCalledOnce();
|
|
expect(mocks.poolQuery).not.toHaveBeenCalled();
|
|
expect(mocks.enqueueUploadSession).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('已取消的交接不启动事务或发布队列', async () => {
|
|
const signal = AbortSignal.abort(new Error('worker stopping'));
|
|
await expect(queueUploadCompletion('user-1', 'session-1', { signal })).rejects.toBe(signal.reason);
|
|
expect(mocks.withTransaction).not.toHaveBeenCalled();
|
|
expect(mocks.enqueueUploadSession).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('等待会话锁期间取消，事务拒绝且不推进 queued', async () => {
|
|
const controller = new AbortController();
|
|
const reason = new Error('deadline exceeded');
|
|
const pending = sessionRow();
|
|
mocks.poolQuery.mockImplementationOnce(async () => {
|
|
controller.abort(reason);
|
|
return queryResult([pending]);
|
|
});
|
|
|
|
await expect(queueUploadCompletion('user-1', pending.id, { signal: controller.signal }))
|
|
.rejects.toBe(reason);
|
|
expect(mocks.poolQuery).toHaveBeenCalledOnce();
|
|
expect(mocks.enqueueUploadSession).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('入队状态写入期间取消仍让事务拒绝，不能在事务外发布队列', async () => {
|
|
const controller = new AbortController();
|
|
const reason = new Error('deadline exceeded');
|
|
const pending = sessionRow();
|
|
mocks.poolQuery.mockResolvedValueOnce(queryResult([pending])).mockImplementationOnce(async () => {
|
|
controller.abort(reason);
|
|
return queryResult([{ ...pending, processing_stage: 'queued' }]);
|
|
});
|
|
|
|
await expect(queueUploadCompletion('user-1', pending.id, { signal: controller.signal }))
|
|
.rejects.toBe(reason);
|
|
expect(mocks.enqueueUploadSession).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('complete 发现签名过期时先持久化 expired 与删除 Outbox，再在事务外返回 410', async () => {
|
|
const pending = sessionRow({ expires_at: '2000-01-01T00:00:00.000Z' });
|
|
const expired = sessionRow({
|
|
...pending,
|
|
status: 'expired',
|
|
processing_stage: 'expired',
|
|
error_code: 'UPLOAD_SESSION_EXPIRED',
|
|
error_message: '上传签名及宽限期已过期',
|
|
});
|
|
mocks.poolQuery
|
|
.mockResolvedValueOnce(queryResult([pending]))
|
|
.mockResolvedValueOnce(queryResult([expired]));
|
|
|
|
await expect(queueUploadCompletion('user-1', pending.id)).rejects.toMatchObject({
|
|
statusCode: 410,
|
|
code: 'UPLOAD_SESSION_EXPIRED',
|
|
});
|
|
expect(mocks.scheduleObjectDeletion).toHaveBeenCalledWith(
|
|
expect.anything(),
|
|
pending.object_key,
|
|
[],
|
|
expect.any(Date),
|
|
);
|
|
});
|
|
|
|
it('恢复器会将取消请求后崩溃且租约过期的 processing 会话终态化并安排清理', async () => {
|
|
const canceled = sessionRow({
|
|
status: 'canceled',
|
|
processing_stage: 'canceled',
|
|
cancel_requested_at: '2026-07-16T10:00:00.000Z',
|
|
error_code: 'UPLOAD_CANCELED',
|
|
error_message: '上传已取消',
|
|
});
|
|
mocks.poolQuery
|
|
.mockResolvedValueOnce(queryResult([canceled]))
|
|
.mockResolvedValueOnce(queryResult([]));
|
|
|
|
await expect(enqueueRecoverableUploadSessions(10)).resolves.toBe(0);
|
|
expect(String(mocks.poolQuery.mock.calls[0]?.[0])).toContain("SET status = 'canceled'");
|
|
expect(mocks.scheduleObjectDeletion).toHaveBeenCalledWith(
|
|
expect.anything(),
|
|
canceled.final_object_key,
|
|
[],
|
|
null,
|
|
);
|
|
});
|
|
|
|
it('完成资产与 session 的同一事务会为 staging key 写入延迟删除 Outbox', async () => {
|
|
const claimed = sessionRow({
|
|
source: 'uploaded-image', asset_type: 'image', content_type: 'image/png', filename: 'cover.png',
|
|
asset_name: 'cover', status: 'processing', processing_stage: 'sealing', attempt: 1,
|
|
lease_token: '00000000-0000-4000-8000-000000000099',
|
|
lease_expires_at: '2099-01-01T00:00:00.000Z',
|
|
folder_id: '00000000-0000-4000-8000-000000000055',
|
|
final_object_key: 'users/user-1/uploads/00000000-0000-4000-8000-000000000001/cover.png',
|
|
});
|
|
installSuccessfulProcessingQueries(claimed);
|
|
mocks.readStoredObjectSize.mockResolvedValue(1024);
|
|
mocks.readUploadedMediaMetadata.mockResolvedValue({
|
|
width: 1, height: 1, durationSeconds: null, hasVideo: true, hasAudio: false,
|
|
formatName: 'png_pipe', frameCount: 1, videoCodec: 'png', majorBrand: null,
|
|
});
|
|
|
|
await expect(processUploadSession(claimed.id)).resolves.toEqual({ skipped: false, assetId: 'asset-1' });
|
|
expect(mocks.resolveAssetFolderId).toHaveBeenCalledWith(expect.anything(), {
|
|
userId: claimed.user_id,
|
|
source: claimed.source,
|
|
folderId: claimed.folder_id,
|
|
});
|
|
const sessionLockIndex = mocks.poolQuery.mock.calls.findIndex(([sql]) => (
|
|
String(sql).includes("WHERE id = $1 AND status = 'processing'") && String(sql).includes('FOR UPDATE')
|
|
));
|
|
expect(sessionLockIndex).toBeGreaterThanOrEqual(0);
|
|
expect(mocks.resolveAssetFolderId.mock.invocationCallOrder[0]).toBeLessThan(
|
|
mocks.poolQuery.mock.invocationCallOrder[sessionLockIndex]!,
|
|
);
|
|
expect(mocks.validateUploadedMediaDecoding).toHaveBeenCalledWith({
|
|
objectKey: claimed.final_object_key,
|
|
assetType: 'image',
|
|
});
|
|
expect(mocks.scheduleObjectDeletion).toHaveBeenCalledWith(
|
|
expect.anything(),
|
|
claimed.object_key,
|
|
[],
|
|
expect.any(Date),
|
|
);
|
|
});
|
|
|
|
it('目录删除先提交后使用数据库最新主目录完成资产，而不使用 worker 的旧目录快照', async () => {
|
|
const oldFolderId = '00000000-0000-4000-8000-000000000055';
|
|
const rootFolderId = '00000000-0000-4000-8000-000000000066';
|
|
const claimed = audioProcessingSession({ folder_id: oldFolderId });
|
|
const locked = {
|
|
...claimed,
|
|
folder_id: rootFolderId,
|
|
asset_name: '数据库最新音频名称',
|
|
};
|
|
installSuccessfulAudioProcessingQueries(claimed, [locked], [locked]);
|
|
mocks.resolveAssetFolderId.mockResolvedValue(rootFolderId);
|
|
installSuccessfulAudioMetadata();
|
|
|
|
await expect(processUploadSession(claimed.id)).resolves.toEqual({ skipped: false, assetId: 'asset-1' });
|
|
|
|
expect(mocks.resolveAssetFolderId).toHaveBeenCalledTimes(1);
|
|
expect(mocks.resolveAssetFolderId).toHaveBeenCalledWith(expect.anything(), {
|
|
userId: claimed.user_id,
|
|
source: 'uploaded-audio',
|
|
folderId: rootFolderId,
|
|
});
|
|
const insert = mocks.poolQuery.mock.calls.find(([sql]) => String(sql).includes('INSERT INTO assets'));
|
|
expect(insert?.[1]?.slice(0, 6)).toEqual([
|
|
claimed.user_id,
|
|
rootFolderId,
|
|
'数据库最新音频名称',
|
|
'audio',
|
|
'uploaded-audio',
|
|
claimed.final_object_key,
|
|
]);
|
|
expect(insert?.[1]?.[1]).not.toBe(oldFolderId);
|
|
});
|
|
|
|
it('观察旧目录后目录已删除时跨事务重读新主目录并只插入一份资产', async () => {
|
|
const oldFolderId = '00000000-0000-4000-8000-000000000055';
|
|
const rootFolderId = '00000000-0000-4000-8000-000000000066';
|
|
const claimed = audioProcessingSession({ folder_id: oldFolderId });
|
|
const moved = { ...claimed, folder_id: rootFolderId };
|
|
installSuccessfulAudioProcessingQueries(claimed, [claimed, moved], [moved]);
|
|
mocks.resolveAssetFolderId
|
|
.mockRejectedValueOnce(Object.assign(new Error('旧目录已删除'), { code: 'ASSET_FOLDER_INVALID' }))
|
|
.mockResolvedValueOnce(rootFolderId);
|
|
installSuccessfulAudioMetadata();
|
|
|
|
await expect(processUploadSession(claimed.id)).resolves.toEqual({ skipped: false, assetId: 'asset-1' });
|
|
|
|
// claim 一次，持久化位置重试两次。
|
|
expect(mocks.withTransaction).toHaveBeenCalledTimes(3);
|
|
expect(mocks.resolveAssetFolderId).toHaveBeenCalledTimes(2);
|
|
expect(mocks.resolveAssetFolderId.mock.calls.map(([, input]) => input.folderId)).toEqual([
|
|
oldFolderId,
|
|
rootFolderId,
|
|
]);
|
|
const inserts = mocks.poolQuery.mock.calls.filter(([sql]) => String(sql).includes('INSERT INTO assets'));
|
|
expect(inserts).toHaveLength(1);
|
|
expect(inserts[0]?.[1]?.[1]).toBe(rootFolderId);
|
|
});
|
|
|
|
it('锁目录后发现会话位置变化会释放事务并有限重试', async () => {
|
|
const oldFolderId = '00000000-0000-4000-8000-000000000055';
|
|
const rootFolderId = '00000000-0000-4000-8000-000000000066';
|
|
const claimed = audioProcessingSession({ folder_id: oldFolderId });
|
|
const moved = { ...claimed, folder_id: rootFolderId };
|
|
installSuccessfulAudioProcessingQueries(claimed, [claimed, moved], [moved, moved]);
|
|
mocks.resolveAssetFolderId
|
|
.mockResolvedValueOnce(oldFolderId)
|
|
.mockResolvedValueOnce(rootFolderId);
|
|
installSuccessfulAudioMetadata();
|
|
|
|
await expect(processUploadSession(claimed.id)).resolves.toEqual({ skipped: false, assetId: 'asset-1' });
|
|
|
|
expect(mocks.withTransaction).toHaveBeenCalledTimes(3);
|
|
const inserts = mocks.poolQuery.mock.calls.filter(([sql]) => String(sql).includes('INSERT INTO assets'));
|
|
expect(inserts).toHaveLength(1);
|
|
expect(inserts[0]?.[1]?.[1]).toBe(rootFolderId);
|
|
});
|
|
|
|
it('缩略图 Put 成功但 session CAS 丢失时将 TOS 孤儿对象写入删除 Outbox', async () => {
|
|
const claimed = sessionRow({
|
|
source: 'uploaded-image', asset_type: 'image', content_type: 'image/png', filename: 'cover.png',
|
|
asset_name: 'cover', status: 'processing', processing_stage: 'sealing', attempt: 1,
|
|
lease_token: '00000000-0000-4000-8000-000000000099',
|
|
lease_expires_at: '2099-01-01T00:00:00.000Z',
|
|
final_object_key: 'users/user-1/uploads/00000000-0000-4000-8000-000000000001/cover.png',
|
|
});
|
|
const orphanThumbnailKey = `thumb/${claimed.id}.webp`;
|
|
mocks.poolQuery.mockImplementation(async (sql: unknown) => {
|
|
const query = String(sql);
|
|
if (query.includes("SET status = 'processing'")) return queryResult([claimed]);
|
|
if (query.includes('SELECT 1 FROM upload_sessions')) return queryResult([{ '?column?': 1 }]);
|
|
if (query.includes('SET sealed_at = COALESCE')) return queryResult([claimed]);
|
|
// 新 worker 已接管：旧 lease token 不得把缩略图 key 写入 session。
|
|
// 必须先匹配 SET 子句；该 SQL 的 WHERE 同样包含 processing_stage = 'thumbnailing'。
|
|
if (query.includes('SET thumbnail_key = $3')) return queryResult([]);
|
|
if (query.includes('SET processing_stage = $3') || query.includes("processing_stage = 'thumbnailing'")) {
|
|
return queryResult([claimed]);
|
|
}
|
|
if (query.includes('SET status = CASE')) return queryResult([]);
|
|
return queryResult([]);
|
|
});
|
|
mocks.readStoredObjectSize.mockResolvedValue(1024);
|
|
mocks.readUploadedMediaMetadata.mockResolvedValue({
|
|
width: 1, height: 1, durationSeconds: null, hasVideo: true, hasAudio: false,
|
|
formatName: 'png_pipe', frameCount: 1, videoCodec: 'png', majorBrand: null,
|
|
});
|
|
mocks.createUploadedImageThumbnail.mockResolvedValue(orphanThumbnailKey);
|
|
|
|
await expect(processUploadSession(claimed.id)).rejects.toMatchObject({ code: 'UPLOAD_LEASE_LOST' });
|
|
expect(mocks.scheduleObjectDeletion).toHaveBeenCalledWith(
|
|
expect.anything(),
|
|
orphanThumbnailKey,
|
|
[],
|
|
expect.any(Date),
|
|
);
|
|
});
|
|
|
|
it('音频会话拒绝包含视频流的伪装文件且不进入完整解码或资产事务', async () => {
|
|
const claimed = sessionRow({
|
|
filename: 'music.mp3', content_type: 'audio/mpeg', source: 'uploaded-audio', asset_type: 'audio',
|
|
status: 'processing', processing_stage: 'sealing', attempt: 1,
|
|
lease_token: '00000000-0000-4000-8000-000000000099',
|
|
lease_expires_at: '2099-01-01T00:00:00.000Z',
|
|
final_object_key: 'users/user-1/uploads/00000000-0000-4000-8000-000000000001/music.mp3',
|
|
});
|
|
installFailedProcessingQueries(claimed);
|
|
mocks.readStoredObjectSize.mockResolvedValue(1024);
|
|
mocks.readUploadedMediaMetadata.mockResolvedValue({
|
|
width: 16, height: 16, durationSeconds: 1, hasVideo: true, hasAudio: true,
|
|
formatName: 'mp3', frameCount: 1, videoCodec: 'h264', majorBrand: null,
|
|
});
|
|
|
|
await expect(processUploadSession(claimed.id)).rejects.toMatchObject({ code: 'UPLOAD_MEDIA_DETECTION_FAILED' });
|
|
expect(mocks.validateUploadedMediaDecoding).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('视频缩略图失败不会静默完成，且上传成功前不写入 thumbnail key', async () => {
|
|
const claimed = sessionRow({
|
|
status: 'processing', processing_stage: 'sealing', attempt: 1,
|
|
lease_token: '00000000-0000-4000-8000-000000000099',
|
|
lease_expires_at: '2099-01-01T00:00:00.000Z',
|
|
});
|
|
installFailedProcessingQueries(claimed, true);
|
|
mocks.readStoredObjectSize.mockResolvedValue(1024);
|
|
mocks.readUploadedMediaMetadata.mockResolvedValue({
|
|
width: 16, height: 16, durationSeconds: 1, hasVideo: true, hasAudio: false,
|
|
formatName: 'mov,mp4,m4a,3gp,3g2,mj2', frameCount: 24, videoCodec: 'h264', majorBrand: 'isom',
|
|
});
|
|
mocks.createUploadedVideoPreview.mockRejectedValue(Object.assign(new Error('thumbnail failed'), {
|
|
code: 'UPLOAD_THUMBNAIL_FAILED',
|
|
}));
|
|
|
|
await expect(processUploadSession(claimed.id)).rejects.toMatchObject({ code: 'UPLOAD_THUMBNAIL_FAILED' });
|
|
expect(mocks.poolQuery.mock.calls.some(([sql]) => (
|
|
String(sql).includes("processing_stage = 'thumbnailing'") && String(sql).includes('thumbnail_key = $3')
|
|
))).toBe(false);
|
|
expect(mocks.enqueueUploadSession).toHaveBeenCalledWith(claimed.id, 2);
|
|
});
|
|
|
|
it('媒体工具错误不把 TOS 签名 URL 写入会话或 BullMQ 失败原因', async () => {
|
|
const claimed = sessionRow({
|
|
status: 'processing', processing_stage: 'sealing', attempt: 1,
|
|
lease_token: '00000000-0000-4000-8000-000000000099',
|
|
lease_expires_at: '2099-01-01T00:00:00.000Z',
|
|
});
|
|
installFailedProcessingQueries(claimed, true);
|
|
mocks.readStoredObjectSize.mockResolvedValue(1024);
|
|
mocks.readUploadedMediaMetadata.mockResolvedValue({
|
|
width: 16, height: 16, durationSeconds: 1, hasVideo: true, hasAudio: false,
|
|
formatName: 'mov,mp4,m4a,3gp,3g2,mj2', frameCount: 24, videoCodec: 'h264', majorBrand: 'isom',
|
|
});
|
|
const signedUrl = 'https://tos.invalid/video.mp4?X-Tos-Signature=do-not-persist';
|
|
mocks.createUploadedVideoPreview.mockRejectedValue(Object.assign(
|
|
new Error(`ffmpeg failed for ${signedUrl}`),
|
|
{ code: 'UPLOAD_THUMBNAIL_FAILED' },
|
|
));
|
|
|
|
const failure = await processUploadSession(claimed.id).catch(error => error as Error & { code?: string });
|
|
expect(failure).toMatchObject({ code: 'UPLOAD_THUMBNAIL_FAILED', message: '缩略图生成失败，请重试' });
|
|
expect(JSON.stringify(failure)).not.toContain('X-Tos-Signature');
|
|
if (!(failure instanceof Error)) throw new Error('预期上传处理抛出安全错误');
|
|
expect(failure.message).not.toContain('tos.invalid');
|
|
|
|
const failedUpdate = mocks.poolQuery.mock.calls.find(([sql]) => String(sql).includes('SET status = CASE'));
|
|
expect(failedUpdate?.[1]?.[3]).toBe('缩略图生成失败，请重试');
|
|
expect(JSON.stringify(failedUpdate?.[1])).not.toContain('X-Tos-Signature');
|
|
expect(JSON.stringify(failedUpdate?.[1])).not.toContain('tos.invalid');
|
|
});
|
|
|
|
it('声明大小与 staging 实际大小不一致时永久失败，不复制也不自动重试', async () => {
|
|
const claimed = sessionRow({
|
|
status: 'processing', processing_stage: 'sealing', attempt: 1,
|
|
lease_token: '00000000-0000-4000-8000-000000000099',
|
|
lease_expires_at: '2099-01-01T00:00:00.000Z',
|
|
});
|
|
mocks.poolQuery.mockImplementation(async (sql: unknown) => {
|
|
const query = String(sql);
|
|
if (query.includes("SET status = 'processing'")) return queryResult([claimed]);
|
|
if (query.includes('SELECT 1 FROM upload_sessions')) return queryResult([{ '?column?': 1 }]);
|
|
if (query.includes('SET status = CASE')) return queryResult([{ ...claimed, status: 'failed', processing_stage: 'failed', retryable: false }]);
|
|
return queryResult([]);
|
|
});
|
|
mocks.readStoredObjectSize.mockResolvedValueOnce(1000);
|
|
|
|
await expect(processUploadSession(claimed.id)).rejects.toMatchObject({ code: 'UPLOAD_SIZE_MISMATCH' });
|
|
expect(mocks.copyStoredObject).not.toHaveBeenCalled();
|
|
expect(mocks.enqueueUploadSession).not.toHaveBeenCalled();
|
|
});
|
|
});
|
|
|
|
function requestHash(value: typeof input) {
|
|
return createHash('sha256').update(JSON.stringify({
|
|
filename: value.filename,
|
|
contentType: value.contentType,
|
|
sizeBytes: value.sizeBytes,
|
|
source: value.source,
|
|
folderId: value.folderId,
|
|
})).digest('hex');
|
|
}
|
|
|
|
function sessionRow(patch: Record<string, unknown> = {}) {
|
|
return {
|
|
id: '00000000-0000-4000-8000-000000000001',
|
|
user_id: 'user-1',
|
|
object_key: 'users/user-1/upload-staging/00000000-0000-4000-8000-000000000001/clip.mp4',
|
|
filename: 'clip.mp4',
|
|
content_type: 'video/mp4',
|
|
max_size_bytes: 1024,
|
|
status: 'pending',
|
|
asset_id: null,
|
|
expires_at: '2099-01-01T00:00:00.000Z',
|
|
completed_at: null,
|
|
updated_at: '2026-07-16T10:00:00.000Z',
|
|
idempotency_key: input.idempotencyKey,
|
|
request_hash: requestHash(input),
|
|
source: 'uploaded-video',
|
|
asset_type: 'video',
|
|
asset_name: 'clip',
|
|
folder_id: null as string | null,
|
|
final_object_key: 'users/user-1/uploads/00000000-0000-4000-8000-000000000001/clip.mp4',
|
|
thumbnail_key: null,
|
|
actual_size_bytes: null,
|
|
sealed_at: null,
|
|
sealed_size_bytes: null,
|
|
retryable: false,
|
|
attempt: 0,
|
|
lease_token: null,
|
|
lease_expires_at: null,
|
|
heartbeat_at: null,
|
|
processing_stage: 'waiting-upload',
|
|
error_code: null,
|
|
error_message: null,
|
|
cancel_requested_at: null,
|
|
retention_until: '2099-01-02T00:00:00.000Z',
|
|
...patch,
|
|
};
|
|
}
|
|
|
|
function queryResult(rows: unknown[]) {
|
|
return { rowCount: rows.length, rows };
|
|
}
|
|
|
|
function audioProcessingSession(patch: Record<string, unknown> = {}) {
|
|
return sessionRow({
|
|
filename: 'music.mp3',
|
|
content_type: 'audio/mpeg',
|
|
source: 'uploaded-audio',
|
|
asset_type: 'audio',
|
|
asset_name: 'music',
|
|
status: 'processing',
|
|
processing_stage: 'sealing',
|
|
attempt: 1,
|
|
lease_token: '00000000-0000-4000-8000-000000000099',
|
|
lease_expires_at: '2099-01-01T00:00:00.000Z',
|
|
folder_id: '00000000-0000-4000-8000-000000000055',
|
|
final_object_key: 'users/user-1/uploads/00000000-0000-4000-8000-000000000001/music.mp3',
|
|
...patch,
|
|
});
|
|
}
|
|
|
|
function installSuccessfulAudioMetadata() {
|
|
mocks.readStoredObjectSize.mockResolvedValue(1024);
|
|
mocks.readUploadedMediaMetadata.mockResolvedValue({
|
|
width: null,
|
|
height: null,
|
|
durationSeconds: 1,
|
|
hasVideo: false,
|
|
hasAudio: true,
|
|
formatName: 'mp3',
|
|
frameCount: null,
|
|
videoCodec: null,
|
|
majorBrand: null,
|
|
});
|
|
}
|
|
|
|
function installSuccessfulAudioProcessingQueries(
|
|
claimed: ReturnType<typeof sessionRow>,
|
|
observedRows: Array<ReturnType<typeof sessionRow>>,
|
|
lockedRows: Array<ReturnType<typeof sessionRow>>,
|
|
) {
|
|
let observedIndex = 0;
|
|
let lockedIndex = 0;
|
|
mocks.poolQuery.mockImplementation(async (sql: unknown) => {
|
|
const query = String(sql);
|
|
if (query.includes("SET status = 'processing'")) return queryResult([claimed]);
|
|
if (query.includes('SELECT 1 FROM upload_sessions')) return queryResult([{ '?column?': 1 }]);
|
|
if (query.includes('SET sealed_at = COALESCE')) return queryResult([claimed]);
|
|
if (query.includes('SET processing_stage = $3')) return queryResult([claimed]);
|
|
if (query.includes('SELECT user_id, source, folder_id')) {
|
|
return queryResult([observedRows[Math.min(observedIndex++, observedRows.length - 1)]!]);
|
|
}
|
|
if (query.includes('WHERE id = $1 AND status = \'processing\'') && query.includes('FOR UPDATE')) {
|
|
return queryResult([lockedRows[Math.min(lockedIndex++, lockedRows.length - 1)]!]);
|
|
}
|
|
if (query.includes('INSERT INTO assets')) return queryResult([{ id: 'asset-1' }]);
|
|
if (query.includes("SET status = 'completed'")) return queryResult([{ ...claimed, status: 'completed' }]);
|
|
return queryResult([]);
|
|
});
|
|
}
|
|
|
|
function installSuccessfulProcessingQueries(claimed: ReturnType<typeof sessionRow>) {
|
|
const persistedThumbnailKey = claimed.asset_type === 'image'
|
|
? `thumb/${claimed.id}.webp`
|
|
: claimed.asset_type === 'video'
|
|
? `thumb/${claimed.id}.jpg`
|
|
: claimed.thumbnail_key;
|
|
mocks.poolQuery.mockImplementation(async (sql: unknown) => {
|
|
const query = String(sql);
|
|
if (query.includes("SET status = 'processing'")) return queryResult([claimed]);
|
|
if (query.includes('SELECT 1 FROM upload_sessions')) return queryResult([{ '?column?': 1 }]);
|
|
if (query.includes('SET sealed_at = COALESCE')) return queryResult([claimed]);
|
|
if (query.includes('SET processing_stage = $3')) return queryResult([claimed]);
|
|
if (query.includes("SET processing_stage = 'thumbnailing'")) return queryResult([claimed]);
|
|
if (query.includes('SET thumbnail_key = $3')) return queryResult([{ ...claimed, thumbnail_key: persistedThumbnailKey }]);
|
|
if (query.includes('SELECT user_id, source, folder_id')) return queryResult([claimed]);
|
|
if (query.includes('WHERE id = $1 AND status = \'processing\'') && query.includes('FOR UPDATE')) {
|
|
return queryResult([{
|
|
...claimed,
|
|
thumbnail_key: persistedThumbnailKey,
|
|
sealed_at: '2026-07-16T10:00:00.000Z',
|
|
sealed_size_bytes: 1024,
|
|
}]);
|
|
}
|
|
if (query.includes('INSERT INTO assets')) return queryResult([{ id: 'asset-1' }]);
|
|
if (query.includes("SET status = 'completed'")) return queryResult([{ ...claimed, status: 'completed' }]);
|
|
return queryResult([]);
|
|
});
|
|
}
|
|
|
|
function installFailedProcessingQueries(claimed: ReturnType<typeof sessionRow>, retryAutomatically = false) {
|
|
mocks.poolQuery.mockImplementation(async (sql: unknown) => {
|
|
const query = String(sql);
|
|
if (query.includes("SET status = 'processing'")) return queryResult([claimed]);
|
|
if (query.includes('SELECT 1 FROM upload_sessions')) return queryResult([{ '?column?': 1 }]);
|
|
if (query.includes('SET sealed_at = COALESCE')) return queryResult([claimed]);
|
|
if (query.includes('SET processing_stage = $3') || query.includes("processing_stage = 'thumbnailing'")) {
|
|
return queryResult([claimed]);
|
|
}
|
|
if (query.includes('SET status = CASE')) {
|
|
return queryResult([{
|
|
...claimed,
|
|
status: retryAutomatically ? 'pending' : 'failed',
|
|
processing_stage: retryAutomatically ? 'queued' : 'failed',
|
|
retryable: false,
|
|
thumbnail_key: retryAutomatically ? `thumb/${claimed.id}.jpg` : claimed.thumbnail_key,
|
|
}]);
|
|
}
|
|
return queryResult([]);
|
|
});
|
|
}
|