From f5f1855b5831cf06a80f488beb2450fcab072d16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 17 Sep 2026 23:00:19 +0800 Subject: [PATCH] feat(api): upload verify, forbid-overwrite copy, idempotent complete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- services/api/src/uploads/queue.provider.ts | 16 +- .../api/src/uploads/uploads.controller.ts | 61 ++- services/api/src/uploads/uploads.service.ts | 318 +++++++++++- services/api/test/uploads.e2e-spec.ts | 469 ++++++++++++++++++ 4 files changed, 850 insertions(+), 14 deletions(-) diff --git a/services/api/src/uploads/queue.provider.ts b/services/api/src/uploads/queue.provider.ts index 9266512..929ea1c 100644 --- a/services/api/src/uploads/queue.provider.ts +++ b/services/api/src/uploads/queue.provider.ts @@ -30,21 +30,22 @@ export class BullMqPhotoProcessQueue { private readonly logger = new Logger(BullMqPhotoProcessQueue.name); private queue: Queue | null = null; + private connection: IORedis | null = null; /** 首次使用时构造；连接错误经 no-op 监听吞掉（见文件头注释）。 */ private getQueue(): Queue { if (this.queue === null) { const url = process.env.REDIS_URL_CN ?? 'redis://localhost:6379'; - const connection = new IORedis(url, { + this.connection = new IORedis(url, { // BullMQ 对阻塞命令的要求：禁用 per-request 重试上限（等待 Worker 语义）。 maxRetriesPerRequest: null, }); - connection.on('error', (err: Error) => { + this.connection.on('error', (err: Error) => { this.logger.warn( `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; } @@ -56,7 +57,8 @@ export class BullMqPhotoProcessQueue { photoId }, { // jobId 以 photoId 派生：并发 complete/补偿扫描重复入队天然去重（A03）。 - jobId: `${PHOTO_PROCESS_JOB_NAME}:${photoId}`, + // 注意 BullMQ 禁止 jobId 含 ':'（保留分隔符），用 '-' 连接。 + jobId: `${PHOTO_PROCESS_JOB_NAME}-${photoId}`, // Task 21 消费侧重试策略：3 次指数退避。 attempts: 3, backoff: { type: 'exponential', delay: 1000 }, @@ -77,6 +79,12 @@ export class BullMqPhotoProcessQueue await this.queue.close().catch(() => undefined); this.queue = null; } + // BullMQ 不接管外部传入连接的生命周期：queue.close 不 disconnect 它， + // 这里显式断开，否则进程（含 e2e 的 app 实例）残留打开的 Redis 句柄。 + if (this.connection !== null) { + await this.connection.quit().catch(() => undefined); + this.connection = null; + } } } diff --git a/services/api/src/uploads/uploads.controller.ts b/services/api/src/uploads/uploads.controller.ts index 9d969f5..abe7dc6 100644 --- a/services/api/src/uploads/uploads.controller.ts +++ b/services/api/src/uploads/uploads.controller.ts @@ -1,18 +1,33 @@ -// services/api/src/uploads/uploads.controller.ts —— /uploads 端点（Task 18 起）。 -// 方法名与契约 operationId 一一对应（initUpload；Task 19 增补 renewUploadCredential/ -// completeUpload/listUploads）。控制器级 SessionGuard：端点全需认证。职责： -// 透传认证用户 + 查询参数规整（parseLimit/parseCursor 复用相册控制器导出）。 +// services/api/src/uploads/uploads.controller.ts —— /uploads 端点（Task 18 起； +// Task 19 增补 listUploads/renewUploadCredential/completeUpload）。方法名与契约 +// operationId 一一对应。控制器级 SessionGuard：端点全需认证。职责：透传认证 +// 用户 + 查询参数规整（parseLimit/parseCursor 复用相册控制器导出，同契约 bounds）。 +// 注意 DTO 必须值导入（非 import type）：@Body 参数类型经 emitDecoratorMetadata +// 进入设计期元数据，import type 会令参数在管线中被置为 undefined（Task 18 实测）。 import { Body, Controller, + Get, + HttpCode, + HttpStatus, + Param, Post, + Query, Req, UseGuards, } 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 { UploadsService } from './uploads.service'; -import type { UploadInitResponse } from './uploads.service'; +import type { + UploadCompleteResponse, + UploadInitResponse, + UploadSessionListResponse, +} from './uploads.service'; import { UploadInitRequestDto } from './dto'; @Controller('uploads') @@ -28,4 +43,38 @@ export class UploadsController { ): Promise { 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 { + 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 { + return this.uploadsService.renew(userOf(req), uploadId); + } + + /** 完成核验：幂等（重复调用返回相同结果，不重复建 job）。 */ + @Post(':uploadId/complete') + @HttpCode(HttpStatus.OK) + completeUpload( + @Param('uploadId') uploadId: string, + @Req() req: AuthenticatedRequest, + ): Promise { + return this.uploadsService.complete(userOf(req), uploadId); + } } diff --git a/services/api/src/uploads/uploads.service.ts b/services/api/src/uploads/uploads.service.ts index 0d17d11..2a10f37 100644 --- a/services/api/src/uploads/uploads.service.ts +++ b/services/api/src/uploads/uploads.service.ts @@ -1,5 +1,6 @@ -// services/api/src/uploads/uploads.service.ts —— 上传域核心逻辑（Task 18 起）。 -// 契约权威：packages/contracts/openapi.yaml tags [uploads]。要点： +// services/api/src/uploads/uploads.service.ts —— 上传域核心逻辑（Task 18 起， +// Task 19 增补 renew/complete/list）。契约权威：packages/contracts/openapi.yaml +// tags [uploads]。要点： // · initUpload 幂等初始化（A03）：@@unique([userId, clientTaskId]) 为幂等锚， // 同键同参返回原会话（201，不重复建照片）；同键异参 409 UPLOAD_SESSION_CONFLICT。 // 并发同键插入由 P2002 捕获后回读既有会话裁决（MySQL 约束冲突不中止事务）。 @@ -13,10 +14,26 @@ // · init 不落 audit（高频低风险；审计锚点在 complete，A06 前置约定）。 // · UPLOAD_MAX_BYTES/SIGNED_URL_TTL_SECONDS 逐请求惰性读取（support）， // 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 { 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 { ApiException } from '../api-exception'; import { PermissionsService } from '../albums/permissions.service'; import { albumForbidden, @@ -25,7 +42,16 @@ import { validationFailed, } from '../albums/support'; import { PrismaService } from '../prisma/prisma.service'; +import { + decodeKeysetCursor, + encodeKeysetCursor, + keysetWhereDesc, +} from '../shared/cursor'; import { STORAGE_PORT } from './storage.provider'; +import { + PHOTO_PROCESS_QUEUE, + type PhotoProcessQueueProducer, +} from './queue.provider'; import type { UploadInitRequestDto } from './dto'; import { albumUploadDisabled, @@ -37,10 +63,17 @@ import { stagingObjectKey, uploadMaxBytes, uploadSessionConflict, + uploadSessionForbidden, + uploadVerifyFailed, } from './support'; export type UploadInitResponse = components['schemas']['UploadInitResponse']; +export type UploadCompleteResponse = + components['schemas']['UploadCompleteResponse']; +export type UploadSessionListResponse = + components['schemas']['UploadSessionList']; +type UploadSessionRecord = components['schemas']['UploadSessionRecord']; type SessionRow = { id: string; @@ -51,12 +84,16 @@ type SessionRow = { albumId: string; }; +type UploadSessionRow = Prisma.upload_sessionsGetPayload; + @Injectable() export class UploadsService { constructor( private readonly prisma: PrismaService, private readonly permissions: PermissionsService, @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 ); } + + /** + * 会话定位 + 归属 + 逐次权限核验（renew/complete 共用）。任何一项不过都抛 + * 同一 403 AUTH_FORBIDDEN（契约：无权/不存在不区分，防 uploadId 枚举）。 + * 返回值保证非空且 userId === user.id。 + */ + private async locateAuthorizedSession( + user: AuthUser, + uploadId: string, + ): Promise { + 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 { + 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 { + 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 { + 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 { + let decoded: ReturnType = 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, + }; + } } diff --git a/services/api/test/uploads.e2e-spec.ts b/services/api/test/uploads.e2e-spec.ts index c70025f..b1a16fa 100644 --- a/services/api/test/uploads.e2e-spec.ts +++ b/services/api/test/uploads.e2e-spec.ts @@ -18,6 +18,7 @@ import { Test } from '@nestjs/testing'; import request from 'supertest'; import { PrismaClient } from '@photo/db-cn'; import { ErrorCodes } from '@photo/contracts'; +import { Queue } from 'bullmq'; import { AppCnModule } from '../src/app-cn.module'; import { hashSessionToken } from '../src/auth/session.guard'; import { API_PREFIX } from '../src/shared/constants'; @@ -424,3 +425,471 @@ describe('uploads init (e2e)', () => { 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 = {}, + ): Promise<{ uploadId: string; photoId: string; objectKey: string; credential: { url: string; method: string; headers: Record } }> => { + 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) => + 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 }, + bytes: number, + contentType?: string, + ): Promise => { + 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 => { + 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(); + 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); + }); +});