feat(api): upload verify, forbid-overwrite copy, idempotent complete
Task 19: POST /uploads/{id}/renew re-signs the SAME staging key for pending
sessions (uploadEnabled=false -> 409 ALBUM_UPLOAD_DISABLED; non-owner/unknown/
removed-member -> 403 AUTH_FORBIDDEN, indistinguishable per contract) and
slides the session expiry — renew is the documented recovery path for an
expired credential. POST /uploads/{id}/complete verifies the staging object
out-of-transaction first (head: missing / byte-count mismatch / contentType
mismatch — the server-side backstop for presign not signing content-type),
then copyIfAbsent's it to the finals key with forbid-overwrite semantics,
then in ONE transaction updates photo(bytes/uploadedAt/uploaded), claims the
session pending->confirmed via conditional updateMany (concurrent completes:
exactly one winner creates the photo_jobs row + upload.complete audit), and
enqueues QUEUE_PHOTO_PROCESS post-commit (jobId derived from photoId for
dedupe; failure only logs — Task 21 sweep compensates on photo_jobs). Verify
failures are request-level 400s: the session stays pending and retryable.
Repeat complete on a confirmed session re-runs copyIfAbsent (created:false,
staging already cleaned is fine) and returns the current processingState —
no second job/audit/enqueue (A03). GET /uploads lists the caller's own
sessions newest-first with (createdAt,id) keyset pagination and the joined
photo processingState/visibility. Both producer and test Queue instances now
quit their caller-owned ioredis connection (BullMQ does not take ownership),
otherwise jest hangs on open handles.
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -30,21 +30,22 @@ export class BullMqPhotoProcessQueue
|
|||||||
{
|
{
|
||||||
private readonly logger = new Logger(BullMqPhotoProcessQueue.name);
|
private readonly logger = new Logger(BullMqPhotoProcessQueue.name);
|
||||||
private queue: Queue | null = null;
|
private queue: Queue | null = null;
|
||||||
|
private connection: IORedis | null = null;
|
||||||
|
|
||||||
/** 首次使用时构造;连接错误经 no-op 监听吞掉(见文件头注释)。 */
|
/** 首次使用时构造;连接错误经 no-op 监听吞掉(见文件头注释)。 */
|
||||||
private getQueue(): Queue {
|
private getQueue(): Queue {
|
||||||
if (this.queue === null) {
|
if (this.queue === null) {
|
||||||
const url = process.env.REDIS_URL_CN ?? 'redis://localhost:6379';
|
const url = process.env.REDIS_URL_CN ?? 'redis://localhost:6379';
|
||||||
const connection = new IORedis(url, {
|
this.connection = new IORedis(url, {
|
||||||
// BullMQ 对阻塞命令的要求:禁用 per-request 重试上限(等待 Worker 语义)。
|
// BullMQ 对阻塞命令的要求:禁用 per-request 重试上限(等待 Worker 语义)。
|
||||||
maxRetriesPerRequest: null,
|
maxRetriesPerRequest: null,
|
||||||
});
|
});
|
||||||
connection.on('error', (err: Error) => {
|
this.connection.on('error', (err: Error) => {
|
||||||
this.logger.warn(
|
this.logger.warn(
|
||||||
`photo-process 队列连接异常(将由 ioredis 自动重连): ${err.message}`,
|
`photo-process 队列连接异常(将由 ioredis 自动重连): ${err.message}`,
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
this.queue = new Queue(QUEUE_PHOTO_PROCESS, { connection });
|
this.queue = new Queue(QUEUE_PHOTO_PROCESS, { connection: this.connection });
|
||||||
}
|
}
|
||||||
return this.queue;
|
return this.queue;
|
||||||
}
|
}
|
||||||
@@ -56,7 +57,8 @@ export class BullMqPhotoProcessQueue
|
|||||||
{ photoId },
|
{ photoId },
|
||||||
{
|
{
|
||||||
// jobId 以 photoId 派生:并发 complete/补偿扫描重复入队天然去重(A03)。
|
// jobId 以 photoId 派生:并发 complete/补偿扫描重复入队天然去重(A03)。
|
||||||
jobId: `${PHOTO_PROCESS_JOB_NAME}:${photoId}`,
|
// 注意 BullMQ 禁止 jobId 含 ':'(保留分隔符),用 '-' 连接。
|
||||||
|
jobId: `${PHOTO_PROCESS_JOB_NAME}-${photoId}`,
|
||||||
// Task 21 消费侧重试策略:3 次指数退避。
|
// Task 21 消费侧重试策略:3 次指数退避。
|
||||||
attempts: 3,
|
attempts: 3,
|
||||||
backoff: { type: 'exponential', delay: 1000 },
|
backoff: { type: 'exponential', delay: 1000 },
|
||||||
@@ -77,6 +79,12 @@ export class BullMqPhotoProcessQueue
|
|||||||
await this.queue.close().catch(() => undefined);
|
await this.queue.close().catch(() => undefined);
|
||||||
this.queue = null;
|
this.queue = null;
|
||||||
}
|
}
|
||||||
|
// BullMQ 不接管外部传入连接的生命周期:queue.close 不 disconnect 它,
|
||||||
|
// 这里显式断开,否则进程(含 e2e 的 app 实例)残留打开的 Redis 句柄。
|
||||||
|
if (this.connection !== null) {
|
||||||
|
await this.connection.quit().catch(() => undefined);
|
||||||
|
this.connection = null;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,18 +1,33 @@
|
|||||||
// services/api/src/uploads/uploads.controller.ts —— /uploads 端点(Task 18 起)。
|
// services/api/src/uploads/uploads.controller.ts —— /uploads 端点(Task 18 起;
|
||||||
// 方法名与契约 operationId 一一对应(initUpload;Task 19 增补 renewUploadCredential/
|
// Task 19 增补 listUploads/renewUploadCredential/completeUpload)。方法名与契约
|
||||||
// completeUpload/listUploads)。控制器级 SessionGuard:端点全需认证。职责:
|
// operationId 一一对应。控制器级 SessionGuard:端点全需认证。职责:透传认证
|
||||||
// 透传认证用户 + 查询参数规整(parseLimit/parseCursor 复用相册控制器导出)。
|
// 用户 + 查询参数规整(parseLimit/parseCursor 复用相册控制器导出,同契约 bounds)。
|
||||||
|
// 注意 DTO 必须值导入(非 import type):@Body 参数类型经 emitDecoratorMetadata
|
||||||
|
// 进入设计期元数据,import type 会令参数在管线中被置为 undefined(Task 18 实测)。
|
||||||
import {
|
import {
|
||||||
Body,
|
Body,
|
||||||
Controller,
|
Controller,
|
||||||
|
Get,
|
||||||
|
HttpCode,
|
||||||
|
HttpStatus,
|
||||||
|
Param,
|
||||||
Post,
|
Post,
|
||||||
|
Query,
|
||||||
Req,
|
Req,
|
||||||
UseGuards,
|
UseGuards,
|
||||||
} from '@nestjs/common';
|
} from '@nestjs/common';
|
||||||
import { userOf } from '../albums/albums.controller';
|
import {
|
||||||
|
parseCursor,
|
||||||
|
parseLimit,
|
||||||
|
userOf,
|
||||||
|
} from '../albums/albums.controller';
|
||||||
import { AuthenticatedRequest, SessionGuard } from '../auth/session.guard';
|
import { AuthenticatedRequest, SessionGuard } from '../auth/session.guard';
|
||||||
import { UploadsService } from './uploads.service';
|
import { UploadsService } from './uploads.service';
|
||||||
import type { UploadInitResponse } from './uploads.service';
|
import type {
|
||||||
|
UploadCompleteResponse,
|
||||||
|
UploadInitResponse,
|
||||||
|
UploadSessionListResponse,
|
||||||
|
} from './uploads.service';
|
||||||
import { UploadInitRequestDto } from './dto';
|
import { UploadInitRequestDto } from './dto';
|
||||||
|
|
||||||
@Controller('uploads')
|
@Controller('uploads')
|
||||||
@@ -28,4 +43,38 @@ export class UploadsController {
|
|||||||
): Promise<UploadInitResponse> {
|
): Promise<UploadInitResponse> {
|
||||||
return this.uploadsService.init(userOf(req), dto);
|
return this.uploadsService.init(userOf(req), dto);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** 上传历史:当前用户自己的会话,新在前,(createdAt,id) 键集游标。 */
|
||||||
|
@Get()
|
||||||
|
listUploads(
|
||||||
|
@Query('cursor') cursor: string | undefined,
|
||||||
|
@Query('limit') limit: string | undefined,
|
||||||
|
@Req() req: AuthenticatedRequest,
|
||||||
|
): Promise<UploadSessionListResponse> {
|
||||||
|
return this.uploadsService.list(
|
||||||
|
userOf(req),
|
||||||
|
parseCursor(cursor),
|
||||||
|
parseLimit(limit),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 续签:仅 pending 会话;返回结构同 initUpload(credential 为新值)。 */
|
||||||
|
@Post(':uploadId/renew')
|
||||||
|
@HttpCode(HttpStatus.OK)
|
||||||
|
renewUploadCredential(
|
||||||
|
@Param('uploadId') uploadId: string,
|
||||||
|
@Req() req: AuthenticatedRequest,
|
||||||
|
): Promise<UploadInitResponse> {
|
||||||
|
return this.uploadsService.renew(userOf(req), uploadId);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 完成核验:幂等(重复调用返回相同结果,不重复建 job)。 */
|
||||||
|
@Post(':uploadId/complete')
|
||||||
|
@HttpCode(HttpStatus.OK)
|
||||||
|
completeUpload(
|
||||||
|
@Param('uploadId') uploadId: string,
|
||||||
|
@Req() req: AuthenticatedRequest,
|
||||||
|
): Promise<UploadCompleteResponse> {
|
||||||
|
return this.uploadsService.complete(userOf(req), uploadId);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
// services/api/src/uploads/uploads.service.ts —— 上传域核心逻辑(Task 18 起)。
|
// services/api/src/uploads/uploads.service.ts —— 上传域核心逻辑(Task 18 起,
|
||||||
// 契约权威:packages/contracts/openapi.yaml tags [uploads]。要点:
|
// Task 19 增补 renew/complete/list)。契约权威:packages/contracts/openapi.yaml
|
||||||
|
// tags [uploads]。要点:
|
||||||
// · initUpload 幂等初始化(A03):@@unique([userId, clientTaskId]) 为幂等锚,
|
// · initUpload 幂等初始化(A03):@@unique([userId, clientTaskId]) 为幂等锚,
|
||||||
// 同键同参返回原会话(201,不重复建照片);同键异参 409 UPLOAD_SESSION_CONFLICT。
|
// 同键同参返回原会话(201,不重复建照片);同键异参 409 UPLOAD_SESSION_CONFLICT。
|
||||||
// 并发同键插入由 P2002 捕获后回读既有会话裁决(MySQL 约束冲突不中止事务)。
|
// 并发同键插入由 P2002 捕获后回读既有会话裁决(MySQL 约束冲突不中止事务)。
|
||||||
@@ -13,10 +14,26 @@
|
|||||||
// · init 不落 audit(高频低风险;审计锚点在 complete,A06 前置约定)。
|
// · init 不落 audit(高频低风险;审计锚点在 complete,A06 前置约定)。
|
||||||
// · UPLOAD_MAX_BYTES/SIGNED_URL_TTL_SECONDS 逐请求惰性读取(support),
|
// · UPLOAD_MAX_BYTES/SIGNED_URL_TTL_SECONDS 逐请求惰性读取(support),
|
||||||
// presign 在事务外执行(SigV4 为本地计算,无网络副作用)。
|
// presign 在事务外执行(SigV4 为本地计算,无网络副作用)。
|
||||||
import { Inject, Injectable } from '@nestjs/common';
|
// · renew(Task 19):仅 pending 会话可续签;凭证过期→renew 是契约规定的恢复
|
||||||
|
// 路径,故 renew 同时滑动顺延会话 expiresAt;相册关闭上传即拒签(§4.9)。
|
||||||
|
// · complete(Task 19):核验失败(head staging:缺失/字节数不符/contentType
|
||||||
|
// 与申明不符——presign 不签名 content-type,Task 11 发现的服务端兜底)是
|
||||||
|
// 请求级 400,会话保持 pending 可重试;通过后 copyIfAbsent(禁止覆盖语义)
|
||||||
|
// 受控复制到正式键,再进事务:photo(bytes/uploadedAt/uploaded) + 会话抢占
|
||||||
|
// 置 confirmed(updateMany 条件更新,并发 complete 恰一人胜出)+ 胜者独占
|
||||||
|
// photo_jobs upsert 与 audit;提交后入队 photo-process(失败仅日志,Task 21
|
||||||
|
// sweep 以 photo_jobs(pending) 为事实源补偿)。confirmed 会话重复 complete
|
||||||
|
// 走幂等路径:copyIfAbsent 复核(originals 已在则 created:false,staging 已
|
||||||
|
// 被清理也不受影响)后直接回显当前 processingState——不重复 job/audit/入队。
|
||||||
|
// · 权限逐次校验(A02):renew/complete 每次调用都重查成员权——被移除成员
|
||||||
|
// 的下一次请求即刻 403(会话/照片 ID 均已持有也不放行)。
|
||||||
|
import { Inject, Injectable, HttpStatus } from '@nestjs/common';
|
||||||
|
import { ErrorCodes } from '@photo/contracts';
|
||||||
import type { components } from '@photo/contracts';
|
import type { components } from '@photo/contracts';
|
||||||
import type { StoragePort } from '@photo/storage';
|
import { StorageError, type StoragePort } from '@photo/storage';
|
||||||
|
import type { Prisma } from '@photo/db-cn';
|
||||||
import type { AuthUser } from '../auth/session.guard';
|
import type { AuthUser } from '../auth/session.guard';
|
||||||
|
import { ApiException } from '../api-exception';
|
||||||
import { PermissionsService } from '../albums/permissions.service';
|
import { PermissionsService } from '../albums/permissions.service';
|
||||||
import {
|
import {
|
||||||
albumForbidden,
|
albumForbidden,
|
||||||
@@ -25,7 +42,16 @@ import {
|
|||||||
validationFailed,
|
validationFailed,
|
||||||
} from '../albums/support';
|
} from '../albums/support';
|
||||||
import { PrismaService } from '../prisma/prisma.service';
|
import { PrismaService } from '../prisma/prisma.service';
|
||||||
|
import {
|
||||||
|
decodeKeysetCursor,
|
||||||
|
encodeKeysetCursor,
|
||||||
|
keysetWhereDesc,
|
||||||
|
} from '../shared/cursor';
|
||||||
import { STORAGE_PORT } from './storage.provider';
|
import { STORAGE_PORT } from './storage.provider';
|
||||||
|
import {
|
||||||
|
PHOTO_PROCESS_QUEUE,
|
||||||
|
type PhotoProcessQueueProducer,
|
||||||
|
} from './queue.provider';
|
||||||
import type { UploadInitRequestDto } from './dto';
|
import type { UploadInitRequestDto } from './dto';
|
||||||
import {
|
import {
|
||||||
albumUploadDisabled,
|
albumUploadDisabled,
|
||||||
@@ -37,10 +63,17 @@ import {
|
|||||||
stagingObjectKey,
|
stagingObjectKey,
|
||||||
uploadMaxBytes,
|
uploadMaxBytes,
|
||||||
uploadSessionConflict,
|
uploadSessionConflict,
|
||||||
|
uploadSessionForbidden,
|
||||||
|
uploadVerifyFailed,
|
||||||
} from './support';
|
} from './support';
|
||||||
|
|
||||||
export type UploadInitResponse =
|
export type UploadInitResponse =
|
||||||
components['schemas']['UploadInitResponse'];
|
components['schemas']['UploadInitResponse'];
|
||||||
|
export type UploadCompleteResponse =
|
||||||
|
components['schemas']['UploadCompleteResponse'];
|
||||||
|
export type UploadSessionListResponse =
|
||||||
|
components['schemas']['UploadSessionList'];
|
||||||
|
type UploadSessionRecord = components['schemas']['UploadSessionRecord'];
|
||||||
|
|
||||||
type SessionRow = {
|
type SessionRow = {
|
||||||
id: string;
|
id: string;
|
||||||
@@ -51,12 +84,16 @@ type SessionRow = {
|
|||||||
albumId: string;
|
albumId: string;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
type UploadSessionRow = Prisma.upload_sessionsGetPayload<Prisma.upload_sessionsDefaultArgs>;
|
||||||
|
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class UploadsService {
|
export class UploadsService {
|
||||||
constructor(
|
constructor(
|
||||||
private readonly prisma: PrismaService,
|
private readonly prisma: PrismaService,
|
||||||
private readonly permissions: PermissionsService,
|
private readonly permissions: PermissionsService,
|
||||||
@Inject(STORAGE_PORT) private readonly storage: StoragePort,
|
@Inject(STORAGE_PORT) private readonly storage: StoragePort,
|
||||||
|
@Inject(PHOTO_PROCESS_QUEUE)
|
||||||
|
private readonly queue: PhotoProcessQueueProducer,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -202,4 +239,277 @@ export class UploadsService {
|
|||||||
extOfObjectKey(session.objectKey) === ext
|
extOfObjectKey(session.objectKey) === ext
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 会话定位 + 归属 + 逐次权限核验(renew/complete 共用)。任何一项不过都抛
|
||||||
|
* 同一 403 AUTH_FORBIDDEN(契约:无权/不存在不区分,防 uploadId 枚举)。
|
||||||
|
* 返回值保证非空且 userId === user.id。
|
||||||
|
*/
|
||||||
|
private async locateAuthorizedSession(
|
||||||
|
user: AuthUser,
|
||||||
|
uploadId: string,
|
||||||
|
): Promise<UploadSessionRow> {
|
||||||
|
const session = await this.prisma.upload_sessions.findUnique({
|
||||||
|
where: { id: uploadId },
|
||||||
|
});
|
||||||
|
if (session === null || session.userId !== user.id) {
|
||||||
|
throw uploadSessionForbidden();
|
||||||
|
}
|
||||||
|
// A02 逐次校验:被移除成员/册消失后的下一次 renew/complete 即刻拦截。
|
||||||
|
if (!(await this.permissions.canUploadToAlbum(user, session.albumId))) {
|
||||||
|
throw uploadSessionForbidden();
|
||||||
|
}
|
||||||
|
return session;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 续签(Task 19):仅 pending 会话;同 staging 键现签新凭证,并滑动顺延会话
|
||||||
|
* expiresAt(契约 UploadCredential.expiresAt:过期后续签再重试——renew 是
|
||||||
|
* 凭证过期的恢复路径,不得因过期而拒绝)。相册关闭上传即拒(§4.9)。
|
||||||
|
*/
|
||||||
|
async renew(
|
||||||
|
user: AuthUser,
|
||||||
|
uploadId: string,
|
||||||
|
): Promise<UploadInitResponse> {
|
||||||
|
const session = await this.locateAuthorizedSession(user, uploadId);
|
||||||
|
if (session.state !== 'pending') {
|
||||||
|
throw uploadSessionConflict('会话状态不允许续签(已完成/取消/过期)');
|
||||||
|
}
|
||||||
|
const album = await this.prisma.albums.findUnique({
|
||||||
|
where: { id: session.albumId },
|
||||||
|
select: { uploadEnabled: true },
|
||||||
|
});
|
||||||
|
if (album === null) {
|
||||||
|
// 相册行已不存在:同 403 防枚举(与不可见同码)。
|
||||||
|
throw uploadSessionForbidden();
|
||||||
|
}
|
||||||
|
if (!album.uploadEnabled) {
|
||||||
|
throw albumUploadDisabled();
|
||||||
|
}
|
||||||
|
const ttl = signedUrlTtlSeconds();
|
||||||
|
const credential = await this.storage.presignPut(
|
||||||
|
session.objectKey,
|
||||||
|
session.expectedBytes,
|
||||||
|
session.contentType,
|
||||||
|
ttl,
|
||||||
|
);
|
||||||
|
await this.prisma.upload_sessions.update({
|
||||||
|
where: { id: session.id },
|
||||||
|
data: { expiresAt: new Date(Date.now() + ttl * 1000) },
|
||||||
|
});
|
||||||
|
return {
|
||||||
|
uploadId: session.id,
|
||||||
|
photoId: session.photoId,
|
||||||
|
objectKey: session.objectKey,
|
||||||
|
credential,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 完成核验(Task 19):head 核验 → 受控复制 → 事务三表 + audit → 提交后入队。
|
||||||
|
* 幂等(A03):confirmed 会话重复调用经 copyIfAbsent 复核(originals 已在则
|
||||||
|
* created:false;staging 已清理亦不受影响)后回显当前状态,不产生第二个
|
||||||
|
* job/audit/入队。核验失败是请求级 400,会话保持 pending 可重试。
|
||||||
|
*/
|
||||||
|
async complete(
|
||||||
|
user: AuthUser,
|
||||||
|
uploadId: string,
|
||||||
|
): Promise<UploadCompleteResponse> {
|
||||||
|
const session = await this.locateAuthorizedSession(user, uploadId);
|
||||||
|
// 注意:契约 complete 无 ALBUM_UPLOAD_DISABLED——已完成的上传不因运营开关
|
||||||
|
// 被阻断(开关拒绝的是新直传与续签)。
|
||||||
|
if (session.state === 'cancelled' || session.state === 'expired') {
|
||||||
|
throw uploadSessionConflict('会话状态不允许完成(已取消或已过期)');
|
||||||
|
}
|
||||||
|
if (
|
||||||
|
session.state === 'pending' &&
|
||||||
|
session.expiresAt.getTime() <= Date.now()
|
||||||
|
) {
|
||||||
|
throw uploadSessionConflict('会话已过期(已取消或已过期)');
|
||||||
|
}
|
||||||
|
const photo = await this.prisma.photos.findUnique({
|
||||||
|
where: { id: session.photoId },
|
||||||
|
});
|
||||||
|
if (photo === null) {
|
||||||
|
// 会话与照片同事务创建,缺行只可能是装配异常/外部破坏,按 5xx 兜底。
|
||||||
|
throw new ApiException(
|
||||||
|
ErrorCodes.INTERNAL_ERROR,
|
||||||
|
'Internal server error',
|
||||||
|
HttpStatus.INTERNAL_SERVER_ERROR,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (session.state === 'confirmed') {
|
||||||
|
// 幂等快路径:originals 已核验过一次;copyIfAbsent 复核(created:false
|
||||||
|
// 路径,不依赖 staging 源仍存在),不重复 job/audit/入队。
|
||||||
|
await this.copyWithVerify(
|
||||||
|
session.objectKey,
|
||||||
|
photo.objectKey,
|
||||||
|
session.expectedBytes,
|
||||||
|
);
|
||||||
|
const current = await this.prisma.photos.findUnique({
|
||||||
|
where: { id: photo.id },
|
||||||
|
select: { processingState: true },
|
||||||
|
});
|
||||||
|
const processingState = (current ?? photo)
|
||||||
|
.processingState as UploadCompleteResponse['processingState'];
|
||||||
|
return { photoId: photo.id, processingState };
|
||||||
|
}
|
||||||
|
|
||||||
|
// ① 事务外存储核验:存在性/字节数/contentType(presign 不签名 content-type,
|
||||||
|
// 服务端兜底——Task 11 审查发现)。
|
||||||
|
const head = await this.storage.head(session.objectKey);
|
||||||
|
if (head === null) {
|
||||||
|
throw uploadVerifyFailed('staging 对象不存在或尚未上传');
|
||||||
|
}
|
||||||
|
if (head.bytes !== session.expectedBytes) {
|
||||||
|
throw uploadVerifyFailed(
|
||||||
|
`字节数与申明不符(expected=${session.expectedBytes}, actual=${head.bytes})`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if (head.contentType !== session.contentType) {
|
||||||
|
throw uploadVerifyFailed(
|
||||||
|
`Content-Type 与申明不符(expected=${session.contentType}, actual=${head.contentType})`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ② 受控复制 staging → 正式键(禁止覆盖语义:目标已在且一致则幂等返回)。
|
||||||
|
await this.copyWithVerify(
|
||||||
|
session.objectKey,
|
||||||
|
photo.objectKey,
|
||||||
|
session.expectedBytes,
|
||||||
|
);
|
||||||
|
|
||||||
|
// ③ 事务:photo 更新 + 会话条件更新抢占(并发 complete 恰一人胜出)+
|
||||||
|
// 胜者独占 photo_jobs 与 audit。
|
||||||
|
let claimed = false;
|
||||||
|
await this.prisma.$transaction(async (tx) => {
|
||||||
|
await tx.photos.update({
|
||||||
|
where: { id: photo.id },
|
||||||
|
data: {
|
||||||
|
bytes: session.expectedBytes,
|
||||||
|
uploadedAt: new Date(),
|
||||||
|
processingState: 'uploaded',
|
||||||
|
},
|
||||||
|
});
|
||||||
|
const race = await tx.upload_sessions.updateMany({
|
||||||
|
where: { id: session.id, state: 'pending' },
|
||||||
|
data: { state: 'confirmed' },
|
||||||
|
});
|
||||||
|
if (race.count === 1) {
|
||||||
|
claimed = true;
|
||||||
|
await tx.photo_jobs.upsert({
|
||||||
|
where: { photoId_type: { photoId: photo.id, type: 'process' } },
|
||||||
|
update: {}, // 已有 job(外部补偿先行/重复投递)不重复创建。
|
||||||
|
create: { photoId: photo.id, type: 'process', state: 'pending' },
|
||||||
|
});
|
||||||
|
await tx.audit_logs.create({
|
||||||
|
data: {
|
||||||
|
actorId: user.id,
|
||||||
|
action: 'upload.complete',
|
||||||
|
resourceId: photo.id,
|
||||||
|
metadata: {
|
||||||
|
uploadId: session.id,
|
||||||
|
bytes: session.expectedBytes,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
// ④ 提交后入队(失败仅日志:photo_jobs(pending) 是持久事实源,Task 21
|
||||||
|
// sweep 补偿;jobId 以 photoId 派生,重复入队天然去重)。
|
||||||
|
if (claimed) {
|
||||||
|
await this.queue.enqueuePhotoProcess(photo.id);
|
||||||
|
}
|
||||||
|
return { photoId: photo.id, processingState: 'uploaded' };
|
||||||
|
}
|
||||||
|
|
||||||
|
/** copyIfAbsent 包装:StorageError(UPLOAD_VERIFY_FAILED) → 同码 400 业务异常。 */
|
||||||
|
private async copyWithVerify(
|
||||||
|
src: string,
|
||||||
|
dest: string,
|
||||||
|
expectedBytes: number,
|
||||||
|
): Promise<void> {
|
||||||
|
try {
|
||||||
|
await this.storage.copyIfAbsent(src, dest, expectedBytes);
|
||||||
|
} catch (err) {
|
||||||
|
if (err instanceof StorageError) {
|
||||||
|
throw uploadVerifyFailed(err.message);
|
||||||
|
}
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 上传历史(Task 19):当前用户自己的会话(不分相册/状态),新在前
|
||||||
|
* (createdAt,id) 键集游标;photo 子对象内嵌服务端处理进度与可见性。
|
||||||
|
*/
|
||||||
|
async list(
|
||||||
|
user: AuthUser,
|
||||||
|
cursor: string | undefined,
|
||||||
|
limit: number,
|
||||||
|
): Promise<UploadSessionListResponse> {
|
||||||
|
let decoded: ReturnType<typeof decodeKeysetCursor> = null;
|
||||||
|
if (cursor !== undefined && cursor !== '') {
|
||||||
|
decoded = decodeKeysetCursor(cursor);
|
||||||
|
if (decoded === null) {
|
||||||
|
throw validationFailed('cursor 不合法');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const where: Prisma.upload_sessionsWhereInput = { userId: user.id };
|
||||||
|
if (decoded !== null) {
|
||||||
|
Object.assign(where, keysetWhereDesc('createdAt', 'id', decoded));
|
||||||
|
}
|
||||||
|
const rows = await this.prisma.upload_sessions.findMany({
|
||||||
|
where,
|
||||||
|
orderBy: [{ createdAt: 'desc' }, { id: 'desc' }],
|
||||||
|
take: limit + 1, // 多取一条探测 hasMore(同相册列表)
|
||||||
|
});
|
||||||
|
const hasMore = rows.length > limit;
|
||||||
|
const page = hasMore ? rows.slice(0, limit) : rows;
|
||||||
|
|
||||||
|
// 相册域式扁平外键:照片二次查询后应用层组装(会话与照片 1:1 同事务创建)。
|
||||||
|
const photos =
|
||||||
|
page.length === 0
|
||||||
|
? []
|
||||||
|
: await this.prisma.photos.findMany({
|
||||||
|
where: { id: { in: page.map((row) => row.photoId) } },
|
||||||
|
select: { id: true, processingState: true, visibility: true },
|
||||||
|
});
|
||||||
|
const photoById = new Map(photos.map((photo) => [photo.id, photo]));
|
||||||
|
|
||||||
|
const last = page.at(-1);
|
||||||
|
const nextCursor =
|
||||||
|
hasMore && last !== undefined
|
||||||
|
? encodeKeysetCursor({
|
||||||
|
sort: last.createdAt.toISOString(),
|
||||||
|
id: last.id,
|
||||||
|
})
|
||||||
|
: null;
|
||||||
|
return {
|
||||||
|
items: page.map((row) => {
|
||||||
|
const photo = photoById.get(row.photoId);
|
||||||
|
const record: UploadSessionRecord = {
|
||||||
|
uploadId: row.id,
|
||||||
|
albumId: row.albumId,
|
||||||
|
clientTaskId: row.clientTaskId,
|
||||||
|
state: row.state as UploadSessionRecord['state'],
|
||||||
|
expectedBytes: row.expectedBytes,
|
||||||
|
contentType: row.contentType,
|
||||||
|
createdAt: row.createdAt.toISOString(),
|
||||||
|
photo: {
|
||||||
|
photoId: row.photoId,
|
||||||
|
processingState:
|
||||||
|
(photo?.processingState ?? 'pending_upload') as UploadSessionRecord['photo']['processingState'],
|
||||||
|
visibility: (photo?.visibility ?? 'unpublished') as UploadSessionRecord['photo']['visibility'],
|
||||||
|
},
|
||||||
|
};
|
||||||
|
return record;
|
||||||
|
}),
|
||||||
|
nextCursor,
|
||||||
|
hasMore: nextCursor !== null,
|
||||||
|
};
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import { Test } from '@nestjs/testing';
|
|||||||
import request from 'supertest';
|
import request from 'supertest';
|
||||||
import { PrismaClient } from '@photo/db-cn';
|
import { PrismaClient } from '@photo/db-cn';
|
||||||
import { ErrorCodes } from '@photo/contracts';
|
import { ErrorCodes } from '@photo/contracts';
|
||||||
|
import { Queue } from 'bullmq';
|
||||||
import { AppCnModule } from '../src/app-cn.module';
|
import { AppCnModule } from '../src/app-cn.module';
|
||||||
import { hashSessionToken } from '../src/auth/session.guard';
|
import { hashSessionToken } from '../src/auth/session.guard';
|
||||||
import { API_PREFIX } from '../src/shared/constants';
|
import { API_PREFIX } from '../src/shared/constants';
|
||||||
@@ -424,3 +425,471 @@ describe('uploads init (e2e)', () => {
|
|||||||
expect(res.body.code).toBe(ErrorCodes.AUTH_TOKEN_EXPIRED);
|
expect(res.body.code).toBe(ErrorCodes.AUTH_TOKEN_EXPIRED);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// ── Task 19:renew / complete / history / 队列入队 ──
|
||||||
|
|
||||||
|
describe('uploads renew & complete & history (e2e)', () => {
|
||||||
|
let app: INestApplication;
|
||||||
|
let prisma: PrismaClient;
|
||||||
|
/** 测试侧独立 Queue 实例(同一 Redis):peek photo-process 队列;beforeAll 清空。 */
|
||||||
|
let queue: Queue;
|
||||||
|
/** 自建 ioredis 连接(BullMQ 不接管所有权,close 后须自行断开,否则 jest 挂起)。 */
|
||||||
|
let queueConnection: import('ioredis').default | null = null;
|
||||||
|
|
||||||
|
|
||||||
|
let ownerId: string;
|
||||||
|
let memberId: string;
|
||||||
|
let removedId: string;
|
||||||
|
let outsiderToken: string;
|
||||||
|
let ownerToken: string;
|
||||||
|
let memberToken: string;
|
||||||
|
let removedToken: string;
|
||||||
|
|
||||||
|
let albumId: string;
|
||||||
|
|
||||||
|
/** 各用例捕获的会话锚点(clientTaskId → 会话/照片 id)。 */
|
||||||
|
let r1: { uploadId: string; photoId: string; objectKey: string };
|
||||||
|
let cBytes: { uploadId: string; photoId: string };
|
||||||
|
let cHappy: { uploadId: string; photoId: string; objectKey: string };
|
||||||
|
|
||||||
|
const initAs = async (
|
||||||
|
token: string,
|
||||||
|
clientTaskId: string,
|
||||||
|
overrides: Record<string, unknown> = {},
|
||||||
|
): Promise<{ uploadId: string; photoId: string; objectKey: string; credential: { url: string; method: string; headers: Record<string, string> } }> => {
|
||||||
|
const res = await request(app.getHttpServer())
|
||||||
|
.post(`/${API_PREFIX}/uploads`)
|
||||||
|
.set('Authorization', `Bearer ${token}`)
|
||||||
|
.send({
|
||||||
|
albumId,
|
||||||
|
clientTaskId,
|
||||||
|
filename: 'a.jpg',
|
||||||
|
bytes: 1000,
|
||||||
|
contentType: 'image/jpeg',
|
||||||
|
...overrides,
|
||||||
|
});
|
||||||
|
expect(res.status).toBe(201);
|
||||||
|
return res.body;
|
||||||
|
};
|
||||||
|
|
||||||
|
const renew = (token: string, uploadId: string) =>
|
||||||
|
request(app.getHttpServer())
|
||||||
|
.post(`/${API_PREFIX}/uploads/${uploadId}/renew`)
|
||||||
|
.set('Authorization', `Bearer ${token}`);
|
||||||
|
|
||||||
|
const complete = (token: string, uploadId: string) =>
|
||||||
|
request(app.getHttpServer())
|
||||||
|
.post(`/${API_PREFIX}/uploads/${uploadId}/complete`)
|
||||||
|
.set('Authorization', `Bearer ${token}`);
|
||||||
|
|
||||||
|
const listUploads = (token: string, query?: Record<string, string | number>) =>
|
||||||
|
request(app.getHttpServer())
|
||||||
|
.get(`/${API_PREFIX}/uploads`)
|
||||||
|
.set('Authorization', `Bearer ${token}`)
|
||||||
|
.query(query ?? {});
|
||||||
|
|
||||||
|
/** 按凭证实写 staging(覆盖默认 content-type 以构造核验失败用例)。 */
|
||||||
|
const putStaging = async (
|
||||||
|
credential: { url: string; method: string; headers: Record<string, string> },
|
||||||
|
bytes: number,
|
||||||
|
contentType?: string,
|
||||||
|
): Promise<boolean> => {
|
||||||
|
const res = await fetch(credential.url, {
|
||||||
|
method: credential.method,
|
||||||
|
headers:
|
||||||
|
contentType === undefined
|
||||||
|
? credential.headers
|
||||||
|
: { 'Content-Type': contentType },
|
||||||
|
body: new Uint8Array(Buffer.alloc(bytes, 7)),
|
||||||
|
});
|
||||||
|
return res.ok;
|
||||||
|
};
|
||||||
|
|
||||||
|
beforeAll(async () => {
|
||||||
|
await resetDb();
|
||||||
|
const moduleRef = await Test.createTestingModule({
|
||||||
|
imports: [AppCnModule],
|
||||||
|
}).compile();
|
||||||
|
app = moduleRef.createNestApplication();
|
||||||
|
app.setGlobalPrefix(API_PREFIX);
|
||||||
|
await app.init();
|
||||||
|
|
||||||
|
prisma = new PrismaClient();
|
||||||
|
|
||||||
|
const mkUser = (account: string) =>
|
||||||
|
prisma.users.create({
|
||||||
|
data: { account, passwordHash: 'x', displayName: account },
|
||||||
|
select: { id: true },
|
||||||
|
});
|
||||||
|
ownerId = (await mkUser('rc-owner')).id;
|
||||||
|
memberId = (await mkUser('rc-member')).id;
|
||||||
|
removedId = (await mkUser('rc-removed')).id;
|
||||||
|
const outsiderId = (await mkUser('rc-outsider')).id;
|
||||||
|
|
||||||
|
const mkToken = async (userId: string): Promise<string> => {
|
||||||
|
const token = randomBytes(32).toString('hex');
|
||||||
|
await prisma.sessions.create({
|
||||||
|
data: {
|
||||||
|
userId,
|
||||||
|
tokenHash: hashSessionToken(token),
|
||||||
|
expiresAt: new Date(Date.now() + 3_600_000),
|
||||||
|
},
|
||||||
|
});
|
||||||
|
return token;
|
||||||
|
};
|
||||||
|
ownerToken = await mkToken(ownerId);
|
||||||
|
memberToken = await mkToken(memberId);
|
||||||
|
removedToken = await mkToken(removedId);
|
||||||
|
outsiderToken = await mkToken(outsiderId);
|
||||||
|
|
||||||
|
const res = await request(app.getHttpServer())
|
||||||
|
.post(`/${API_PREFIX}/albums`)
|
||||||
|
.set('Authorization', `Bearer ${ownerToken}`)
|
||||||
|
.send({
|
||||||
|
translations: [
|
||||||
|
{ locale: 'zh-CN', title: '完成核验相册' },
|
||||||
|
{ locale: 'en', title: 'Complete' },
|
||||||
|
],
|
||||||
|
defaultLocale: 'zh-CN',
|
||||||
|
startAt: '2026-12-01T02:00:00.000Z',
|
||||||
|
});
|
||||||
|
expect(res.status).toBe(201);
|
||||||
|
albumId = res.body.id as string;
|
||||||
|
await prisma.album_members.createMany({
|
||||||
|
data: [
|
||||||
|
{ albumId, userId: memberId, role: 'collaborator' },
|
||||||
|
{ albumId, userId: removedId, role: 'collaborator' },
|
||||||
|
],
|
||||||
|
});
|
||||||
|
|
||||||
|
// 测试侧队列实例:与生产者同一 Redis/队列名(契约常量),先清空历史。
|
||||||
|
const { QUEUE_PHOTO_PROCESS } = await import('@photo/contracts');
|
||||||
|
const IORedis = (await import('ioredis')).default;
|
||||||
|
queueConnection = new IORedis(
|
||||||
|
process.env.REDIS_URL_CN ?? 'redis://localhost:6379',
|
||||||
|
{ maxRetriesPerRequest: null },
|
||||||
|
);
|
||||||
|
queue = new Queue(QUEUE_PHOTO_PROCESS, { connection: queueConnection });
|
||||||
|
await queue.obliterate({ force: true });
|
||||||
|
});
|
||||||
|
|
||||||
|
afterAll(async () => {
|
||||||
|
await app.close();
|
||||||
|
await prisma.$disconnect();
|
||||||
|
await queue.close();
|
||||||
|
// BullMQ 对外部传入的连接不接管生命周期:queue.close 不 disconnect 它。
|
||||||
|
await queueConnection?.quit().catch(() => undefined);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('renew → 200 新凭证同 objectKey;会话续期仍 pending', async () => {
|
||||||
|
const init = await initAs(memberToken, 'R1');
|
||||||
|
r1 = init;
|
||||||
|
const res = await renew(memberToken, init.uploadId);
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
expect(Object.keys(res.body).sort()).toEqual(INIT_KEYS);
|
||||||
|
expect(res.body.uploadId).toBe(init.uploadId);
|
||||||
|
expect(res.body.photoId).toBe(init.photoId);
|
||||||
|
expect(res.body.objectKey).toBe(init.objectKey);
|
||||||
|
expect(res.body.credential.method).toBe('PUT');
|
||||||
|
expect(res.body.credential.headers['Content-Type']).toBe('image/jpeg');
|
||||||
|
expect(new Date(res.body.credential.expiresAt).getTime()).toBeGreaterThan(
|
||||||
|
Date.now(),
|
||||||
|
);
|
||||||
|
|
||||||
|
const session = await prisma.upload_sessions.findUnique({
|
||||||
|
where: { id: init.uploadId },
|
||||||
|
});
|
||||||
|
expect(session?.state).toBe('pending');
|
||||||
|
// renew 滑动续期会话 expiresAt(契约:凭证过期 → renew 再试)。
|
||||||
|
expect(session?.expiresAt.getTime()).toBeGreaterThan(Date.now() + 800_000);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('renew 守卫:uploadEnabled=false → 409 ALBUM_UPLOAD_DISABLED;非本人/未知会话/被移除成员 → 403', async () => {
|
||||||
|
// 相册关闭上传:续签被拒(§4.9),owner 亦同(开关是相册级运营状态)。
|
||||||
|
await prisma.albums.update({
|
||||||
|
where: { id: albumId },
|
||||||
|
data: { uploadEnabled: false },
|
||||||
|
});
|
||||||
|
try {
|
||||||
|
const disabled = await renew(memberToken, r1.uploadId);
|
||||||
|
expect(disabled.status).toBe(409);
|
||||||
|
expect(disabled.body.code).toBe(ErrorCodes.ALBUM_UPLOAD_DISABLED);
|
||||||
|
} finally {
|
||||||
|
await prisma.albums.update({
|
||||||
|
where: { id: albumId },
|
||||||
|
data: { uploadEnabled: true },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
// 被移除成员:先建会话(此时仍是成员),移除后 renew/complete 双拦截(A02)。
|
||||||
|
const removedSession = await initAs(removedToken, 'C-removed');
|
||||||
|
await prisma.album_members.delete({
|
||||||
|
where: { albumId_userId: { albumId, userId: removedId } },
|
||||||
|
});
|
||||||
|
const removedRenew = await renew(removedToken, removedSession.uploadId);
|
||||||
|
expect(removedRenew.status).toBe(403);
|
||||||
|
expect(removedRenew.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN);
|
||||||
|
|
||||||
|
const removedComplete = await complete(removedToken, removedSession.uploadId);
|
||||||
|
expect(removedComplete.status).toBe(403);
|
||||||
|
expect(removedComplete.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN);
|
||||||
|
|
||||||
|
// 非本人会话与未知会话同码 403(契约:不区分以避免枚举)。
|
||||||
|
const notOwner = await renew(ownerToken, r1.uploadId);
|
||||||
|
expect(notOwner.status).toBe(403);
|
||||||
|
expect(notOwner.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN);
|
||||||
|
|
||||||
|
const unknown = await renew(memberToken, 'no-such-upload');
|
||||||
|
expect(unknown.status).toBe(403);
|
||||||
|
expect(unknown.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('complete:未上传 → 400 UPLOAD_VERIFY_FAILED,会话保持 pending 可重试', async () => {
|
||||||
|
const init = await initAs(memberToken, 'C-before');
|
||||||
|
const res = await complete(memberToken, init.uploadId);
|
||||||
|
expect(res.status).toBe(400);
|
||||||
|
expect(res.body.code).toBe(ErrorCodes.UPLOAD_VERIFY_FAILED);
|
||||||
|
const session = await prisma.upload_sessions.findUnique({
|
||||||
|
where: { id: init.uploadId },
|
||||||
|
});
|
||||||
|
expect(session?.state).toBe('pending');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('complete:字节数不符 → 400;补传正确字节数后 complete 成功(请求级失败不毒化会话)', async () => {
|
||||||
|
const init = await initAs(memberToken, 'C-bytes');
|
||||||
|
cBytes = { uploadId: init.uploadId, photoId: init.photoId };
|
||||||
|
expect(await putStaging(init.credential, 500)).toBe(true); // 申明 1000 实传 500
|
||||||
|
|
||||||
|
const mismatch = await complete(memberToken, init.uploadId);
|
||||||
|
expect(mismatch.status).toBe(400);
|
||||||
|
expect(mismatch.body.code).toBe(ErrorCodes.UPLOAD_VERIFY_FAILED);
|
||||||
|
const session = await prisma.upload_sessions.findUnique({
|
||||||
|
where: { id: init.uploadId },
|
||||||
|
});
|
||||||
|
expect(session?.state).toBe('pending');
|
||||||
|
|
||||||
|
// 重试路径:重传正确字节数(同 staging 键覆盖写)→ complete 通过。
|
||||||
|
expect(await putStaging(init.credential, 1000)).toBe(true);
|
||||||
|
const retry = await complete(memberToken, init.uploadId);
|
||||||
|
expect(retry.status).toBe(200);
|
||||||
|
expect(retry.body.processingState).toBe('uploaded');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('complete:contentType 不符(presign 不签名 content-type,服务端兜底)→ 400', async () => {
|
||||||
|
const init = await initAs(memberToken, 'C-type');
|
||||||
|
// 声明 image/jpeg,实传 application/octet-stream:PUT 成功(签名不覆盖该头),
|
||||||
|
// complete 时 head().contentType 与申明值比对兜底。
|
||||||
|
expect(await putStaging(init.credential, 1000, 'application/octet-stream')).toBe(true);
|
||||||
|
const res = await complete(memberToken, init.uploadId);
|
||||||
|
expect(res.status).toBe(400);
|
||||||
|
expect(res.body.code).toBe(ErrorCodes.UPLOAD_VERIFY_FAILED);
|
||||||
|
const session = await prisma.upload_sessions.findUnique({
|
||||||
|
where: { id: init.uploadId },
|
||||||
|
});
|
||||||
|
expect(session?.state).toBe('pending');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('complete 成功 → 200 uploaded;photo/session/photo_jobs/audit 同事务落库;重复 complete 幂等(A03)', async () => {
|
||||||
|
const init = await initAs(memberToken, 'C-happy');
|
||||||
|
cHappy = { uploadId: init.uploadId, photoId: init.photoId, objectKey: init.objectKey };
|
||||||
|
expect(await putStaging(init.credential, 1000)).toBe(true);
|
||||||
|
|
||||||
|
const res = await complete(memberToken, init.uploadId);
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
expect(Object.keys(res.body).sort()).toEqual(['photoId', 'processingState']);
|
||||||
|
expect(res.body).toEqual({ photoId: init.photoId, processingState: 'uploaded' });
|
||||||
|
|
||||||
|
// DB 四处一致性。
|
||||||
|
const photo = await prisma.photos.findUnique({ where: { id: init.photoId } });
|
||||||
|
expect(photo).toMatchObject({
|
||||||
|
bytes: 1000,
|
||||||
|
uploadedAt: expect.any(Date),
|
||||||
|
processingState: 'uploaded',
|
||||||
|
visibility: 'unpublished',
|
||||||
|
});
|
||||||
|
const session = await prisma.upload_sessions.findUnique({
|
||||||
|
where: { id: init.uploadId },
|
||||||
|
});
|
||||||
|
expect(session?.state).toBe('confirmed');
|
||||||
|
|
||||||
|
const jobs = await prisma.photo_jobs.findMany({
|
||||||
|
where: { photoId: init.photoId },
|
||||||
|
});
|
||||||
|
expect(jobs).toHaveLength(1);
|
||||||
|
expect(jobs[0]).toMatchObject({ type: 'process', state: 'pending' });
|
||||||
|
|
||||||
|
const audits = await prisma.audit_logs.findMany({
|
||||||
|
where: { action: 'upload.complete', resourceId: init.photoId },
|
||||||
|
});
|
||||||
|
expect(audits).toHaveLength(1);
|
||||||
|
expect(audits[0]?.actorId).toBe(memberId);
|
||||||
|
|
||||||
|
// 幂等重复 complete:同结果,不产生第二个 job/audit。
|
||||||
|
const again = await complete(memberToken, init.uploadId);
|
||||||
|
expect(again.status).toBe(200);
|
||||||
|
expect(again.body).toEqual({ photoId: init.photoId, processingState: 'uploaded' });
|
||||||
|
expect(
|
||||||
|
await prisma.photo_jobs.count({ where: { photoId: init.photoId } }),
|
||||||
|
).toBe(1);
|
||||||
|
expect(
|
||||||
|
await prisma.audit_logs.count({
|
||||||
|
where: { action: 'upload.complete', resourceId: init.photoId },
|
||||||
|
}),
|
||||||
|
).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('受控复制核验:originals 与 staging 并存且字节一致;删除 staging 后 re-complete 仍 200(copyIfAbsent created:false 路径)', async () => {
|
||||||
|
const { createStorageFromEnv } = await import('@photo/storage');
|
||||||
|
const storage = createStorageFromEnv();
|
||||||
|
|
||||||
|
const photo = await prisma.photos.findUnique({ where: { id: cHappy.photoId } });
|
||||||
|
expect(photo?.objectKey).toBe(
|
||||||
|
`photo-live/dev/albums/${albumId}/originals/${cHappy.photoId}.jpg`,
|
||||||
|
);
|
||||||
|
|
||||||
|
const stagingHead = await storage.head(cHappy.objectKey);
|
||||||
|
const finalHead = await storage.head(photo?.objectKey as string);
|
||||||
|
expect(stagingHead?.bytes).toBe(1000);
|
||||||
|
expect(finalHead?.bytes).toBe(1000);
|
||||||
|
|
||||||
|
// 手工删除 staging(模拟生命周期清理/超时)后重复 complete:originals 已在,
|
||||||
|
// copyIfAbsent 幂等返回 created:false,不再依赖 staging 源。
|
||||||
|
const { DeleteObjectCommand, S3Client } = await import('@aws-sdk/client-s3');
|
||||||
|
const s3 = new S3Client({
|
||||||
|
endpoint: process.env.S3_ENDPOINT,
|
||||||
|
region: process.env.S3_REGION,
|
||||||
|
credentials: {
|
||||||
|
accessKeyId: process.env.S3_ACCESS_KEY as string,
|
||||||
|
secretAccessKey: process.env.S3_SECRET_KEY as string,
|
||||||
|
},
|
||||||
|
forcePathStyle: true,
|
||||||
|
});
|
||||||
|
try {
|
||||||
|
await s3.send(
|
||||||
|
new DeleteObjectCommand({
|
||||||
|
Bucket: process.env.S3_BUCKET as string,
|
||||||
|
Key: cHappy.objectKey,
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
} finally {
|
||||||
|
await s3.destroy();
|
||||||
|
}
|
||||||
|
expect(await storage.head(cHappy.objectKey)).toBeNull();
|
||||||
|
|
||||||
|
const res = await complete(memberToken, cHappy.uploadId);
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
expect(res.body.processingState).toBe('uploaded');
|
||||||
|
expect((await storage.head(photo?.objectKey as string))?.bytes).toBe(1000);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('GET /uploads:本人会话新在前,photo 状态内嵌,游标翻页,跨用户隔离', async () => {
|
||||||
|
// 回写互不相同的 createdAt,排序断言与写入时序解耦(同 members 列表先例)。
|
||||||
|
const order: Array<{ taskId: string; at: string }> = [
|
||||||
|
{ taskId: 'R1', at: '2026-09-12T08:00:00.000Z' }, // 最旧
|
||||||
|
{ taskId: 'C-before', at: '2026-09-12T09:00:00.000Z' },
|
||||||
|
{ taskId: 'C-bytes', at: '2026-09-12T10:00:00.000Z' },
|
||||||
|
{ taskId: 'C-type', at: '2026-09-12T11:00:00.000Z' },
|
||||||
|
{ taskId: 'C-happy', at: '2026-09-12T12:00:00.000Z' }, // 最新
|
||||||
|
];
|
||||||
|
for (const { taskId, at } of order) {
|
||||||
|
await prisma.upload_sessions.updateMany({
|
||||||
|
where: { userId: memberId, clientTaskId: taskId },
|
||||||
|
data: { createdAt: new Date(at) },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
const page1 = await listUploads(memberToken, { limit: 2 });
|
||||||
|
expect(page1.status).toBe(200);
|
||||||
|
expect(
|
||||||
|
page1.body.items.map((i: { clientTaskId: string }) => i.clientTaskId),
|
||||||
|
).toEqual(['C-happy', 'C-type']);
|
||||||
|
expect(Object.keys(page1.body.items[0]).sort()).toEqual([
|
||||||
|
'albumId',
|
||||||
|
'clientTaskId',
|
||||||
|
'contentType',
|
||||||
|
'createdAt',
|
||||||
|
'expectedBytes',
|
||||||
|
'photo',
|
||||||
|
'state',
|
||||||
|
'uploadId',
|
||||||
|
]);
|
||||||
|
expect(Object.keys(page1.body.items[0].photo).sort()).toEqual([
|
||||||
|
'photoId',
|
||||||
|
'processingState',
|
||||||
|
'visibility',
|
||||||
|
]);
|
||||||
|
// confirmed 会话内嵌服务端进度;pending 会话保持 pending_upload/unpublished。
|
||||||
|
expect(page1.body.items[0]).toMatchObject({
|
||||||
|
state: 'confirmed',
|
||||||
|
albumId,
|
||||||
|
expectedBytes: 1000,
|
||||||
|
contentType: 'image/jpeg',
|
||||||
|
photo: {
|
||||||
|
photoId: cHappy.photoId,
|
||||||
|
processingState: 'uploaded',
|
||||||
|
visibility: 'unpublished',
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(page1.body.items[1]).toMatchObject({
|
||||||
|
state: 'pending',
|
||||||
|
photo: { processingState: 'pending_upload', visibility: 'unpublished' },
|
||||||
|
});
|
||||||
|
expect(page1.body.hasMore).toBe(true);
|
||||||
|
expect(typeof page1.body.nextCursor).toBe('string');
|
||||||
|
|
||||||
|
const page2 = await listUploads(memberToken, {
|
||||||
|
limit: 2,
|
||||||
|
cursor: page1.body.nextCursor,
|
||||||
|
});
|
||||||
|
expect(page2.status).toBe(200);
|
||||||
|
expect(
|
||||||
|
page2.body.items.map((i: { clientTaskId: string }) => i.clientTaskId),
|
||||||
|
).toEqual(['C-bytes', 'C-before']);
|
||||||
|
expect(page2.body.hasMore).toBe(true);
|
||||||
|
|
||||||
|
const page3 = await listUploads(memberToken, {
|
||||||
|
limit: 2,
|
||||||
|
cursor: page2.body.nextCursor,
|
||||||
|
});
|
||||||
|
expect(
|
||||||
|
page3.body.items.map((i: { clientTaskId: string }) => i.clientTaskId),
|
||||||
|
).toEqual(['R1']);
|
||||||
|
expect(page3.body.hasMore).toBe(false);
|
||||||
|
expect(page3.body.nextCursor).toBeNull();
|
||||||
|
|
||||||
|
// 跨用户隔离:removed(仅有已被拦截的会话)与 outsider 看不到 member 会话。
|
||||||
|
const outsiderList = await listUploads(outsiderToken);
|
||||||
|
expect(outsiderList.status).toBe(200);
|
||||||
|
expect(outsiderList.body.items).toHaveLength(0);
|
||||||
|
|
||||||
|
const removedList = await listUploads(removedToken);
|
||||||
|
expect(removedList.status).toBe(200);
|
||||||
|
expect(removedList.body.items).toHaveLength(1); // 仅自己的 C-removed
|
||||||
|
|
||||||
|
// 非法 cursor → 400。
|
||||||
|
const bad = await listUploads(memberToken, { cursor: 'garbage' });
|
||||||
|
expect(bad.status).toBe(400);
|
||||||
|
expect(bad.body.code).toBe(ErrorCodes.VALIDATION_FAILED);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('BullMQ 入队:complete 后 photo-process 队列恰有一条对应 job(重复 complete 不重复入队)', async () => {
|
||||||
|
// C-bytes(重试成功)与 C-happy 各完成一次 → 各入队一次;jobId 以 photoId
|
||||||
|
// 派生去重。等待短暂时间让 enqueue 落 Redis(生产者为 await,理论即时)。
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 200));
|
||||||
|
const jobs = await queue.getJobs(
|
||||||
|
['waiting', 'delayed', 'active', 'completed', 'failed'],
|
||||||
|
0,
|
||||||
|
-1,
|
||||||
|
);
|
||||||
|
const byPhoto = new Map<string, number>();
|
||||||
|
for (const job of jobs) {
|
||||||
|
const photoId = (job.data as { photoId?: string }).photoId ?? '(none)';
|
||||||
|
byPhoto.set(photoId, (byPhoto.get(photoId) ?? 0) + 1);
|
||||||
|
}
|
||||||
|
expect(byPhoto.get(cHappy.photoId)).toBe(1);
|
||||||
|
expect(byPhoto.get(cBytes.photoId)).toBe(1);
|
||||||
|
// 会话归属成员的全部完成(两次 complete 幂等重复不重复入队)——队列中不
|
||||||
|
// 存在其他 member 照片之外的 job。
|
||||||
|
expect(byPhoto.size).toBe(2);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user