diff --git a/packages/contracts/src/__tests__/watermark.test.ts b/packages/contracts/src/__tests__/watermark.test.ts new file mode 100644 index 0000000..f8180c7 --- /dev/null +++ b/packages/contracts/src/__tests__/watermark.test.ts @@ -0,0 +1,121 @@ +// packages/contracts/src/__tests__/watermark.test.ts —— 水印配置共享形状测试 +// (P4a Task 3):normalize 填齐缺省不改已给值;parse 对脏数据的防御回退; +// buildAlbumUpdatedPayloadCore 的水印副本位(watermarkConfigVersion = 有效版本) +// 与翻译排序确定性。这三者是 api/worker 双侧共用的唯一权威,漂移即事故。 +import { + normalizeWatermarkConfig, + parseStoredWatermarkConfig, + WATERMARK_DEFAULT_POSITION, + WATERMARK_DEFAULT_OPACITY, + WATERMARK_DEFAULT_SCALE_PERCENT, + WATERMARK_DEFAULT_FONT_SIZE, + WATERMARK_POSITIONS, + buildAlbumUpdatedPayloadCore, + type AlbumPayloadRow, +} from '../index.js'; + +describe('normalizeWatermarkConfig', () => { + it('填齐全部缺省(与 openapi WatermarkConfig 可选字段一一对应)', () => { + expect(normalizeWatermarkConfig({ enabled: true, text: { content: 'x' } })).toEqual({ + enabled: true, + position: WATERMARK_DEFAULT_POSITION, + opacity: WATERMARK_DEFAULT_OPACITY, + scalePercent: WATERMARK_DEFAULT_SCALE_PERCENT, + text: { content: 'x', fontSize: WATERMARK_DEFAULT_FONT_SIZE }, + }); + }); + + it('不改写已给值;enabled=false 允许无文字(关闭语义仅回显)', () => { + expect( + normalizeWatermarkConfig({ + enabled: false, + position: 'topLeft', + opacity: 80, + scalePercent: 30, + text: { content: '', fontSize: 40 }, + }), + ).toEqual({ + enabled: false, + position: 'topLeft', + opacity: 80, + scalePercent: 30, + text: { content: '', fontSize: 40 }, + }); + expect(normalizeWatermarkConfig({ enabled: false })).toMatchObject({ + text: { content: '' }, + }); + }); +}); + +describe('parseStoredWatermarkConfig', () => { + it('round-trip:normalize 后的值原样读回', () => { + const stored = normalizeWatermarkConfig({ + enabled: true, + position: 'center', + opacity: 10, + scalePercent: 50, + text: { content: '水印', fontSize: 72 }, + }); + expect(parseStoredWatermarkConfig(stored)).toEqual(stored); + }); + + it('脏数据防御:非对象 / enabled 非 boolean → null(「配置不可用」)', () => { + expect(parseStoredWatermarkConfig(null)).toBeNull(); + expect(parseStoredWatermarkConfig('enabled')).toBeNull(); + expect(parseStoredWatermarkConfig({ position: 'center' })).toBeNull(); + expect(parseStoredWatermarkConfig({ enabled: 'yes' })).toBeNull(); + }); + + it('缺字段回落缺省;非法 position 回落缺省(不抛错)', () => { + expect(parseStoredWatermarkConfig({ enabled: true })).toEqual( + normalizeWatermarkConfig({ enabled: true }), + ); + const badPosition = parseStoredWatermarkConfig({ + enabled: true, + position: 'middle', + text: { content: 'x' }, + }); + expect(badPosition?.position).toBe(WATERMARK_DEFAULT_POSITION); + expect(WATERMARK_POSITIONS).not.toContain('middle'); + }); +}); + +describe('buildAlbumUpdatedPayloadCore', () => { + const album: AlbumPayloadRow = { + id: 'album-1', + publicToken: 'pt', + status: 'active', + uploadEnabled: true, + publicEnabled: true, + downloadEnabled: false, + originalDownloadEnabled: false, + watermarkEnabled: true, + watermarkEffectiveVersion: 3, + defaultLocale: 'zh-CN', + timezone: 'Asia/Shanghai', + startAt: new Date('2026-11-01T10:00:00Z'), + endAt: null, + entityVersion: 7, + }; + + it('水印副本位 = 有效版本(不是最新创建版本);翻译 locale 升序', () => { + const payload = buildAlbumUpdatedPayloadCore(album, [ + { locale: 'en', title: 'T', description: null, venue: null, address: null }, + { locale: 'zh-CN', title: '标题', description: null, venue: null, address: null }, + ]); + expect(payload.watermarkEnabled).toBe(true); + expect(payload.watermarkConfigVersion).toBe(3); + expect(payload.translations.map((t) => t.locale)).toEqual(['en', 'zh-CN']); + expect(payload.entityVersion).toBe(7); + expect(payload.endAt).toBeNull(); + }); + + it('关闭态:enabled=false 携带版本(ov 回落 browse 键的判定依据)', () => { + const payload = buildAlbumUpdatedPayloadCore( + { ...album, watermarkEnabled: false, watermarkEffectiveVersion: 4 }, + [], + ); + expect(payload.watermarkEnabled).toBe(false); + expect(payload.watermarkConfigVersion).toBe(4); + }); +}); diff --git a/packages/contracts/src/album-payload.ts b/packages/contracts/src/album-payload.ts new file mode 100644 index 0000000..f36503a --- /dev/null +++ b/packages/contracts/src/album-payload.ts @@ -0,0 +1,70 @@ +// packages/contracts/src/album-payload.ts —— album.updated 载荷构建器(P4a Task 3 +// 起为跨服务共享实现)。此前唯一实现在 api src/sync/payloads.ts;水印任务完成时 +// **worker** 也要写 album.updated outbox(有效版本推进须同步 ov),为避免两份 +// 实现漂移,构建器收敛到契约包:输入为结构化行形状(不依赖 Prisma 类型), +// api 的 sync/payloads.ts 包装本实现并叠加 categories 快照参数,worker 直接调用。 +import type { AlbumUpdatedPayload } from './events.js'; + +/** 构建器所需的主行列子集(cn albums 行的结构化形状;两消费方的 Prisma 行天然满足)。 */ +export interface AlbumPayloadRow { + id: string; + publicToken: string; + status: string; + uploadEnabled: boolean; + publicEnabled: boolean; + downloadEnabled: boolean; + originalDownloadEnabled: boolean; + /** 有效水印位(见 watermark.ts 版本语义;列由 P4a Task 3 迁移新增,行必含)。 */ + watermarkEnabled: boolean; + watermarkEffectiveVersion: number; + defaultLocale: string; + timezone: string; + startAt: Date; + endAt: Date | null; + entityVersion: number; +} + +/** 翻译行的结构化形状(cn album_translations 行满足)。 */ +export interface AlbumTranslationRow { + locale: string; + title: string; + description: string | null; + venue: string | null; + address: string | null; +} + +/** + * album.updated 载荷(无 categories 版):标量全集 + 翻译(locale 升序)+ + * 水印副本位。与 api sync/payloads.ts 的完整版逐字节同形(categories 是其 + * 独有的可选叠加参数,业务事务内事件与 worker 事件均不携带)。 + */ +export function buildAlbumUpdatedPayloadCore( + album: AlbumPayloadRow, + translations: AlbumTranslationRow[], +): AlbumUpdatedPayload { + return { + albumId: album.id, + publicToken: album.publicToken, + status: album.status, + uploadEnabled: album.uploadEnabled, + publicEnabled: album.publicEnabled, + downloadEnabled: album.downloadEnabled, + originalDownloadEnabled: album.originalDownloadEnabled, + watermarkEnabled: album.watermarkEnabled, + watermarkConfigVersion: album.watermarkEffectiveVersion, + defaultLocale: album.defaultLocale, + timezone: album.timezone, + startAt: album.startAt.toISOString(), + endAt: album.endAt === null ? null : album.endAt.toISOString(), + translations: [...translations] + .sort((a, b) => (a.locale < b.locale ? -1 : 1)) + .map((t) => ({ + locale: t.locale, + title: t.title, + description: t.description, + venue: t.venue, + address: t.address, + })), + entityVersion: album.entityVersion, + }; +} diff --git a/packages/contracts/src/constants.ts b/packages/contracts/src/constants.ts index 4fbfdc0..6a7eaac 100644 --- a/packages/contracts/src/constants.ts +++ b/packages/contracts/src/constants.ts @@ -4,6 +4,9 @@ /** 照片处理 BullMQ 队列名(api enqueue → worker 消费;补偿扫描 Task 21)。 */ export const QUEUE_PHOTO_PROCESS = 'photo-process'; +/** 水印历史重生成 BullMQ 队列名(P4a Task 3:api watermark-jobs 入队 → worker 消费)。 */ +export const QUEUE_WATERMARK = 'watermark-regen'; + // ── P2 双地域同步与互动(决策初值 D07/D11,验收前可调;env 覆盖键见 .env.example)── // 运行时可调项的默认值集中在此,服务实现(Task 3+)读 env、缺省回落本表; // HEARTBEAT_MS 属 WS 消息契约,随 realtime.ts 导出(此处不重复定义避免重名导出)。 diff --git a/packages/contracts/src/events.ts b/packages/contracts/src/events.ts index 968a39a..abc12c6 100644 --- a/packages/contracts/src/events.ts +++ b/packages/contracts/src/events.ts @@ -47,6 +47,10 @@ export interface AlbumUpdatedPayload { publicEnabled: boolean; downloadEnabled: boolean; originalDownloadEnabled: boolean; + /** P4a A18:水印副本位(ov 模式键服务)。语义见 watermark.ts 注释——取「有效版本」而非最新创建版本。 */ + watermarkEnabled: boolean; + /** ov 侧模式键 `variants/{photoId}/{watermarkConfigVersion}/watermark.jpg` 的版本段。 */ + watermarkConfigVersion: number; defaultLocale: string; timezone: string; startAt: string; // ISO UTC diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 4dd21cd..972c4f4 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -7,4 +7,6 @@ export * from './errors.js'; export * from './events.js'; export * from './internal-sync.js'; export * from './realtime.js'; +export * from './watermark.js'; +export * from './album-payload.js'; export type * from './generated/openapi.js'; diff --git a/packages/contracts/src/watermark.ts b/packages/contracts/src/watermark.ts new file mode 100644 index 0000000..c9feefa --- /dev/null +++ b/packages/contracts/src/watermark.ts @@ -0,0 +1,114 @@ +// packages/contracts/src/watermark.ts —— 水印配置的跨服务共享形状与缺省值 +// (P4a Task 3)。三个消费方经同一模块引用,杜绝形状漂移: +// · api albums 域(POST watermark-jobs:DTO 校验 → normalize 落库); +// · worker 重生成消费者(读 watermark_configs.config Json → 本形状使用); +// · worker 发布路径(新照片按相册当前配置即产水印变体)。 +// +// 版本语义(本文件是权威表述,实现侧注释引用): +// · albums.watermarkVersion = 最新「已创建」配置版本——POST 即推进,新照片 +// 发布立刻套用(契约 WatermarkJobCreated.configVersion 描述:新照片即刻套用); +// · albums.watermarkEnabled + watermarkEffectiveVersion = 「有效」版本——仅当该 +// 版本的历史重生成任务 done(或创建时相册无已发布照片、任务即建即 done)才 +// 推进。ov 副本与公开侧的模式键/变体选择以有效版本为准:对象键 +// `variants/{photoId}/{version}/watermark.jpg` 只有 done 任务保证存在, +// 用有效版本杜绝「同步了新版本号但对象尚未生成」的破图窗口(A20: +// 任务完成前旧浏览图保持可用,失败照片保留旧图)。 +// · enabled=false 也是一版配置(记录关闭动作):关闭后发布的新照片无水印; +// 关闭版重生成任务 = 逐张删除 watermark 变体行(对象保留,历史可溯)。 +export const WATERMARK_POSITIONS = [ + 'topLeft', + 'topRight', + 'bottomLeft', + 'bottomRight', + 'center', +] as const; +export type WatermarkPosition = (typeof WATERMARK_POSITIONS)[number]; + +// ── 服务端缺省值(契约 WatermarkConfig「可选字段缺省用服务端默认值」的唯一权威)── +export const WATERMARK_DEFAULT_POSITION: WatermarkPosition = 'bottomRight'; +export const WATERMARK_DEFAULT_OPACITY = 50; +export const WATERMARK_DEFAULT_SCALE_PERCENT = 20; +export const WATERMARK_DEFAULT_FONT_SIZE = 24; + +/** + * 归一化后的存储形状(watermark_configs.config)。API 保存时经 + * normalizeWatermarkConfig 填齐缺省值落库——worker 与 GET 读到的永远是完整 + * 配置,不再各自解释缺省。enabled=false 时 text.content 允许为空串 + * (关闭语义不绘制文字,其余字段仅作回显)。 + */ +export interface StoredWatermarkConfig { + enabled: boolean; + position: WatermarkPosition; + /** 不透明度百分比 0..100。 */ + opacity: number; + /** 水印层宽度相对浏览图宽度的百分比 5..50。 */ + scalePercent: number; + text: { + content: string; + /** 名义字号 8..72(渲染后再按 scalePercent 缩放,见 worker 实现)。 */ + fontSize: number; + }; +} + +/** 契约 WatermarkConfig(创建请求 / 当前配置响应)。 */ +export type WatermarkConfigInput = { + enabled: boolean; + position?: WatermarkPosition; + opacity?: number; + scalePercent?: number; + text?: { content?: string; fontSize?: number }; +}; + +/** 缺省值填齐(不改写已给值;越界值由 API DTO 层 400 拒绝,此处不重复校验)。 */ +export function normalizeWatermarkConfig( + raw: WatermarkConfigInput, +): StoredWatermarkConfig { + return { + enabled: raw.enabled, + position: raw.position ?? WATERMARK_DEFAULT_POSITION, + opacity: raw.opacity ?? WATERMARK_DEFAULT_OPACITY, + scalePercent: raw.scalePercent ?? WATERMARK_DEFAULT_SCALE_PERCENT, + text: { + content: raw.text?.content ?? '', + fontSize: raw.text?.fontSize ?? WATERMARK_DEFAULT_FONT_SIZE, + }, + }; +} + +/** + * 读取侧解析(watermark_configs.config Json → StoredWatermarkConfig):行由 + * normalizeWatermarkConfig 写入、形状可信,此解析仅对脏数据防御——缺字段回 + * 落缺省值,结构破坏(非对象 / enabled 非 boolean)返回 null(调用方按 + * 「配置不可用」处理:重生成任务整败 / 公开侧回落无水印)。 + */ +export function parseStoredWatermarkConfig( + value: unknown, +): StoredWatermarkConfig | null { + if (typeof value !== 'object' || value === null) { + return null; + } + const raw = value as Partial & { + text?: Partial; + }; + if (typeof raw.enabled !== 'boolean') { + return null; + } + const num = (v: unknown): number | undefined => + typeof v === 'number' && Number.isFinite(v) ? v : undefined; + const str = (v: unknown): string | undefined => + typeof v === 'string' ? v : undefined; + return normalizeWatermarkConfig({ + enabled: raw.enabled, + position: + raw.position !== undefined && + (WATERMARK_POSITIONS as readonly string[]).includes(raw.position) + ? raw.position + : undefined, + opacity: num(raw.opacity), + scalePercent: num(raw.scalePercent), + text: { + content: str(raw.text?.content) ?? '', + fontSize: num(raw.text?.fontSize), + }, + }); +} diff --git a/packages/db-cn/prisma/migrations/20260920090329_watermark_versioning/migration.sql b/packages/db-cn/prisma/migrations/20260920090329_watermark_versioning/migration.sql new file mode 100644 index 0000000..c01df73 --- /dev/null +++ b/packages/db-cn/prisma/migrations/20260920090329_watermark_versioning/migration.sql @@ -0,0 +1,39 @@ +-- P4a Task 3:水印版本化——albums 双版本位列(watermarkVersion 已有 P1 起, +-- 本迁移补「有效版本位」watermarkEnabled/watermarkEffectiveVersion,语义见 +-- packages/contracts/src/watermark.ts)+ 配置版本账本 watermark_configs + +-- 历史重生成任务 watermark_jobs(id 兼作对外 jobId 与 BullMQ jobId)。 +-- 校对手决:Prisma 生成 DDL 固定 utf8mb4_unicode_ci,逐表改写为库级约定的 +-- utf8mb4_0900_bin(albumId/id/errorCode 等字符串列的大小写敏感语义依赖它; +-- 新增两列非 VARCHAR,无校对关注)。 +-- AlterTable +ALTER TABLE `albums` ADD COLUMN `watermarkEffectiveVersion` INTEGER NOT NULL DEFAULT 0, + ADD COLUMN `watermarkEnabled` BOOLEAN NOT NULL DEFAULT false; + +-- CreateTable +CREATE TABLE `watermark_configs` ( + `id` VARCHAR(191) NOT NULL, + `albumId` VARCHAR(64) NOT NULL, + `configVersion` INTEGER NOT NULL, + `config` JSON NOT NULL, + `createdAt` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + + UNIQUE INDEX `watermark_configs_albumId_configVersion_key`(`albumId`, `configVersion`), + PRIMARY KEY (`id`) +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin; + +-- CreateTable +CREATE TABLE `watermark_jobs` ( + `id` VARCHAR(191) NOT NULL, + `albumId` VARCHAR(64) NOT NULL, + `configVersion` INTEGER NOT NULL, + `state` VARCHAR(16) NOT NULL DEFAULT 'pending', + `progress` INTEGER NOT NULL DEFAULT 0, + `total` INTEGER NOT NULL DEFAULT 0, + `failed` INTEGER NOT NULL DEFAULT 0, + `errorCode` VARCHAR(64) NULL, + `createdAt` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + `updatedAt` DATETIME(6) NOT NULL, + + INDEX `watermark_jobs_albumId_createdAt_idx`(`albumId`, `createdAt`), + PRIMARY KEY (`id`) +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin; diff --git a/packages/db-cn/prisma/schema.prisma b/packages/db-cn/prisma/schema.prisma index 6e0ee61..faa22c1 100644 --- a/packages/db-cn/prisma/schema.prisma +++ b/packages/db-cn/prisma/schema.prisma @@ -65,7 +65,15 @@ model albums { publicEnabled Boolean @default(true) downloadEnabled Boolean @default(true) originalDownloadEnabled Boolean @default(false) + // ── 水印版本化(P4a Task 3)── 双版本位语义(权威表述见 contracts watermark.ts): + // watermarkVersion = 最新「已创建」配置版本(POST 即推进,新照片发布即刻套用, + // 也就是契约 AlbumSummary.watermarkVersion / WatermarkJobCreated.configVersion); + // watermarkEnabled + watermarkEffectiveVersion = 「有效」版本位——仅当该版本的 + // 历史重生成任务 done 才推进(worker 完成事务维护,禁用版同样推进以翻转 + // enabled=false)。ov 副本与公开侧变体选择以有效版本为准(对象存在性保证)。 watermarkVersion Int @default(0) + watermarkEnabled Boolean @default(false) + watermarkEffectiveVersion Int @default(0) entityVersion Int @default(1) // P2 Task 3:album.updated 事件版本(同 photos 模式) // P2 Task 9 统计回传副本位(ov→cn):viewCountReplica 为 ov album_stats // 权威值的替换式副本;statsVersion 是回传替换版本(仅更高 statVersion 替换, @@ -205,6 +213,37 @@ model photo_jobs { @@unique([photoId, type]) @@index([state, nextRetryAt]) } +// —— 水印版本化(P4a Task 3)—— +// watermark_configs:配置版本账本(append-only,每 POST 一行,config 为归一化 +// StoredWatermarkConfig——contracts watermark.ts 的形状,缺省已填齐); +// watermark_jobs:历史重生成任务(id 兼作对外 jobId 与 BullMQ jobId——cuid 无 +// ':',满足 BullMQ 保留分隔符约束)。state 语义:pending=已入队待消费; +// running=消费中(progress/failed 随照片逐张推进);done=全部处理完成(允许 +// failed>0 的部分成功——失败张数记录在案,重试=重新 POST 新任务,按照片粒度 +// 幂等);failed=整任务中止(配置缺失等任务级错误,errorCode 必有)。 +model watermark_configs { + id String @id @default(cuid()) + albumId String @db.VarChar(64) + configVersion Int + config Json + createdAt DateTime @default(now()) @db.DateTime(6) + + @@unique([albumId, configVersion]) +} +model watermark_jobs { + id String @id @default(cuid()) + albumId String @db.VarChar(64) + configVersion Int + state String @default("pending") @db.VarChar(16) // pending|running|done|failed + progress Int @default(0) + total Int @default(0) + failed Int @default(0) + errorCode String? @db.VarChar(64) + createdAt DateTime @default(now()) @db.DateTime(6) + updatedAt DateTime @updatedAt @db.DateTime(6) + + @@index([albumId, createdAt]) +} model outbox_events { id String @id @default(cuid()) type String @db.VarChar(64) diff --git a/packages/db-ov/prisma/migrations/20260920090513_watermark_replica_columns/migration.sql b/packages/db-ov/prisma/migrations/20260920090513_watermark_replica_columns/migration.sql new file mode 100644 index 0000000..ecc9b96 --- /dev/null +++ b/packages/db-ov/prisma/migrations/20260920090513_watermark_replica_columns/migration.sql @@ -0,0 +1,8 @@ +-- P4a Task 3:albums_ov 水印副本位(album.updated 载荷投影,语义见 +-- packages/contracts/src/watermark.ts)——watermarkEnabled=true 时 ov 公开侧 +-- 浏览键切换为模式键 `variants/{photoId}/{watermarkConfigVersion}/watermark.jpg` +-- (确定性推导,无逐照片同步;对象存在性由 cn 侧「有效版本」保证)。 +-- 两列均非 VARCHAR,无校对手决关注(沿用库级 utf8mb4_0900_bin 表默认)。 +-- AlterTable +ALTER TABLE `albums_ov` ADD COLUMN `watermarkConfigVersion` INTEGER NOT NULL DEFAULT 0, + ADD COLUMN `watermarkEnabled` BOOLEAN NOT NULL DEFAULT false; diff --git a/packages/db-ov/prisma/schema.prisma b/packages/db-ov/prisma/schema.prisma index f9544fe..75051c7 100644 --- a/packages/db-ov/prisma/schema.prisma +++ b/packages/db-ov/prisma/schema.prisma @@ -38,6 +38,12 @@ model albums_ov { publicEnabled Boolean @default(true) downloadEnabled Boolean @default(true) originalDownloadEnabled Boolean @default(false) + // P4a Task 3 水印副本位(album.updated 载荷投影;语义见 contracts watermark.ts): + // 有效版本(存在 done 重生成任务)——watermarkEnabled=true 时 ov 公开侧浏览键 + // 用模式键 `variants/{photoId}/{watermarkConfigVersion}/watermark.jpg`(确定性 + // 推导,无逐照片同步;对象存在性由有效版本保证)。 + watermarkEnabled Boolean @default(false) + watermarkConfigVersion Int @default(0) emergencyDisabled Boolean @default(false) translations Json // [{locale,title,description,venue,address}](cn 原文数组) categories Json // [{id,sortOrder,translations:[{locale,name}]}] diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index f4bd71d..f285676 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -9,6 +9,12 @@ export { MinioStorageAdapter } from './minio.adapter'; export type { MinioStorageConfig } from './minio.adapter'; export { StorageError, ErrorCodes } from './errors'; export type { StoragePort, UploadCredential } from './port'; +export { + OBJECT_KEY_ROOT, + variantsObjectKey, + watermarkVariantObjectKey, +} from './keys'; +export type { PhotoVariantName } from './keys'; /** * 从 S3_* 环境变量构造 MinIO/S3 兼容适配器(仓库根 .env 经 dotenv-cli 或 diff --git a/packages/storage/src/keys.ts b/packages/storage/src/keys.ts new file mode 100644 index 0000000..c4283aa --- /dev/null +++ b/packages/storage/src/keys.ts @@ -0,0 +1,35 @@ +// packages/storage/src/keys.ts —— 对象键布局的跨服务权威段(P4a Task 3 起)。 +// 此前 OBJECT_KEY_ROOT 在 api uploads/support.ts 与 worker publish.service.ts +// 各持一份字面量;水印引入第三处消费方(ov 公开侧模式键推导,无任何 DB 行可 +// 依赖),收敛到存储包单一出处,杜绝多环境前缀化时(P5)三处漂移。 +// P1a 单环境(env=dev)硬编码;多环境前缀化属 P5 部署事项。 + +/** 对象键根(与各 test/reset.ts 的 MinIO 清理前缀一致)。 */ +export const OBJECT_KEY_ROOT = 'photo-live/dev'; + +/** photo_variants 变体名(与 db-cn photo_variants.variant 取值一致)。 */ +export type PhotoVariantName = 'thumb' | 'browse' | 'watermark'; + +/** 变体对象键(worker 写入位 / ov 模式键推导位,同一实现)。 */ +export function variantsObjectKey( + albumId: string, + photoId: string, + version: number, + variant: PhotoVariantName, +): string { + return `${OBJECT_KEY_ROOT}/albums/${albumId}/variants/${photoId}/${version}/${variant}.jpg`; +} + +/** + * 水印变体键的便捷推导(P4a Task 3 ov 侧核心约定):键由 (albumId, photoId, + * configVersion) 确定性构成——ov 副本不逐照片同步变体键,公开侧按相册有效 + * watermarkConfigVersion 直接推导(对象存在性由 cn 侧有效版本语义保证,见 + * packages/contracts/src/watermark.ts)。 + */ +export function watermarkVariantObjectKey( + albumId: string, + photoId: string, + configVersion: number, +): string { + return variantsObjectKey(albumId, photoId, configVersion, 'watermark'); +} diff --git a/services/api/src/albums/albums.module.ts b/services/api/src/albums/albums.module.ts index 0ca4c6c..5746c34 100644 --- a/services/api/src/albums/albums.module.ts +++ b/services/api/src/albums/albums.module.ts @@ -1,7 +1,9 @@ // services/api/src/albums/albums.module.ts —— 相册模块(Task 14 起步,Task 15 // 补全,Task 16 增补 PATCH,Task 17 增补成员/分类,P1b Task 3 增补资产上传, -// P4a Task 2 增补 admin 交接端点)。PermissionsService(A02 权限基石)+ -// AlbumsService/MembersService/CategoriesService/AssetsService 与对应控制器。 +// P4a Task 2 增补 admin 交接端点,P4a Task 3 增补水印配置/重生成端点)。 +// PermissionsService(A02 权限基石)+ +// AlbumsService/MembersService/CategoriesService/AssetsService/ +// WatermarkService 与对应控制器。 // export PermissionsService 供 uploads/photos 等后续模块经模块间 DI 复用同一 // 份权限判定逻辑;AlbumsService 等暂无外部消费者不 export。 // StoragePort 装配说明(P1b Task 3):UploadsModule 为 PermissionsService 反向 @@ -18,6 +20,9 @@ import { CategoriesService } from './categories.service'; import { MembersController } from './members.controller'; import { MembersService } from './members.service'; import { PermissionsService } from './permissions.service'; +import { WatermarkController } from './watermark.controller'; +import { WatermarkService } from './watermark.service'; +import { watermarkQueueProvider } from './watermark-queue.provider'; import { storageProvider } from '../uploads/storage.provider'; @Module({ @@ -27,6 +32,7 @@ import { storageProvider } from '../uploads/storage.provider'; CategoriesController, AssetsController, AdminAlbumsController, + WatermarkController, ], providers: [ PermissionsService, @@ -34,6 +40,8 @@ import { storageProvider } from '../uploads/storage.provider'; MembersService, CategoriesService, AssetsService, + WatermarkService, + watermarkQueueProvider, storageProvider, ], exports: [PermissionsService], diff --git a/services/api/src/albums/albums.service.ts b/services/api/src/albums/albums.service.ts index 16545ba..d3b4350 100644 --- a/services/api/src/albums/albums.service.ts +++ b/services/api/src/albums/albums.service.ts @@ -173,6 +173,8 @@ export class AlbumsService { publicEnabled: true, downloadEnabled: true, originalDownloadEnabled: true, + watermarkEnabled: true, + watermarkEffectiveVersion: true, defaultLocale: true, timezone: true, startAt: true, diff --git a/services/api/src/albums/dto.ts b/services/api/src/albums/dto.ts index 231c5fa..ef22c99 100644 --- a/services/api/src/albums/dto.ts +++ b/services/api/src/albums/dto.ts @@ -317,6 +317,83 @@ export class AlbumTransferRequestDto { keepAsCollaborator!: boolean; } +// ── 水印(P4a Task 3)── + +/** 契约 WatermarkPosition 枚举(与 contracts WATERMARK_POSITIONS 同值)。 */ +export const WATERMARK_POSITION_VALUES = [ + 'topLeft', + 'topRight', + 'bottomLeft', + 'bottomRight', + 'center', +] as const; +export type WatermarkPositionValue = (typeof WATERMARK_POSITION_VALUES)[number]; + +/** 契约 WatermarkTextConfig:content 1..64;fontSize 8..72。 */ +class WatermarkTextConfigDto { + @IsOptional() + @IsString() + @IsNotEmpty() + @MaxLength(64) + content?: string; + + @IsOptional() + @IsInt() + @Min(8) + @Max(72) + fontSize?: number; +} + +/** + * enabled=true 时 text.content 必填(技术方案 7.2:P4a 文字水印——无内容的 + * 开启配置在 API 层即 400,worker 不再解释「开启但无字」的形态)。约束挂在 + * enabled 上(text 缺席时其字段级装饰器不运行,挂 text 无法拒绝缺省)。 + */ +@ValidatorConstraint({ name: 'watermarkTextRequired', async: false }) +class WatermarkTextRequiredConstraint implements ValidatorConstraintInterface { + validate(value: unknown, args: ValidationArguments): boolean { + if (value !== true) { + return true; // 关闭配置无文字要求。 + } + const { text } = args.object as { text?: { content?: unknown } }; + return ( + typeof text?.content === 'string' && (text.content as string).length > 0 + ); + } + + defaultMessage(): string { + return 'enabled=true 时必须提供 text.content(文字水印内容)'; + } +} + +/** 契约 WatermarkConfig(创建请求;数值边界对齐契约 min/max)。 */ +export class WatermarkConfigDto { + @IsBoolean() + @Validate(WatermarkTextRequiredConstraint) + enabled!: boolean; + + @IsOptional() + @IsIn(WATERMARK_POSITION_VALUES) + position?: WatermarkPositionValue; + + @IsOptional() + @IsInt() + @Min(0) + @Max(100) + opacity?: number; + + @IsOptional() + @IsInt() + @Min(5) + @Max(50) + scalePercent?: number; + + @IsOptional() + @ValidateNested() + @Type(() => WatermarkTextConfigDto) + text?: WatermarkTextConfigDto; +} + // ── 资产(P1b Task 3)── /** 契约 AssetKind:banner(观众页宣传图,PublicAlbum.banner 消费方);cover/logo 建链就绪,展示消费随后续阶段。 */ diff --git a/services/api/src/albums/watermark-queue.provider.ts b/services/api/src/albums/watermark-queue.provider.ts new file mode 100644 index 0000000..60b51ff --- /dev/null +++ b/services/api/src/albums/watermark-queue.provider.ts @@ -0,0 +1,86 @@ +// services/api/src/albums/watermark-queue.provider.ts —— watermark-regen 队列 +// 生产者(P4a Task 3)。与 uploads/queue.provider.ts 同一 lazy-safe 约定: +// · Queue 实例(含 ioredis 连接)首次 enqueue 时才创建——Redis 不可用时 +// 应用启动/请求管线不受影响; +// · 入队失败只记日志不抛错:watermark_jobs(pending) 行是持久事实源。与 +// photo-process 不同,本队列暂无补偿扫描(sweep 只认领 photo_jobs)—— +// 孤儿 pending 行的兜底 = 重新 POST 配置(新任务按照片粒度幂等重跑, +// 契约「failed 可重新发起」同语义);补偿扫描并入 sweep 登记 P4b。 +// · ioredis 'error' 挂 no-op 监听(未监听的 error 事件崩进程)。 +// jobId 即 watermark_jobs.id(cuid 无 ':',满足 BullMQ 保留分隔符约束)。 +import { Injectable, Logger, type OnModuleDestroy } from '@nestjs/common'; +import { Queue } from 'bullmq'; +import IORedis from 'ioredis'; +import { QUEUE_WATERMARK } from '@photo/contracts'; + +/** DI token。 */ +export const WATERMARK_QUEUE = Symbol('WATERMARK_QUEUE'); + +/** 生产者窄接口:服务只依赖 enqueue,便于测试替身。 */ +export interface WatermarkQueueProducer { + enqueueWatermarkRegen(jobId: string): Promise; +} + +/** BullMQ job 名称(与 worker 消费者同值)。 */ +const WATERMARK_REGEN_JOB_NAME = 'regen'; + +@Injectable() +export class BullMqWatermarkQueue + implements WatermarkQueueProducer, OnModuleDestroy +{ + private readonly logger = new Logger(BullMqWatermarkQueue.name); + private queue: Queue | null = null; + private connection: IORedis | null = null; + + private getQueue(): Queue { + if (this.queue === null) { + const url = process.env.REDIS_URL_CN ?? 'redis://localhost:6379'; + this.connection = new IORedis(url, { + maxRetriesPerRequest: null, // BullMQ 阻塞命令要求 + }); + this.connection.on('error', (err: Error) => { + this.logger.warn( + `watermark-regen 队列连接异常(将由 ioredis 自动重连): ${err.message}`, + ); + }); + this.queue = new Queue(QUEUE_WATERMARK, { connection: this.connection }); + } + return this.queue; + } + + async enqueueWatermarkRegen(jobId: string): Promise { + try { + await this.getQueue().add( + WATERMARK_REGEN_JOB_NAME, + { jobId }, + { + jobId, // 行 id 即 BullMQ jobId:重复入队天然去重。 + attempts: 3, + backoff: { type: 'exponential', delay: 1000 }, + }, + ); + } catch (err) { + this.logger.error( + `watermark-regen 入队失败(重新 POST 可重跑)jobId=${jobId}: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + } + } + + async onModuleDestroy(): Promise { + if (this.queue !== null) { + await this.queue.close().catch(() => undefined); + this.queue = null; + } + if (this.connection !== null) { + await this.connection.quit().catch(() => undefined); + this.connection = null; + } + } +} + +export const watermarkQueueProvider = { + provide: WATERMARK_QUEUE, + useClass: BullMqWatermarkQueue, +}; diff --git a/services/api/src/albums/watermark.controller.ts b/services/api/src/albums/watermark.controller.ts new file mode 100644 index 0000000..75d6ee9 --- /dev/null +++ b/services/api/src/albums/watermark.controller.ts @@ -0,0 +1,52 @@ +// services/api/src/albums/watermark.controller.ts —— /albums/{albumId}/ +// watermark-jobs 端点(P4a Task 3)。方法名与契约 operationId 一一对应 +// (createWatermarkJob/getWatermarkJobs)。控制器级 SessionGuard:两端点全需 +// 认证。控制器只做认证用户与路径参数透传;权限与 HTTP 映射由 WatermarkService +// 抛 ApiException(404 ALBUM_NOT_FOUND / 403 AUTH_FORBIDDEN / 400 +// VALIDATION_FAILED)。POST 202:任务异步受理(A20 原子切换——完成前旧浏览图 +// 保持可用),进度经 GET 轮询。 +import { + Body, + Controller, + Get, + HttpCode, + HttpStatus, + Param, + Post, + Req, + UseGuards, +} from '@nestjs/common'; +import { AuthenticatedRequest, SessionGuard } from '../auth/session.guard'; +import { userOf } from './albums.controller'; +import { WatermarkConfigDto } from './dto'; +import { + WatermarkService, + type WatermarkJobCreatedResponse, + type WatermarkJobsStatusResponse, +} from './watermark.service'; + +@Controller('albums/:albumId/watermark-jobs') +@UseGuards(SessionGuard) +export class WatermarkController { + constructor(private readonly watermarkService: WatermarkService) {} + + /** 契约 202:配置已保存、任务已受理。 */ + @Post() + @HttpCode(HttpStatus.ACCEPTED) + createWatermarkJob( + @Param('albumId') albumId: string, + @Body() dto: WatermarkConfigDto, + @Req() req: AuthenticatedRequest, + ): Promise { + return this.watermarkService.create(userOf(req), albumId, dto); + } + + /** 当前配置 + 历史任务(创建时间倒序)。 */ + @Get() + getWatermarkJobs( + @Param('albumId') albumId: string, + @Req() req: AuthenticatedRequest, + ): Promise { + return this.watermarkService.getStatus(userOf(req), albumId); + } +} diff --git a/services/api/src/albums/watermark.service.ts b/services/api/src/albums/watermark.service.ts new file mode 100644 index 0000000..7ec9f35 --- /dev/null +++ b/services/api/src/albums/watermark.service.ts @@ -0,0 +1,258 @@ +// services/api/src/albums/watermark.service.ts —— 水印配置与历史重生成任务 +// 端点逻辑(P4a Task 3,A18/A20)。契约权威:openapi /albums/{albumId}/ +// watermark-jobs(createWatermarkJob/getWatermarkJobs)。权限:canViewAlbum +// (404 防枚举)→ canManageAlbum(仅创建者/admin,403)——同相册 PATCH 语义。 +// +// 版本语义权威表述在 packages/contracts/src/watermark.ts(此处摘要): +// · albums.watermarkVersion = 最新「已创建」配置版本——POST 即推进,新照片 +// 发布立刻套用(WatermarkJobCreated.configVersion 描述); +// · albums.watermarkEnabled + watermarkEffectiveVersion = 「有效」版本——仅当 +// 该版本任务 done 才由 worker 完成事务推进(0 张照片的退化任务在创建事务 +// 内即建即 done、当场推进)。ov 副本与公开侧变体选择以有效版本为准。 +// +// create 事务(乐观重试同 albums.service.update:entityVersion 条件 updateMany +// 为串行化点,先占位再写从属行,败方整事务重跑、版本号按新读值重算无空洞): +// ① 占位:albums.watermarkVersion 推进 + entityVersion+1(autoDone 时一并 +// 推进有效版本位); +// ② watermark_configs 追加新版本(config 经 normalizeWatermarkConfig 填齐 +// 缺省——worker/GET 读到的永远是完整形状); +// ③ watermark_jobs 建行(total = 当前已发布照片数;0 张 → 即建即 done); +// ④ outbox album.updated(载荷水印位为推进后的有效版本——ov 只在对象存在性 +// 有保证后看到新版本号;常规优先级:水印变化不影响公开可见性); +// ⑤ audit album.watermark {configVersion, enabled, jobId}。 +// 提交后入队 QUEUE_WATERMARK(失败不回滚:job 行是事实源,见 +// watermark-queue.provider;autoDone 无照片可处理、无需入队)。 +import { HttpStatus, Inject, Injectable } from '@nestjs/common'; +import { + ErrorCodes, + normalizeWatermarkConfig, + parseStoredWatermarkConfig, + WATERMARK_DEFAULT_FONT_SIZE, + WATERMARK_DEFAULT_OPACITY, + WATERMARK_DEFAULT_POSITION, + WATERMARK_DEFAULT_SCALE_PERCENT, +} from '@photo/contracts'; +import type { components } from '@photo/contracts'; +import type { Prisma } from '@photo/db-cn'; +import { ApiException } from '../api-exception'; +import type { AuthUser } from '../auth/session.guard'; +import { PrismaService } from '../prisma/prisma.service'; +import { buildAlbumUpdatedPayload } from '../sync/payloads'; +import type { WatermarkConfigDto } from './dto'; +import { PermissionsService } from './permissions.service'; +import { albumForbidden, albumNotFound } from './support'; +import { + WATERMARK_QUEUE, + type WatermarkQueueProducer, +} from './watermark-queue.provider'; + +export type WatermarkJobCreatedResponse = + components['schemas']['WatermarkJobCreated']; +export type WatermarkJobsStatusResponse = + components['schemas']['WatermarkJobsStatus']; + +/** GET 任务列表上限(契约「历史任务(创建时间倒序)」,前端轮询场景足够)。 */ +export const WATERMARK_JOBS_LIST_LIMIT = 50; + +/** 乐观并发重试上限(同 albums.service.update)。 */ +const OPTIMISTIC_ATTEMPTS = 3; + +@Injectable() +export class WatermarkService { + constructor( + private readonly prisma: PrismaService, + private readonly permissions: PermissionsService, + @Inject(WATERMARK_QUEUE) + private readonly queue: WatermarkQueueProducer, + ) {} + + /** + * 保存新配置版本并发起历史重生成(202:异步受理,进度经 GET 查询)。 + */ + async create( + user: AuthUser, + albumId: string, + dto: WatermarkConfigDto, + ): Promise { + if (!(await this.permissions.canViewAlbum(user, albumId))) { + throw albumNotFound(); + } + if (!(await this.permissions.canManageAlbum(user, albumId))) { + throw albumForbidden('仅创建者或管理员可管理水印'); + } + const config = normalizeWatermarkConfig(dto); + + let jobId = ''; + let configVersion = 0; + let needsWorker = false; + let outcome: 'done' | 'retry' = 'retry'; + for (let attempt = 0; attempt < OPTIMISTIC_ATTEMPTS && outcome === 'retry'; attempt++) { + outcome = await this.prisma.$transaction(async (tx) => { + const album = await tx.albums.findUnique({ where: { id: albumId } }); + if (album === null) { + // admin 视角 canView 恒真,删除竞态兜底同码 404。 + throw albumNotFound(); + } + const nextVersion = album.watermarkVersion + 1; + const total = await tx.photos.count({ + where: { albumId, visibility: 'published' }, + }); + // 0 张已发布照片:无历史可重生成,任务即建即 done、有效版本当场推进 + // (关闭版同样——「空相册关水印」立即生效)。 + const autoDone = total === 0; + + const claimed = await tx.albums.updateMany({ + where: { id: albumId, entityVersion: album.entityVersion }, + data: { + watermarkVersion: nextVersion, + entityVersion: { increment: 1 }, + ...(autoDone + ? { + watermarkEffectiveVersion: nextVersion, + watermarkEnabled: config.enabled, + } + : {}), + }, + }); + if (claimed.count === 0) { + return 'retry' as const; + } + await tx.watermark_configs.create({ + data: { + albumId, + configVersion: nextVersion, + config: config as unknown as Prisma.InputJsonValue, + }, + }); + const job = await tx.watermark_jobs.create({ + data: { + albumId, + configVersion: nextVersion, + state: autoDone ? 'done' : 'pending', + total, + }, + select: { id: true }, + }); + const updated = await tx.albums.findUniqueOrThrow({ + where: { id: albumId }, + }); + const translations = await tx.album_translations.findMany({ + where: { albumId }, + orderBy: { locale: 'asc' }, + }); + await tx.outbox_events.create({ + data: { + type: 'album.updated', + albumId, + entityVersion: updated.entityVersion, + priority: 'normal', + payload: buildAlbumUpdatedPayload( + updated, + translations, + ) as unknown as Prisma.InputJsonValue, + }, + }); + await tx.audit_logs.create({ + data: { + actorId: user.id, + action: 'album.watermark', + resourceId: albumId, + metadata: { + configVersion: nextVersion, + enabled: config.enabled, + jobId: job.id, + }, + }, + }); + jobId = job.id; + configVersion = nextVersion; + needsWorker = !autoDone; + return 'done' as const; + }); + } + if (outcome !== 'done') { + throw new ApiException( + ErrorCodes.INTERNAL_ERROR, + 'Internal server error', + HttpStatus.INTERNAL_SERVER_ERROR, + ); + } + if (needsWorker) { + await this.queue.enqueueWatermarkRegen(jobId); + } + return { jobId, configVersion }; + } + + /** + * 当前配置(最新版本;从未配置 → null)与任务列表(创建时间倒序,上限 50)。 + */ + async getStatus( + user: AuthUser, + albumId: string, + ): Promise { + if (!(await this.permissions.canViewAlbum(user, albumId))) { + throw albumNotFound(); + } + if (!(await this.permissions.canManageAlbum(user, albumId))) { + throw albumForbidden('仅创建者或管理员可查看水印状态'); + } + const album = await this.prisma.albums.findUnique({ + where: { id: albumId }, + select: { id: true }, + }); + if (album === null) { + throw albumNotFound(); + } + const latestConfig = await this.prisma.watermark_configs.findFirst({ + where: { albumId }, + orderBy: { configVersion: 'desc' }, + }); + const jobs = await this.prisma.watermark_jobs.findMany({ + where: { albumId }, + orderBy: { createdAt: 'desc' }, + take: WATERMARK_JOBS_LIST_LIMIT, + }); + return { + current: + latestConfig === null + ? null + : { + configVersion: latestConfig.configVersion, + // 行由 normalizeWatermarkConfig 写入;读取侧仍走解析防御脏数据 + // (结构破坏按 null——「配置不可用」而非 5xx,状态端点不因单行 + // 脏数据整页失败)。 + config: parseStoredWatermarkConfig(latestConfig.config) ?? { + enabled: false, + position: WATERMARK_DEFAULT_POSITION, + opacity: WATERMARK_DEFAULT_OPACITY, + scalePercent: WATERMARK_DEFAULT_SCALE_PERCENT, + text: { + content: '', + fontSize: WATERMARK_DEFAULT_FONT_SIZE, + }, + }, + }, + jobs: jobs.map((job) => ({ + jobId: job.id, + configVersion: job.configVersion, + state: job.state as components['schemas']['WatermarkJobState'], + progress: job.progress, + total: job.total, + failed: job.failed, + ...(job.errorCode === null + ? {} + : { + // worker 任务码(WATERMARK_CONFIG_MISSING 等)是 VarChar(64) 内部 + // 码,契约 Error 枚举未收录——非枚举值统一译 INTERNAL_ERROR 保持 + // 线上响应契约合法(openapi 增补 WATERMARK_* 枚举值登记后续任务)。 + errorCode: ( + Object.values(ErrorCodes) as readonly string[] + ).includes(job.errorCode) + ? (job.errorCode as components['schemas']['ErrorCode']) + : ErrorCodes.INTERNAL_ERROR, + }), + createdAt: job.createdAt.toISOString(), + })), + }; + } +} diff --git a/services/api/src/public-ov/public-ov.service.ts b/services/api/src/public-ov/public-ov.service.ts index f6858e4..fac0802 100644 --- a/services/api/src/public-ov/public-ov.service.ts +++ b/services/api/src/public-ov/public-ov.service.ts @@ -13,6 +13,9 @@ // 生效 banner」):cn 侧语言偏好解析(locale 行 → 全局行)在 ov 副本上 // 退化为最新键——多语言 banner 的 ov 侧精确解析登记为 P5 展示 refinement // (本地仿真 e2e 用单全局 banner 验证同形); +// · 浏览图水印(P4a Task 3):ov 侧无变体副本表,水印开启时 browse 键由 +// 模式推导(variants/{photoId}/{有效版本}/watermark.jpg,cn worker 直写 +// 同一对象空间)——对象存在性由 done 任务收尾不变量保证(见 ovBrowseKey); // · 列表新增 sort=hot(plan Task 6 / 技术方案 §7):likeTotal 倒序 + photoId // 倒序决胜,**短期快照语义**——每相册内存缓存排好序的 photoId 清单 // (TTL 缺省 30s,env HOT_SORT_TTL_MS 可调,惰性读取),窗口内翻页对同一 @@ -36,6 +39,7 @@ import { OvPrismaService } from '../prisma-ov/ov-prisma.module'; import { signedUrlTtlSeconds } from '../uploads/support'; import { albumNotPublic, photoNotPublic } from '../public/support'; import { STORAGE_PORT_OV } from './storage-ov.provider'; +import { watermarkVariantObjectKey } from '@photo/storage'; import type { StoragePort } from '@photo/storage'; export type PublicAlbumResponse = components['schemas']['PublicAlbum']; @@ -80,6 +84,13 @@ interface HotSnapshot { expiresAt: number; } +/** 相册级水印服务上下文(albums_ov 副本位 + 归属 albumId,见 ovBrowseKey)。 */ +interface WatermarkContext { + albumId: string; + enabled: boolean; + configVersion: number; +} + /** * 热门快照 offset 游标:base64url JSON 数字(快照内下一页起始下标)。快照在 * TTL 内冻结,下标分页严格稳定(无并发漂移);解码失败 → null(调用方 400, @@ -205,7 +216,12 @@ export class PublicOvService { : {}; if (query.sort === 'hot') { - return this.listHot(album.albumId, categoryFilter, query); + return this.listHot( + album.albumId, + categoryFilter, + query, + this.watermarkContext(album), + ); } let decoded: ReturnType = null; @@ -230,7 +246,7 @@ export class PublicOvService { }); const hasMore = rows.length > query.limit; const page = hasMore ? rows.slice(0, query.limit) : rows; - const items = await this.toItems(page); + const items = await this.toItems(page, this.watermarkContext(album)); // 游标取自「DB 页末行」(同 cn:被跳过的行不进下一页重扫)。 const last = page.at(-1); const nextCursor = @@ -270,7 +286,13 @@ export class PublicOvService { width: photo.width, height: photo.height, thumb: await this.presign(photo.thumbObjectKey), - browse: await this.presign(photo.browseObjectKey), + browse: await this.presign( + this.ovBrowseKey( + this.watermarkContext(album), + photo.photoId, + photo.browseObjectKey, + ), + ), likeCount: photo.likeTotal, publishedAt: (photo.publishedAt ?? new Date(0)).toISOString(), categoryId: photo.categoryId, @@ -282,6 +304,7 @@ export class PublicOvService { albumId: string, categoryFilter: { categoryId?: string }, query: { cursor?: string; limit: number }, + watermark: WatermarkContext, ): Promise { let offset = 0; if (query.cursor !== undefined && query.cursor !== '') { @@ -305,7 +328,7 @@ export class PublicOvService { const ordered = pageIds .map((id) => byId.get(id)) .filter((row) => row !== undefined); - const items = await this.toItems(ordered); + const items = await this.toItems(ordered, watermark); const nextCursor = hasMore ? encodeHotCursor(offset + query.limit) : null; return { items, nextCursor, hasMore: nextCursor !== null }; } @@ -347,6 +370,7 @@ export class PublicOvService { likeTotal: number; publishedAt: Date | null; }>, + watermark: WatermarkContext, ): Promise { const items: PublicPhotoListResponse['items'] = []; for (const photo of page) { @@ -358,7 +382,9 @@ export class PublicOvService { width: photo.width, height: photo.height, thumb: await this.presign(photo.thumbObjectKey), - browse: await this.presign(photo.browseObjectKey), + browse: await this.presign( + this.ovBrowseKey(watermark, photo.photoId, photo.browseObjectKey), + ), likeCount: photo.likeTotal, publishedAt: (photo.publishedAt ?? new Date(0)).toISOString(), }); @@ -366,6 +392,42 @@ export class PublicOvService { return items; } + /** albums_ov 水印副本位(watermarkEnabled + watermarkConfigVersion + 归属)。 */ + private watermarkContext(album: { + albumId: string; + watermarkEnabled: boolean; + watermarkConfigVersion: number; + }): WatermarkContext { + return { + albumId: album.albumId, + enabled: album.watermarkEnabled, + configVersion: album.watermarkConfigVersion, + }; + } + + /** + * ov 浏览图对象键(P4a Task 3):水印开启时用**模式键**—— + * variants/{photoId}/{watermarkConfigVersion}/watermark.jpg(同一对象空间, + * cn worker 直写;ov 无变体副本表,键由 cn 行结构确定)。副本位携带的是 + * **有效版本**(albums.watermarkEffectiveVersion,仅 done 任务推进):该版本 + * 键的存在性由重生成任务的收尾不变量保证(每张已发布照片处理成功或回退拷贝 + * 旧图到目标键——失败照片保留旧图,A20),disable 版任务完成后 enabled=false + * 回落 browse 内联键(水印键对象保留,历史可溯)。 + */ + private ovBrowseKey( + watermark: WatermarkContext, + photoId: string, + browseObjectKey: string, + ): string { + return watermark.enabled && watermark.configVersion > 0 + ? watermarkVariantObjectKey( + watermark.albumId, + photoId, + watermark.configVersion, + ) + : browseObjectKey; + } + /** publicToken → albums_ov 副本;不存在 / 不可见(含紧急停用)统一 404。 */ private async findPublicAlbumOv(token: string) { const album = await this.prisma.albums_ov.findUnique({ diff --git a/services/api/src/public/public.service.ts b/services/api/src/public/public.service.ts index 1251654..a41f7db 100644 --- a/services/api/src/public/public.service.ts +++ b/services/api/src/public/public.service.ts @@ -222,6 +222,15 @@ export class PublicService { bucket.set(row.variant, row); } + // P4a Task 3 水印浏览图选择(cn):albums.watermarkEnabled(有效版本开关) + // 且照片有 watermark 变体行 → 现签水印对象键;否则 browse。行是对象 put + // 成功后才落库的(worker 先写对象再 upsert 行),故行存在 ⇒ 对象存在; + // 失败照片行保留旧版本键(重生成失败不覆盖行——A20「失败照片保留旧图」), + // 新发布照片行在最新版本键(WatermarkJobCreated:新照片即刻套用)——两者 + // 都直接用行的 objectKey,无需版本比较。disable 版任务逐张删行:删到哪张 + // 哪张回落 browse(渐进撤下,语义见 contracts watermark.ts)。 + const serveWatermark = album.watermarkEnabled; + const items: PublicPhotoListResponse['items'] = []; for (const photo of page) { const bucket = variantsByPhoto.get(photo.id); @@ -230,8 +239,14 @@ export class PublicService { if (thumbRow === undefined || browseRow === undefined) { continue; } + const watermarkRow = serveWatermark + ? bucket?.get('watermark') + : undefined; + const browseKey = + watermarkRow !== undefined ? watermarkRow.objectKey : browseRow.objectKey; // 宽高取 browse 变体尺寸(契约「浏览图宽高,用于列表占位」);变体行缺 // 尺寸时回退 photos 主行(原始尺寸),两侧均缺 → 同样跳过(必返字段)。 + // 水印图与 browse 同源同尺寸(compose 基于 browse 管线),占位一致。 const width = browseRow.width ?? photo.width; const height = browseRow.height ?? photo.height; if (width === null || height === null) { @@ -242,7 +257,7 @@ export class PublicService { width, height, thumb: await this.presign(thumbRow.objectKey), - browse: await this.presign(browseRow.objectKey), + browse: await this.presign(browseKey), likeCount: photo.likeCount, // published 行的 publishedAt 必非空(worker 发布事务写入);?? 兜底仅 // 收窄类型,不构成可观察行为。 @@ -290,6 +305,11 @@ export class PublicService { }); const thumbRow = variants.find((v) => v.variant === 'thumb'); const browseRow = variants.find((v) => v.variant === 'browse'); + // 水印浏览图选择同列表(有效开关 + 行存在 ⇒ 用行的对象键;A20 失败照片 + // 保留旧图——行未被失败任务覆盖,仍指向旧版本键)。 + const watermarkRow = album.watermarkEnabled + ? variants.find((v) => v.variant === 'watermark') + : undefined; const width = browseRow?.width ?? photo.width; const height = browseRow?.height ?? photo.height; if ( @@ -301,12 +321,14 @@ export class PublicService { // post-publish 不变量破坏:公开面按「不可展示」处理(同列表跳过语义)。 throw photoNotPublic(); } + const browseKey = + watermarkRow !== undefined ? watermarkRow.objectKey : browseRow.objectKey; return { photoId: photo.id, width, height, thumb: await this.presign(thumbRow.objectKey), - browse: await this.presign(browseRow.objectKey), + browse: await this.presign(browseKey), likeCount: photo.likeCount, publishedAt: (photo.publishedAt ?? new Date(0)).toISOString(), categoryId: photo.categoryId, diff --git a/services/api/src/sync-inbox/sync-inbox.service.ts b/services/api/src/sync-inbox/sync-inbox.service.ts index 840450c..68e55b7 100644 --- a/services/api/src/sync-inbox/sync-inbox.service.ts +++ b/services/api/src/sync-inbox/sync-inbox.service.ts @@ -221,6 +221,13 @@ export class SyncInboxService { const categorySnapshot = Array.isArray(p.categories) ? (p.categories as unknown as Prisma.InputJsonValue) : undefined; + // P4a Task 3 水印副本位:旧事件/手写载荷可能缺字段——防御性归一(缺省= + // 关闭/0),语义上「缺水印位」与「水印关闭」同形。 + const watermarkEnabled = p.watermarkEnabled === true; + const watermarkConfigVersion = + typeof p.watermarkConfigVersion === 'number' && p.watermarkConfigVersion > 0 + ? Math.floor(p.watermarkConfigVersion) + : 0; await tx.albums_ov.upsert({ where: { albumId: event.entityId }, create: { @@ -230,6 +237,8 @@ export class SyncInboxService { publicEnabled: p.publicEnabled, downloadEnabled: p.downloadEnabled, originalDownloadEnabled: p.originalDownloadEnabled, + watermarkEnabled, + watermarkConfigVersion, translations: p.translations as unknown as Prisma.InputJsonValue, categories: categorySnapshot ?? [], defaultLocale: p.defaultLocale, @@ -244,6 +253,8 @@ export class SyncInboxService { publicEnabled: p.publicEnabled, downloadEnabled: p.downloadEnabled, originalDownloadEnabled: p.originalDownloadEnabled, + watermarkEnabled, + watermarkConfigVersion, translations: p.translations as unknown as Prisma.InputJsonValue, ...(categorySnapshot !== undefined ? { categories: categorySnapshot } diff --git a/services/api/src/sync/payloads.ts b/services/api/src/sync/payloads.ts index 704f935..fdb3b3f 100644 --- a/services/api/src/sync/payloads.ts +++ b/services/api/src/sync/payloads.ts @@ -10,7 +10,7 @@ // bootstrap/reconcile 同一出口);worker publish.service 的首次发布载荷来自 // 流水线内存值(同构不同源),保持不动。 import { HttpStatus } from '@nestjs/common'; -import { ErrorCodes } from '@photo/contracts'; +import { ErrorCodes, buildAlbumUpdatedPayloadCore } from '@photo/contracts'; import type { AlbumUpdatedPayload, AssetUpdatedPayload, @@ -31,6 +31,8 @@ type AlbumRow = Pick< | 'publicEnabled' | 'downloadEnabled' | 'originalDownloadEnabled' + | 'watermarkEnabled' + | 'watermarkEffectiveVersion' | 'defaultLocale' | 'timezone' | 'startAt' @@ -52,6 +54,8 @@ type PhotoVariantRow = Prisma.photo_variantsGetPayload< /** * album.updated 载荷:albums_ov 展示所需标量 + 翻译全集(locale 升序)。 + * P4a Task 3 起标量全集(含水印副本位)的构建收敛到 contracts 的共享实现 + * (worker 完成事务同用,见 buildAlbumUpdatedPayloadCore);本包装仅叠加 * categories 可选参数:仅 bootstrap/reconcile 快照事件传入(ov 整体替换 * albums_ov.categories 的收敛原语,见 contracts events.ts 注释);业务事务内 * 增量事件不传(分类元素归 category.* 事件管)。输出按 (sortOrder, id) 升序 @@ -62,48 +66,26 @@ export function buildAlbumUpdatedPayload( translations: AlbumTranslationRow[], categories?: AlbumCategoryInput[], ): AlbumUpdatedPayload { - return { - albumId: album.id, - publicToken: album.publicToken, - status: album.status, - uploadEnabled: album.uploadEnabled, - publicEnabled: album.publicEnabled, - downloadEnabled: album.downloadEnabled, - originalDownloadEnabled: album.originalDownloadEnabled, - defaultLocale: album.defaultLocale, - timezone: album.timezone, - startAt: album.startAt.toISOString(), - endAt: album.endAt === null ? null : album.endAt.toISOString(), - translations: [...translations] - .sort((a, b) => (a.locale < b.locale ? -1 : 1)) - .map((t) => ({ - locale: t.locale, - title: t.title, - description: t.description, - venue: t.venue, - address: t.address, - })), - ...(categories === undefined - ? {} - : { - categories: [...categories] - .sort((a, b) => - a.category.sortOrder !== b.category.sortOrder - ? a.category.sortOrder - b.category.sortOrder - : a.category.id < b.category.id - ? -1 - : 1, - ) - .map(({ category, translations: categoryTranslations }) => ({ - id: category.id, - sortOrder: category.sortOrder, - translations: [...categoryTranslations] - .sort((a, b) => (a.locale < b.locale ? -1 : 1)) - .map((t) => ({ locale: t.locale, name: t.name })), - })), - }), - entityVersion: album.entityVersion, - }; + const payload = buildAlbumUpdatedPayloadCore(album, translations); + if (categories === undefined) { + return payload; + } + payload.categories = [...categories] + .sort((a, b) => + a.category.sortOrder !== b.category.sortOrder + ? a.category.sortOrder - b.category.sortOrder + : a.category.id < b.category.id + ? -1 + : 1, + ) + .map(({ category, translations: categoryTranslations }) => ({ + id: category.id, + sortOrder: category.sortOrder, + translations: [...categoryTranslations] + .sort((a, b) => (a.locale < b.locale ? -1 : 1)) + .map((t) => ({ locale: t.locale, name: t.name })), + })); + return payload; } /** 快照 categories 参数形状:分类行 + 该分类的翻译行(见函数注释)。 */ diff --git a/services/api/src/uploads/support.ts b/services/api/src/uploads/support.ts index c1a5a5c..c9bc719 100644 --- a/services/api/src/uploads/support.ts +++ b/services/api/src/uploads/support.ts @@ -6,11 +6,15 @@ import { randomBytes } from 'node:crypto'; import { HttpStatus } from '@nestjs/common'; import { ErrorCodes } from '@photo/contracts'; +import { OBJECT_KEY_ROOT } from '@photo/storage'; import { ApiException } from '../api-exception'; import { albumNotFound } from '../albums/support'; -/** 对象键根(与 test/reset.ts 的 MinIO 清理前缀一致)。 */ -export const OBJECT_KEY_ROOT = 'photo-live/dev'; +/** + * 对象键根:P4a Task 3 起权威出处收敛到 @photo/storage keys.ts(ov 公开侧 + * 水印模式键推导引入第三消费方),此处转出口维持既有导入面。 + */ +export { OBJECT_KEY_ROOT }; /** UPLOAD_MAX_BYTES 缺省 50MB(与 .env.example 同步)。 */ export const DEFAULT_UPLOAD_MAX_BYTES = 52_428_800; diff --git a/services/api/test/e2e.full-chain.e2e-spec.ts b/services/api/test/e2e.full-chain.e2e-spec.ts index 66e3d88..ffeaa83 100644 --- a/services/api/test/e2e.full-chain.e2e-spec.ts +++ b/services/api/test/e2e.full-chain.e2e-spec.ts @@ -19,6 +19,7 @@ import { PrismaClient } from '@photo/db-cn'; import { ErrorCodes } from '@photo/contracts'; import { createStorageFromEnv, + variantsObjectKey, type MinioStorageAdapter, } from '@photo/storage'; import { AppCnModule } from '../src/app-cn.module'; @@ -28,10 +29,7 @@ import { API_PREFIX } from '../src/shared/constants'; import { createPhotoProcessWorker, } from '../../worker/src/process.processor'; -import { - PublishService, - variantsObjectKey, -} from '../../worker/src/publish.service'; +import { PublishService } from '../../worker/src/publish.service'; import { resetDb } from './reset'; jest.setTimeout(180_000); diff --git a/services/api/test/reset.ts b/services/api/test/reset.ts index 00ca5fb..4cd8775 100644 --- a/services/api/test/reset.ts +++ b/services/api/test/reset.ts @@ -24,6 +24,10 @@ export const RESET_TABLES: readonly string[] = [ 'photo_variants', 'upload_sessions', 'photos', + // 水印域(P4a Task 3):任务/配置账本先于 albums(watermarkVersion 等位列 + // 随 albums 清零)。 + 'watermark_jobs', + 'watermark_configs', // 相册域(Task 9):子表(翻译/资产/成员)在前,albums 在后;相册域为扁平 // 字符串外键(无 FK 约束),此处顺序仅为逻辑可读,不依赖 FK 检查开关。 'category_translations', diff --git a/services/api/test/watermark.e2e-spec.ts b/services/api/test/watermark.e2e-spec.ts new file mode 100644 index 0000000..8dedeca --- /dev/null +++ b/services/api/test/watermark.e2e-spec.ts @@ -0,0 +1,687 @@ +// services/api/test/watermark.e2e-spec.ts —— P4a Task 3 水印版本化 + 历史重生成 +// 全链路 e2e。ONE spec 串完整业务流(cn API → worker 重生成(直驱,BullMQ 消费 +// 由 worker/test/watermark.e2e-spec.ts 覆盖)→ cn/ov 公开侧服务 → ov 副本传播)。 +// 进程内起两个真实入口(cn + ov,同 public-ov.e2e-spec 模式:OutboxSender 指向 +// ov 真实 URL,ov 读到的数据与生产链路同源)。覆盖(任务验收): +// ① POST watermark-jobs(enabled):202 {jobId, configVersion};GET 状态形状 +// (current=最新配置缺省填齐 / jobs 创建时间倒序 / pending+total); +// ② 新照片发布即套用最新版(POST 后、任务 done 前发布 → watermark 变体对象 +// 已产且与 browse 字节不同——契约 WatermarkJobCreated 语义); +// ③ 历史重生成(3 张):直驱 processJob → done(progress=total=3, failed=0) +// → cn 公开列表 browse 现签 URL 取回**水印字节**(≠ browse 变体对象); +// ④ A20:原图 objectKey 与字节全程零触碰; +// ⑤ ov 传播:drainSender 后 albums_ov.watermarkEnabled/ConfigVersion 就位; +// ov 公开列表 browse URL = 模式键 variants/{photoId}/{v}/watermark.jpg, +// 取回字节与 cn 水印对象一致; +// ⑥ 失败注入:删 1 张原图 → 新版本任务 done 且 failed=1;失败张 cn 保留旧版 +// 水印(行未覆盖)、ov 模式键 = 旧图拷贝(同字节——失败照片保留旧图); +// 恢复原图后 re-POST → failed=0 全量收敛; +// ⑦ 关闭版重生成:done 后 cn/ov 公开列表回落 browse 键;水印行删除、对象 +// 保留(历史可溯); +// ⑧ 空相册:POST 即建即 done(无照片可重生成),有效位当场推进并同步 ov; +// ⑨ 权限:协作者 POST/GET → 403;圈外人 → 404;admin → 202/200; +// ⑩ 校验:enabled=true 缺 text.content → 400;opacity/scalePercent/fontSize +// 越界、position 非法 → 400。 +import { randomBytes } from 'node:crypto'; +import { Readable } from 'node:stream'; +import { DeleteObjectCommand, S3Client } from '@aws-sdk/client-s3'; +import { NestFactory } from '@nestjs/core'; +import { type NestExpressApplication } from '@nestjs/platform-express'; +import request from 'supertest'; +import sharp from 'sharp'; +import { PrismaClient } from '@photo/db-cn'; +import { PrismaClient as OvPrismaClient } from '@photo/db-ov'; +import { ErrorCodes } from '@photo/contracts'; +import { + createStorageFromEnv, + watermarkVariantObjectKey, + type MinioStorageAdapter, + type StoragePort, +} from '@photo/storage'; +import { AppCnModule } from '../src/app-cn.module'; +import { AppOvModule } from '../src/app-ov.module'; +import { hashSessionToken } from '../src/auth/session.guard'; +import { configureApp } from '../src/shared/configure-app'; +import { ContractWsAdapter } from '../src/realtime/contract-ws.adapter'; +import { PublishService } from '../../worker/src/publish.service'; +import { + WatermarkRegenService, +} from '../../worker/src/watermark.service'; +import { OutboxSender } from '../../sync/src/sender'; +import { resetDb } from './reset'; +import { resetOvDb } from './reset-ov'; + +jest.setTimeout(180_000); + +const SYNC_KEY = process.env.SYNC_INTERNAL_KEY ?? 'dev-sync-secret-change-me'; + +/** 测试内生成真实 JPEG(带纹理,水印差异可观测;无二进制入库)。 */ +async function makeJpeg( + width: number, + height: number, + tint: number, +): Promise { + const svg = Buffer.from( + `` + + Array.from( + { length: 10 }, + (_, i) => + ``, + ).join('') + + '', + ); + return sharp(svg).jpeg({ quality: 85 }).toBuffer(); +} + +/** 流式读取对象为单缓冲(与 worker readOriginal 同约定)。 */ +async function readObject(storage: StoragePort, key: string): Promise { + const stream = (await storage.getStream(key)) as Readable; + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array)); + } + return Buffer.concat(chunks); +} + +/** S3 协议直删单对象(StoragePort 无删除——YAGNI,仅测试注入用)。 */ +async function deleteObject(key: string): Promise { + const client = 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 client.send(new DeleteObjectCommand({ Bucket: process.env.S3_BUCKET, Key: key })); + } finally { + await client.destroy(); + } +} + +describe('watermark (e2e)', () => { + let cnApp: NestExpressApplication; + let ovApp: NestExpressApplication; + let cnBase: string; + let ovBase: string; + let prismaCn: PrismaClient; + let prismaOv: OvPrismaClient; + let storage: MinioStorageAdapter; + let sender: OutboxSender; + let regen: WatermarkRegenService; + let publish: PublishService; + + let ownerId: string; + let ownerToken: string; + let collabToken: string; + let outsiderToken: string; + let adminToken: string; + + let albumId: string; + let publicToken: string; + /** 上传顺序的三张照片(v1 重生成目标)。 */ + let photoIds: string[]; + let originals: Map; + let photoIdOfFailed: string; + + /** 排空发送器:循环 runOnce 直至无未投递事件(同 public-ov spec)。 */ + async function drainSender(): Promise { + for (let round = 0; round < 20; round++) { + const stats = await sender.runOnce(); + if (stats.picked === 0) { + return; + } + } + throw new Error('drainSender 超时:20 轮后 outbox 仍有未投递事件'); + } + + /** 上传全链路:init → presign PUT 实写 MinIO → complete → 直驱发布事务。 */ + async function uploadAndPublish(clientTaskId: string, jpeg: Buffer): Promise { + const init = await request(cnBase) + .post('/v1/uploads') + .set('Authorization', `Bearer ${ownerToken}`) + .send({ + albumId, + clientTaskId, + filename: 'wm.jpg', + bytes: jpeg.length, + contentType: 'image/jpeg', + }); + expect(init.status).toBe(201); + const { uploadId, photoId, credential } = init.body as { + uploadId: string; + photoId: string; + credential: { url: string; method: string; headers: Record }; + }; + const put = await fetch(credential.url, { + method: credential.method, + headers: credential.headers, + body: new Uint8Array(jpeg), + }); + expect(put.ok).toBe(true); + const done = await request(cnBase) + .post(`/v1/uploads/${uploadId}/complete`) + .set('Authorization', `Bearer ${ownerToken}`); + expect(done.status).toBe(200); + expect(await publish.processPhoto(photoId)).toBe('published'); + originals.set(photoId, jpeg); + return photoId; + } + + const postWatermark = (token: string, body: object) => + request(cnBase) + .post(`/v1/albums/${albumId}/watermark-jobs`) + .set('Authorization', `Bearer ${token}`) + .send(body); + + const getWatermark = (token: string) => + request(cnBase) + .get(`/v1/albums/${albumId}/watermark-jobs`) + .set('Authorization', `Bearer ${token}`); + + /** 取回照片的 browse 变体对象字节(对照基准)。 */ + const browseObjectOf = async (photoId: string): Promise => { + const row = await prismaCn.photo_variants.findUniqueOrThrow({ + where: { photoId_variant: { photoId, variant: 'browse' } }, + }); + return readObject(storage, row.objectKey); + }; + + beforeAll(async () => { + expect(SYNC_KEY).not.toBe(''); + await resetDb(); + await resetOvDb(); + + cnApp = await NestFactory.create(AppCnModule, { + rawBody: true, + }); + configureApp(cnApp); + await cnApp.listen(0); + cnBase = await cnApp.getUrl(); + + ovApp = await NestFactory.create(AppOvModule, { + rawBody: true, + }); + ovApp.useWebSocketAdapter(new ContractWsAdapter(ovApp)); + configureApp(ovApp); + await ovApp.listen(0); + ovBase = await ovApp.getUrl(); + + prismaCn = new PrismaClient(); + prismaOv = new OvPrismaClient(); + storage = createStorageFromEnv(); + sender = new OutboxSender({ prisma: prismaCn, ovApiUrl: ovBase, internalKey: SYNC_KEY }); + regen = new WatermarkRegenService(prismaCn, storage); + publish = new PublishService(prismaCn, storage); + + const mkUser = (account: string, extra: { role?: string } = {}) => + prismaCn.users.create({ + data: { + account, + role: extra.role ?? 'staff', + passwordHash: 'x', + displayName: account, + }, + select: { id: true }, + }); + ownerId = (await mkUser('wm-owner')).id; + const collabId = (await mkUser('wm-collab')).id; + const outsiderId = (await mkUser('wm-outsider')).id; + const adminId = (await mkUser('wm-admin', { role: 'admin' })).id; + + const mkToken = async (userId: string): Promise => { + const token = randomBytes(32).toString('hex'); + await prismaCn.sessions.create({ + data: { + userId, + tokenHash: hashSessionToken(token), + expiresAt: new Date(Date.now() + 3_600_000), + }, + }); + return token; + }; + ownerToken = await mkToken(ownerId); + collabToken = await mkToken(collabId); + outsiderToken = await mkToken(outsiderId); + adminToken = await mkToken(adminId); + + const album = await request(cnBase) + .post('/v1/albums') + .set('Authorization', `Bearer ${ownerToken}`) + .send({ + translations: [ + { locale: 'zh-CN', title: '水印验收相册' }, + { locale: 'en', title: 'Watermark Gala' }, + ], + defaultLocale: 'zh-CN', + startAt: '2026-11-01T10:00:00.000Z', + }); + expect(album.status).toBe(201); + albumId = album.body.id as string; + publicToken = album.body.publicToken as string; + + // 协作者(可见不可管)与三张已发布照片。 + const member = await request(cnBase) + .post(`/v1/albums/${albumId}/members`) + .set('Authorization', `Bearer ${ownerToken}`) + .send({ userId: collabId, role: 'collaborator' }); + expect(member.status).toBe(201); + + originals = new Map(); + photoIds = []; + for (let i = 0; i < 3; i++) { + photoIds.push( + await uploadAndPublish( + `wm-seed-${i}`, + await makeJpeg(900 - i * 60, 600 - i * 40, 0x70 + i * 0x18), + ), + ); + } + await drainSender(); + }); + + afterAll(async () => { + await cnApp.close(); + await ovApp.close(); + await prismaCn.$disconnect(); + await prismaOv.$disconnect(); + }); + + it('① POST(enabled)→ 202 {jobId, configVersion};GET 状态形状(缺省填齐/pending)', async () => { + const res = await postWatermark(ownerToken, { + enabled: true, + text: { content: '© Photo Live' }, + }); + expect(res.status).toBe(202); + expect(res.body).toEqual({ jobId: expect.any(String), configVersion: 1 }); + expect((res.body.jobId as string).length).toBeGreaterThan(0); + + // albums.watermarkVersion(最新创建版本)即刻推进;有效位未动(任务未 done)。 + const album = await prismaCn.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkVersion).toBe(1); + expect(album.watermarkEffectiveVersion).toBe(0); + expect(album.watermarkEnabled).toBe(false); + + // 审计行。 + const audit = await prismaCn.audit_logs.findFirstOrThrow({ + where: { action: 'album.watermark', resourceId: albumId }, + }); + expect(audit.metadata).toMatchObject({ configVersion: 1, enabled: true }); + + const status = await getWatermark(ownerToken); + expect(status.status).toBe(200); + expect(status.body.current).toEqual({ + configVersion: 1, + // 缺省填齐(契约 WatermarkConfig 可选字段的服务端默认值)。 + config: { + enabled: true, + position: 'bottomRight', + opacity: 50, + scalePercent: 20, + text: { content: '© Photo Live', fontSize: 24 }, + }, + }); + expect(status.body.jobs).toHaveLength(1); + expect(status.body.jobs[0]).toMatchObject({ + configVersion: 1, + state: 'pending', + progress: 0, + total: 3, + failed: 0, + }); + expect(status.body.jobs[0].jobId).toBe(res.body.jobId); + expect(typeof status.body.jobs[0].createdAt).toBe('string'); + expect(status.body.jobs[0].errorCode).toBeUndefined(); + }); + + it('② 新照片发布即套用最新版:POST 后发布 → watermark 变体对象已产且 ≠ browse', async () => { + const photoId = await uploadAndPublish('wm-after-post', await makeJpeg(760, 500, 0x99)); + photoIds.push(photoId); + + // 发布路径写入 watermark 行(version=1=最新创建版本)+ 对象存在且非空 jpeg, + // 与 browse 变体字节不同(水印本身即差异——A18「新照片即刻套用」)。 + const row = await prismaCn.photo_variants.findUnique({ + where: { photoId_variant: { photoId, variant: 'watermark' } }, + }); + expect(row?.version).toBe(1); + expect(row?.objectKey).toBe(watermarkVariantObjectKey(albumId, photoId, 1)); + const wmObject = await readObject(storage, row!.objectKey); + expect(wmObject.length).toBeGreaterThan(0); + expect((await sharp(wmObject).metadata()).format).toBe('jpeg'); + expect(wmObject.equals(await browseObjectOf(photoId))).toBe(false); + }); + + it('③ 历史重生成 → done;cn 公开列表服务水印字节;④ A20 原图零触碰;⑤ ov 模式键', async () => { + const pending = await getWatermark(ownerToken); + const jobId = pending.body.jobs[0].jobId as string; + expect(await regen.processJob(jobId)).toBe('done'); + + const status = await getWatermark(ownerToken); + expect(status.body.jobs[0]).toMatchObject({ + state: 'done', + // total = POST 时点的已发布照片快照(3);progress 含 ② 追加的 1 张 + // (发布中途的新照片自然纳入处理——progress 计「已处理」,可超过 total)。 + progress: 4, + total: 3, + failed: 0, + }); + + // 有效位推进(仅 done 后——ov 切换的前提)。 + const album = await prismaCn.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkEffectiveVersion).toBe(1); + expect(album.watermarkEnabled).toBe(true); + + // cn 公开列表:每条 browse 现签 URL 取回的是**水印字节**(≠ browse 变体)。 + const list = await request(cnBase) + .get(`/v1/public/albums/${publicToken}/photos?limit=60`); + expect(list.status).toBe(200); + expect(list.body.items.map((i: { photoId: string }) => i.photoId).sort()).toEqual( + [...photoIds].sort(), + ); + for (const item of list.body.items as Array<{ + photoId: string; + browse: { url: string }; + }>) { + const fetched = Buffer.from(await (await fetch(item.browse.url)).arrayBuffer()); + expect(fetched.length).toBeGreaterThan(0); + expect(fetched.equals(await browseObjectOf(item.photoId))).toBe(false); + } + + // A20:原图对象键与字节零触碰。 + for (const [photoId, original] of originals) { + const photo = await prismaCn.photos.findUniqueOrThrow({ where: { id: photoId } }); + expect(photo.objectKey).toBe( + `photo-live/dev/albums/${albumId}/originals/${photoId}.jpg`, + ); + expect(await readObject(storage, photo.objectKey)).toEqual(original); + } + + // ov:副本位 + 模式键。发送器排空后 album.updated(有效位推进)到达 ov。 + await drainSender(); + const ovAlbum = await prismaOv.albums_ov.findUniqueOrThrow({ + where: { albumId }, + }); + expect(ovAlbum.watermarkEnabled).toBe(true); + expect(ovAlbum.watermarkConfigVersion).toBe(1); + + const ovList = await request(ovBase) + .get(`/v1/public/albums/${publicToken}/photos?limit=60`); + expect(ovList.status).toBe(200); + expect(ovList.body.items).toHaveLength(photoIds.length); + for (const item of ovList.body.items as Array<{ + photoId: string; + browse: { url: string }; + }>) { + // 模式键(确定性问题:URL 路径含版本段 watermark 对象键)。 + expect(item.browse.url).toContain( + `/albums/${albumId}/variants/${item.photoId}/1/watermark.jpg`, + ); + // 取回字节与 cn 水印对象一致(同一对象、两域各自现签)。 + const cnRow = await prismaCn.photo_variants.findUniqueOrThrow({ + where: { photoId_variant: { photoId: item.photoId, variant: 'watermark' } }, + }); + const fetched = Buffer.from(await (await fetch(item.browse.url)).arrayBuffer()); + expect(fetched.equals(await readObject(storage, cnRow.objectKey))).toBe(true); + } + }); + + it('⑥ 失败注入:删 1 张原图 → 新版本 done 且 failed=1;失败张两侧保留旧图;恢复后 re-POST 收敛', async () => { + photoIdOfFailed = photoIds[1]!; + const failedPhoto = await prismaCn.photos.findUniqueOrThrow({ + where: { id: photoIdOfFailed }, + }); + const failedOriginal = originals.get(photoIdOfFailed)!; + await deleteObject(failedPhoto.objectKey); + + // v1 水印对象字节(失败张的「旧图」基准)。 + const v1Key = watermarkVariantObjectKey(albumId, photoIdOfFailed, 1); + const v1Object = await readObject(storage, v1Key); + + const res = await postWatermark(ownerToken, { + enabled: true, + text: { content: 'v2 水印' }, + position: 'topRight', + }); + expect(res.status).toBe(202); + expect(res.body.configVersion).toBe(2); + expect(await regen.processJob(res.body.jobId)).toBe('done'); + + const status = await getWatermark(ownerToken); + expect(status.body.jobs[0]).toMatchObject({ + configVersion: 2, + state: 'done', + progress: 4, + total: 4, + failed: 1, + }); + + // 成功张:行推进到 v2。 + for (const photoId of photoIds.filter((id) => id !== photoIdOfFailed)) { + const row = await prismaCn.photo_variants.findUniqueOrThrow({ + where: { photoId_variant: { photoId, variant: 'watermark' } }, + }); + expect(row.version).toBe(2); + } + // 失败张:cn 行保留 v1(未覆盖——保留旧图),ov 模式键 v2 = v1 旧图拷贝。 + const failedRow = await prismaCn.photo_variants.findUniqueOrThrow({ + where: { photoId_variant: { photoId: photoIdOfFailed, variant: 'watermark' } }, + }); + expect(failedRow.version).toBe(1); + expect( + await readObject(storage, watermarkVariantObjectKey(albumId, photoIdOfFailed, 2)), + ).toEqual(v1Object); + + await drainSender(); + // cn:失败张 browse URL 取回 v1 旧图字节;成功张取回 v2。 + const cnList = await request(cnBase) + .get(`/v1/public/albums/${publicToken}/photos?limit=60`); + const byPhoto = new Map( + (cnList.body.items as Array<{ photoId: string; browse: { url: string } }>).map( + (i) => [i.photoId, i.browse.url], + ), + ); + expect( + Buffer.from(await (await fetch(byPhoto.get(photoIdOfFailed)!)).arrayBuffer()), + ).toEqual(v1Object); + const okPhoto = photoIds[0]!; + const okV2 = await readObject( + storage, + watermarkVariantObjectKey(albumId, okPhoto, 2), + ); + expect( + Buffer.from(await (await fetch(byPhoto.get(okPhoto)!)).arrayBuffer()), + ).toEqual(okV2); + + // ov:失败张模式键 v2 同样是旧图字节(与 cn 一致——A20 双域同语义)。 + const ovList = await request(ovBase) + .get(`/v1/public/albums/${publicToken}/photos?limit=60`); + const ovByPhoto = new Map( + (ovList.body.items as Array<{ photoId: string; browse: { url: string } }>).map( + (i) => [i.photoId, i.browse.url], + ), + ); + expect(ovByPhoto.get(photoIdOfFailed)).toContain( + `/variants/${photoIdOfFailed}/2/watermark.jpg`, + ); + expect( + Buffer.from(await (await fetch(ovByPhoto.get(photoIdOfFailed)!)).arrayBuffer()), + ).toEqual(v1Object); + + // —— 恢复原图 → re-POST v3 → failed=0 全量收敛(重新发起语义)—— + await storage.put(failedPhoto.objectKey, failedOriginal, 'image/jpeg'); + const retry = await postWatermark(ownerToken, { + enabled: true, + text: { content: 'v3 retry' }, + }); + expect(retry.status).toBe(202); + expect(retry.body.configVersion).toBe(3); + expect(await regen.processJob(retry.body.jobId)).toBe('done'); + + const after = await getWatermark(ownerToken); + expect(after.body.jobs[0]).toMatchObject({ + configVersion: 3, + state: 'done', + failed: 0, + progress: 4, + }); + const recovered = await prismaCn.photo_variants.findUniqueOrThrow({ + where: { photoId_variant: { photoId: photoIdOfFailed, variant: 'watermark' } }, + }); + expect(recovered.version).toBe(3); + expect( + (await readObject(storage, watermarkVariantObjectKey(albumId, photoIdOfFailed, 3))).equals( + v1Object, + ), + ).toBe(false); + }); + + it('⑦ 关闭版重生成:两侧回落 browse 键;水印行删除、对象保留', async () => { + const res = await postWatermark(ownerToken, { enabled: false }); + expect(res.status).toBe(202); + expect(res.body.configVersion).toBe(4); + expect(await regen.processJob(res.body.jobId)).toBe('done'); + + const album = await prismaCn.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkEnabled).toBe(false); + expect(album.watermarkEffectiveVersion).toBe(4); + + // 行全删(对象保留——历史可溯)。 + expect( + await prismaCn.photo_variants.count({ where: { photoId: { in: photoIds }, variant: 'watermark' } }), + ).toBe(0); + for (const photoId of photoIds) { + expect( + await storage.head(watermarkVariantObjectKey(albumId, photoId, 3)), + ).not.toBeNull(); + } + + await drainSender(); + const ovAlbum = await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId } }); + expect(ovAlbum.watermarkEnabled).toBe(false); + expect(ovAlbum.watermarkConfigVersion).toBe(4); + + // cn/ov 公开列表:browse URL 全部回落 browse 变体键,取回字节=browse 对象。 + for (const base of [cnBase, ovBase]) { + const list = await request(base) + .get(`/v1/public/albums/${publicToken}/photos?limit=60`); + expect(list.status).toBe(200); + for (const item of list.body.items as Array<{ + photoId: string; + browse: { url: string }; + }>) { + expect(item.browse.url).toContain(`/variants/${item.photoId}/1/browse.jpg`); + const fetched = Buffer.from(await (await fetch(item.browse.url)).arrayBuffer()); + expect(fetched.equals(await browseObjectOf(item.photoId))).toBe(true); + } + } + }); + + it('⑧ 空相册:POST 即建即 done,有效位当场推进并同步 ov', async () => { + const emptyAlbum = await request(cnBase) + .post('/v1/albums') + .set('Authorization', `Bearer ${ownerToken}`) + .send({ + translations: [{ locale: 'zh-CN', title: '空相册水印' }], + defaultLocale: 'zh-CN', + startAt: '2026-11-20T10:00:00.000Z', + }); + expect(emptyAlbum.status).toBe(201); + const emptyId = emptyAlbum.body.id as string; + + const res = await request(cnBase) + .post(`/v1/albums/${emptyId}/watermark-jobs`) + .set('Authorization', `Bearer ${ownerToken}`) + .send({ enabled: true, text: { content: '空相册' } }); + expect(res.status).toBe(202); + + const status = await request(cnBase) + .get(`/v1/albums/${emptyId}/watermark-jobs`) + .set('Authorization', `Bearer ${ownerToken}`); + expect(status.body.jobs[0]).toMatchObject({ state: 'done', total: 0, progress: 0 }); + const album = await prismaCn.albums.findUniqueOrThrow({ where: { id: emptyId } }); + expect(album.watermarkEnabled).toBe(true); + expect(album.watermarkEffectiveVersion).toBe(1); + + await drainSender(); + const ovAlbum = await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId: emptyId } }); + expect(ovAlbum.watermarkEnabled).toBe(true); + expect(ovAlbum.watermarkConfigVersion).toBe(1); + }); + + it('⑨ 权限:协作者 403 / 圈外人 404 / admin 可管;未认证 401', async () => { + const collabPost = await postWatermark(collabToken, { + enabled: true, + text: { content: 'x' }, + }); + expect(collabPost.status).toBe(403); + expect(collabPost.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN); + + const collabGet = await getWatermark(collabToken); + expect(collabGet.status).toBe(403); + expect(collabGet.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN); + + const outsider = await request(cnBase) + .post(`/v1/albums/${albumId}/watermark-jobs`) + .set('Authorization', `Bearer ${outsiderToken}`) + .send({ enabled: true, text: { content: 'x' } }); + expect(outsider.status).toBe(404); + expect(outsider.body.code).toBe(ErrorCodes.ALBUM_NOT_FOUND); + + const anon = await request(cnBase) + .get(`/v1/albums/${albumId}/watermark-jobs`); + expect(anon.status).toBe(401); + + const adminStatus = await getWatermark(adminToken); + expect(adminStatus.status).toBe(200); + const adminPost = await postWatermark(adminToken, { + enabled: true, + text: { content: 'admin v5' }, + }); + expect(adminPost.status).toBe(202); + expect(adminPost.body.configVersion).toBe(5); + }); + + it('⑩ 校验:enabled=true 缺 text.content / 越界值 / 非法 position → 400', async () => { + const cases: Array> = [ + { enabled: true }, // 缺 text.content + { enabled: true, text: { content: '' } }, // 空串 + { enabled: true, text: { content: 'x' }, opacity: 101 }, + { enabled: true, text: { content: 'x' }, opacity: -1 }, + { enabled: true, text: { content: 'x' }, scalePercent: 4 }, + { enabled: true, text: { content: 'x' }, scalePercent: 51 }, + { enabled: true, text: { content: 'x' }, position: 'middle' }, + { enabled: true, text: { content: 'x', fontSize: 7 } }, + { enabled: true, text: { content: 'x', fontSize: 73 } }, + { enabled: 'yes', text: { content: 'x' } }, + ]; + for (const body of cases) { + const res = await postWatermark(ownerToken, body); + expect(res.status).toBe(400); + expect(res.body.code).toBe(ErrorCodes.VALIDATION_FAILED); + } + }); + + it('GET jobs 全史:创建时间倒序(configVersion 降序)且条目字段完整', async () => { + const status = await getWatermark(ownerToken); + expect(status.status).toBe(200); + const jobs = status.body.jobs as Array<{ + jobId: string; + configVersion: number; + state: string; + progress: number; + total: number; + failed: number; + createdAt: string; + }>; + expect(jobs.length).toBeGreaterThanOrEqual(5); + const versions = jobs.map((j) => j.configVersion); + expect([...versions].sort((a, b) => b - a)).toEqual(versions); + for (const job of jobs) { + expect(Object.keys(job).sort()).toEqual( + ['configVersion', 'createdAt', 'failed', 'jobId', 'progress', 'state', 'total'].sort(), + ); + } + }); +}); diff --git a/services/worker/src/main.ts b/services/worker/src/main.ts index cb2486a..24a437c 100644 --- a/services/worker/src/main.ts +++ b/services/worker/src/main.ts @@ -1,17 +1,20 @@ -// services/worker/src/main.ts —— photo worker 进程入口(Task 21)。构建产物 -// dist/main.js 由根 dev:worker / 生产启动命令运行;env 一律来自进程环境(根 -// dotenv-cli 注入),此处不读文件。启动顺序:Prisma + MinIO 适配器 → 两条专用 -// ioredis 连接(Worker 阻塞命令须独占连接;Queue/sweep 入队共用另一条)→ -// BullMQ Worker(并发 2)+ 30s 补偿扫描。优雅退出:SIGINT/SIGTERM 先停扫描, -// worker.close()(等待在飞 job)后逐个关闭连接(Task 19 gotcha:BullMQ 不接管 -// 外部连接,须显式 quit;未监听的 error 事件会崩进程,全部挂日志监听)。 +// services/worker/src/main.ts —— photo worker 进程入口(Task 21;P4a Task 3 +// 增补水印重生成消费者)。构建产物 dist/main.js 由根 dev:worker / 生产启动 +// 命令运行;env 一律来自进程环境(根 dotenv-cli 注入),此处不读文件。 +// 启动顺序:Prisma + MinIO 适配器 → 三条专用 ioredis 连接(Worker 阻塞命令 +// 须独占连接——photo 与 watermark 各一条;Queue/sweep 入队共用另一条)→ +// BullMQ Worker ×2(photo 并发 2 / watermark 并发 1)+ 30s 补偿扫描。 +// 优雅退出:SIGINT/SIGTERM 先停扫描,worker.close()(等待在飞 job)后逐个 +// 关闭连接(Task 19 gotcha:BullMQ 不接管外部连接,须显式 quit;未监听的 +// error 事件会崩进程,全部挂日志监听)。 import { Queue } from 'bullmq'; import IORedis from 'ioredis'; import { pino, type Logger } from 'pino'; import { PrismaClient } from '@photo/db-cn'; -import { QUEUE_PHOTO_PROCESS } from '@photo/contracts'; +import { QUEUE_PHOTO_PROCESS, QUEUE_WATERMARK } from '@photo/contracts'; import { createStorageFromEnv } from '@photo/storage'; import { createPhotoProcessWorker } from './process.processor'; +import { createWatermarkRegenWorker } from './watermark.processor'; import { startSweepInterval } from './sweep'; const SWEEP_INTERVAL_MS = 30_000; @@ -31,6 +34,7 @@ async function main(): Promise { const prisma = new PrismaClient(); const storage = createStorageFromEnv(); const workerConnection = createConnection(logger, 'worker'); + const watermarkConnection = createConnection(logger, 'watermark-worker'); const queueConnection = createConnection(logger, 'queue'); const sweepQueue = new Queue(QUEUE_PHOTO_PROCESS, { connection: queueConnection }); const worker = createPhotoProcessWorker({ @@ -39,9 +43,20 @@ async function main(): Promise { connection: workerConnection, logger, }); + const watermarkWorker = createWatermarkRegenWorker({ + prisma, + storage, + connection: watermarkConnection, + logger, + }); const sweep = startSweepInterval(SWEEP_INTERVAL_MS, { prisma, queue: sweepQueue, logger }); logger.info( - { queue: QUEUE_PHOTO_PROCESS, concurrency: 2, sweepIntervalMs: SWEEP_INTERVAL_MS }, + { + queue: QUEUE_PHOTO_PROCESS, + watermarkQueue: QUEUE_WATERMARK, + concurrency: 2, + sweepIntervalMs: SWEEP_INTERVAL_MS, + }, 'photo worker started', ); @@ -53,8 +68,10 @@ async function main(): Promise { sweep.stop(); try { await worker.close(); // 等待在飞 job 完成 + await watermarkWorker.close(); await sweepQueue.close(); await workerConnection.quit().catch(() => undefined); + await watermarkConnection.quit().catch(() => undefined); await queueConnection.quit().catch(() => undefined); await prisma.$disconnect(); } catch (err) { diff --git a/services/worker/src/publish.service.ts b/services/worker/src/publish.service.ts index 14f246d..1c46f48 100644 --- a/services/worker/src/publish.service.ts +++ b/services/worker/src/publish.service.ts @@ -13,12 +13,18 @@ import { Readable } from 'node:stream'; import type { Prisma, PrismaClient } from '@photo/db-cn'; import type { StoragePort } from '@photo/storage'; -import type { PhotoPublishedPayload } from '@photo/contracts'; +import { + variantsObjectKey, + watermarkVariantObjectKey, +} from '@photo/storage'; +import { + parseStoredWatermarkConfig, + type PhotoPublishedPayload, + type StoredWatermarkConfig, +} from '@photo/contracts'; import type { Logger } from 'pino'; import { runPipeline } from './pipeline'; - -/** 对象键根(与 api src/uploads/support.ts 的 OBJECT_KEY_ROOT 一致:P1a 单环境)。 */ -const OBJECT_KEY_ROOT = 'photo-live/dev'; +import { composeWatermarkBrowse, type WatermarkComposeResult } from './watermark'; /** 变体键版本段(对应 photo_variants.version;水印/规格升级时递增)。 */ const VARIANT_VERSION = 1; @@ -31,13 +37,11 @@ export type ProcessOutcome = | 'already' // 重复投递:已 ready(或并发赢家已处理),幂等 no-op | 'skipped'; // 照片不存在 / 状态不可处理(pending_upload、processing 等) -export function variantsObjectKey( - albumId: string, - photoId: string, - version: number, - variant: 'thumb' | 'browse', -): string { - return `${OBJECT_KEY_ROOT}/albums/${albumId}/variants/${photoId}/${version}/${variant}.jpg`; +/** 发布路径的水印产出现状(P4a Task 3,见 resolveWatermarkTargets 注释)。 */ +interface PublishWatermarkTarget { + version: number; + config: StoredWatermarkConfig; + out: WatermarkComposeResult; } export class PublishService { @@ -86,6 +90,15 @@ export class PublishService { await this.storage.put(thumbKey, pipeline.thumb, 'image/jpeg'); await this.storage.put(browseKey, pipeline.browse, 'image/jpeg'); + // P4a Task 3:发布路径水印变体(「新照片默认套用最新版本」的落点)。 + // 对象先于事务写(键含版本段、内容确定,重复写无害),行 upsert 在发布 + // 事务内与 thumb/browse 同生共死。 + const watermarkTargets = await this.produceWatermarkVariants( + photo.albumId, + photoId, + input, + ); + const publishedAt = new Date(); const outcome = await this.prisma.$transaction(async (tx) => { // 条件更新 = 处理权抢占:WHERE 仍处可处理态。匹配 0 → 已被处理,幂等返回。 @@ -146,6 +159,29 @@ export class PublishService { height: pipeline.browseHeight, }, }); + // 水印变体行(targets 已按版本升序,末次 upsert 后行版本 = 最大写入版本)。 + for (const target of watermarkTargets) { + const key = watermarkVariantObjectKey(photo.albumId, photoId, target.version); + await tx.photo_variants.upsert({ + where: { photoId_variant: { photoId, variant: 'watermark' } }, + update: { + objectKey: key, + bytes: target.out.buffer.length, + version: target.version, + width: target.out.width, + height: target.out.height, + }, + create: { + photoId, + variant: 'watermark', + objectKey: key, + bytes: target.out.buffer.length, + version: target.version, + width: target.out.width, + height: target.out.height, + }, + }); + } const payload: PhotoPublishedPayload = { photoId, albumId: photo.albumId, @@ -184,4 +220,67 @@ export class PublishService { } return Buffer.concat(chunks); } + + /** + * 发布路径的水印变体产出现状(P4a Task 3)。目标版本集合的推导(见 + * contracts watermark.ts 版本语义): + * · 最新配置(albums.watermarkVersion)开启 → 写最新版键; + * · 相册「有效版本」(watermarkEffectiveVersion,ov 侧仍在服务的旧版)与 + * 最新版不同且开启 → **双写**:重生成窗口内发布的新照片同时落有效版键, + * 杜绝 ov 模式键指向尚未生成的对象(破图窗口);cn 侧行版本取最大值; + * · 最新配置关闭但有效配置仍开启(关闭版重生成进行中)→ 写有效版键: + * cn 公开侧跟有效位(watermarkEnabled 仍 true)继续服务旧水印行,直至 + * 关闭任务逐张删行渐进回落 browse;ov 侧模式键在切换前继续有对象可服务 + * (窗口内两地域一致展示旧图——A20 原子切换语义)。 + * 两配置一致时仅合成一次(同 config 结果缓存)。 + */ + private async produceWatermarkVariants( + albumId: string, + photoId: string, + input: Buffer, + ): Promise { + const album = await this.prisma.albums.findUnique({ + where: { id: albumId }, + select: { + watermarkVersion: true, + watermarkEnabled: true, + watermarkEffectiveVersion: true, + }, + }); + if (album === null) { + return []; + } + const versions = new Set(); + if (album.watermarkVersion > 0) { + versions.add(album.watermarkVersion); + } + if (album.watermarkEnabled && album.watermarkEffectiveVersion > 0) { + versions.add(album.watermarkEffectiveVersion); + } + if (versions.size === 0) { + return []; + } + const configRows = await this.prisma.watermark_configs.findMany({ + where: { albumId, configVersion: { in: [...versions] } }, + }); + const byVersion = new Map(); + for (const row of configRows) { + const parsed = parseStoredWatermarkConfig(row.config); + if (parsed !== null) { + byVersion.set(row.configVersion, parsed); + } + } + const targets: PublishWatermarkTarget[] = []; + for (const version of [...versions].sort((a, b) => a - b)) { + const config = byVersion.get(version); + if (config === undefined || !config.enabled) { + continue; + } + const out = await composeWatermarkBrowse(input, config); + const key = watermarkVariantObjectKey(albumId, photoId, version); + await this.storage.put(key, out.buffer, 'image/jpeg'); + targets.push({ version, config, out }); + } + return targets; + } } diff --git a/services/worker/src/watermark.processor.ts b/services/worker/src/watermark.processor.ts new file mode 100644 index 0000000..534d11a --- /dev/null +++ b/services/worker/src/watermark.processor.ts @@ -0,0 +1,80 @@ +// services/worker/src/watermark.processor.ts —— QUEUE_WATERMARK 消费者(P4a +// Task 3)。与 photo-process 消费者同构的 Worker 包装: +// · 照片级失败在 WatermarkRegenService 内部消化(failed 计数 + 继续),不抛错; +// · 任务级错误(完成事务乐观重试耗尽、DB 异常等)抛给 BullMQ 重试(3 次指数 +// 退避 1s 起);终败判定同 process.processor(BullMQ 在 moveToFailed 之后才 +// 自增 attemptsMade,当前尝试 = attemptsMade + 1),耗尽置 job 行 failed + +// errorCode 兜底;服务级中止(配置缺失等)由服务自行落 failed,不进重试。 +// jobId 即 watermark_jobs.id(cuid,无 ':'——BullMQ 保留分隔符约束天然满足)。 +import { Worker, type Job } from 'bullmq'; +import type IORedis from 'ioredis'; +import type { PrismaClient } from '@photo/db-cn'; +import type { StoragePort } from '@photo/storage'; +import type { Logger } from 'pino'; +import { QUEUE_WATERMARK } from '@photo/contracts'; +import { WatermarkRegenService, type WatermarkJobOutcome } from './watermark.service'; + +export interface WatermarkRegenJobData { + jobId: string; +} + +/** job 名称(api 生产者同值)。 */ +export const WATERMARK_REGEN_JOB_NAME = 'regen'; + +/** 与 api 生产者一致的重试策略。 */ +export const WATERMARK_JOB_ATTEMPTS = 3; +export const WATERMARK_JOB_BACKOFF = { type: 'exponential', delay: 1000 } as const; + +/** 终败 errorCode 兜底(非服务级中止码:网络/DB 等意外错误)。 */ +export const WATERMARK_FALLBACK_ERROR_CODE = 'WATERMARK_REGEN_FAILED'; + +export interface WatermarkRegenWorkerOptions { + prisma: PrismaClient; + storage: StoragePort; + /** Worker 专用连接(阻塞命令),调用方负责 quit。 */ + connection: IORedis; + concurrency?: number; + logger?: Logger; +} + +export function createWatermarkRegenWorker( + options: WatermarkRegenWorkerOptions, +): Worker { + const { prisma } = options; + const service = new WatermarkRegenService(prisma, options.storage, options.logger); + + const processor = async ( + job: Job, + ): Promise => { + const jobId = job.data.jobId; + try { + const outcome = await service.processJob(jobId); + options.logger?.info({ jobId, outcome }, 'watermark job 完成'); + return outcome; + } catch (err) { + const isFinalAttempt = job.attemptsMade + 1 >= (job.opts.attempts ?? 1); + if (isFinalAttempt) { + options.logger?.error( + { jobId, err: err instanceof Error ? err.message : String(err) }, + 'watermark job 终败(重试耗尽)', + ); + await prisma.watermark_jobs + .updateMany({ + where: { id: jobId, state: { in: ['pending', 'running', 'failed'] } }, + data: { state: 'failed', errorCode: WATERMARK_FALLBACK_ERROR_CODE }, + }) + .catch(() => undefined); // 落库失败仅丢审计位,不掩盖原始错误 + } + throw err; // 交还 BullMQ:未耗尽则按退避重试,耗尽则入 failed set + } + }; + + return new Worker( + QUEUE_WATERMARK, + processor, + { + connection: options.connection, + concurrency: options.concurrency ?? 1, // 相册级任务串行起步(多相册仍并行) + }, + ); +} diff --git a/services/worker/src/watermark.service.ts b/services/worker/src/watermark.service.ts new file mode 100644 index 0000000..bcf08cb --- /dev/null +++ b/services/worker/src/watermark.service.ts @@ -0,0 +1,314 @@ +// services/worker/src/watermark.service.ts —— 水印历史重生成服务(P4a Task 3, +// 「显式任务」语义的实现体:水印变更默认仅影响新照片,历史照片经本任务重derive)。 +// 逐张处理(分批 50):原图 → composeWatermarkBrowse(浏览图基线 + 文字层)→ +// 写版本化水印键 `variants/{photoId}/{configVersion}/watermark.jpg` → upsert +// photo_variants(variant='watermark', version=configVersion)。原子性口径(A20): +// · 对象先写、行后 upsert——cn 公开侧的选图规则是「有效开关开启且行存在 ⇒ +// 用行的键」(行只会指向已写成功的对象),处理中的照片始终有 browse 变体 +// 兜底,绝无破图; +// · 原图(originals)与 browse 变体零触碰(e2e A20 断言); +// · 照片级失败(原图缺失/解码失败)计入 failed 并继续——部分成功允许, +// job 仍 done(failed>0 供前端展示;重试 = 重新 POST 新任务,按照片粒度 +// 幂等:同 configVersion+photo 重复 upsert 天然收敛)。失败张**旧图保留**: +// 行不动(cn 继续服务旧键)+ 旧图对象拷贝到目标版本键(ov 模式键在有效 +// 版本推进后仍有对象可服务——失败照片保留旧图,A20;见 fallbackKeepOldImage); +// · 任务级中止(配置缺失/不可解析)→ state failed + errorCode,照片零处理 +// (有效版本不推进,ov/cn 继续旧配置,杜绝半切换状态)。 +// 关闭版(enabled=false)重生成 = 逐张删除 watermark 变体行(对象保留供历史 +// 版本回溯,成本可忽略);完成后有效位 enabled=false,两侧公开服务回落 browse。 +// 完成事务:job 置 done + 推进 albums「有效版本位」(watermarkEffectiveVersion +// /watermarkEnabled,仅更高版本推进——乱序完成的旧任务自然让位)+ entityVersion+1 +// + outbox album.updated(载荷经 contracts 共享构建器,ov 副本据此切换模式键)。 +import { Readable } from 'node:stream'; +import type { Prisma, PrismaClient } from '@photo/db-cn'; +import { + buildAlbumUpdatedPayloadCore, + parseStoredWatermarkConfig, + type StoredWatermarkConfig, +} from '@photo/contracts'; +import { watermarkVariantObjectKey, type StoragePort } from '@photo/storage'; +import type { Logger } from 'pino'; +import { composeWatermarkBrowse } from './watermark'; + +/** 照片分批大小(逐批查询,游标推进;发布中途的新照片自然纳入后续批次)。 */ +const PHOTO_BATCH_SIZE = 50; + +/** 任务级错误码(照片级失败不使用——只计数)。 */ +export const WATERMARK_ERROR_CONFIG_MISSING = 'WATERMARK_CONFIG_MISSING'; +export const WATERMARK_ERROR_CONFIG_INVALID = 'WATERMARK_CONFIG_INVALID'; +export const WATERMARK_ERROR_ALBUM_MISSING = 'WATERMARK_ALBUM_MISSING'; + +export type WatermarkJobOutcome = + | 'done' // 全部处理完成(允许 failed>0 部分成功) + | 'failed' // 任务级中止(errorCode 已落行) + | 'skipped'; // 行不存在 / 已 done 的重复投递 + +export class WatermarkRegenService { + constructor( + private readonly prisma: PrismaClient, + private readonly storage: StoragePort, + private readonly logger?: Logger, + ) {} + + /** 处理一个重生成任务(BullMQ 投递与 e2e 直驱共用入口)。 */ + async processJob(jobRowId: string): Promise { + const job = await this.prisma.watermark_jobs.findUnique({ + where: { id: jobRowId }, + }); + if (job === null || job.state === 'done') { + return 'skipped'; + } + // 认领(含 BullMQ 整任务重试对 failed 行的复活)+ 计数清零重跑(按照片 + // 幂等,重复处理无副作用)。 + const claimed = await this.prisma.watermark_jobs.updateMany({ + where: { id: jobRowId, state: { in: ['pending', 'running', 'failed'] } }, + data: { state: 'running', progress: 0, failed: 0, errorCode: null }, + }); + if (claimed.count === 0) { + return 'skipped'; + } + + const configRow = await this.prisma.watermark_configs.findUnique({ + where: { + albumId_configVersion: { + albumId: job.albumId, + configVersion: job.configVersion, + }, + }, + }); + const config = parseStoredWatermarkConfig(configRow?.config); + if (configRow === null) { + return this.abortJob(jobRowId, WATERMARK_ERROR_CONFIG_MISSING); + } + if (config === null) { + return this.abortJob(jobRowId, WATERMARK_ERROR_CONFIG_INVALID); + } + const albumExists = await this.prisma.albums.findUnique({ + where: { id: job.albumId }, + select: { id: true }, + }); + if (albumExists === null) { + return this.abortJob(jobRowId, WATERMARK_ERROR_ALBUM_MISSING); + } + + let processed = 0; + let failed = 0; + let cursor: string | undefined; + for (;;) { + const photos = await this.prisma.photos.findMany({ + where: { + albumId: job.albumId, + visibility: 'published', + ...(cursor === undefined ? {} : { id: { gt: cursor } }), + }, + orderBy: { id: 'asc' }, + take: PHOTO_BATCH_SIZE, + select: { id: true, objectKey: true }, + }); + if (photos.length === 0) { + break; + } + for (const photo of photos) { + try { + await this.processPhoto(photo.id, photo.objectKey, job.albumId, job.configVersion, config); + } catch (err) { + failed += 1; + this.logger?.warn( + { jobId: jobRowId, photoId: photo.id, err: err instanceof Error ? err.message : String(err) }, + '水印重生成单张失败(计入 failed,继续)', + ); + if (config.enabled) { + // A20「失败照片保留旧图」的 ov 侧保障:旧图对象拷贝到目标版本键, + // 使 done 后 ov 模式键(有效版本)仍有对象可服务。best-effort—— + // 拷贝再失败(存储整体故障)只影响 ov 该张,cn 旧行照常服务。 + try { + await this.fallbackKeepOldImage(photo.id, job.albumId, job.configVersion); + } catch (fallbackErr) { + this.logger?.warn( + { + jobId: jobRowId, + photoId: photo.id, + err: fallbackErr instanceof Error ? fallbackErr.message : String(fallbackErr), + }, + '水印重生成失败张的旧图拷贝亦失败(ov 该张模式键将缺对象)', + ); + } + } + } + processed += 1; + cursor = photo.id; + await this.prisma.watermark_jobs.update({ + where: { id: jobRowId }, + data: { progress: processed, failed }, + }); + } + } + + await this.completeJob(jobRowId, job.albumId, job.configVersion, config); + this.logger?.info( + { jobId: jobRowId, albumId: job.albumId, configVersion: job.configVersion, total: processed, failed }, + '水印重生成任务完成', + ); + return 'done'; + } + + /** + * 单张:enabled → 产水印变体(对象先写、行后 upsert);disabled → 删行 + * (对象保留)。抛错由调用方计数继续。 + */ + private async processPhoto( + photoId: string, + objectKey: string, + albumId: string, + configVersion: number, + config: StoredWatermarkConfig, + ): Promise { + if (!config.enabled) { + await this.prisma.photo_variants.deleteMany({ + where: { photoId, variant: 'watermark' }, + }); + return; + } + const input = await this.readOriginal(objectKey); + const out = await composeWatermarkBrowse(input, config); + const key = watermarkVariantObjectKey(albumId, photoId, configVersion); + await this.storage.put(key, out.buffer, 'image/jpeg'); + await this.prisma.photo_variants.upsert({ + where: { photoId_variant: { photoId, variant: 'watermark' } }, + update: { + objectKey: key, + bytes: out.buffer.length, + version: configVersion, + width: out.width, + height: out.height, + }, + create: { + photoId, + variant: 'watermark', + objectKey: key, + bytes: out.buffer.length, + version: configVersion, + width: out.width, + height: out.height, + }, + }); + } + + /** + * 失败张旧图保留(A20,对象级兜底——行不动):把照片当前实际展示的旧图 + * (旧水印行对象;无水印行则 browse 对象)拷贝到目标版本水印键。done 后 + * ov 模式键指向有效版本,此拷贝保证该键有对象(旧图字节);cn 侧不受影响 + * (旧行未覆盖,继续服务旧键)。源=目标(同版本重跑)时跳过。 + */ + private async fallbackKeepOldImage( + photoId: string, + albumId: string, + configVersion: number, + ): Promise { + const variants = await this.prisma.photo_variants.findMany({ + where: { photoId, variant: { in: ['watermark', 'browse'] } }, + }); + const source = + variants.find((v) => v.variant === 'watermark') ?? + variants.find((v) => v.variant === 'browse'); + if (source === undefined) { + return; // 连 browse 行都缺(post-publish 不变量破坏):无可拷贝,放弃。 + } + const targetKey = watermarkVariantObjectKey(albumId, photoId, configVersion); + if (source.objectKey === targetKey) { + return; + } + const bytes = await this.readOriginal(source.objectKey); + await this.storage.put(targetKey, bytes, 'image/jpeg'); + } + + /** 任务级中止:照片零处理,行置 failed + errorCode。 */ + private async abortJob( + jobRowId: string, + errorCode: string, + ): Promise { + await this.prisma.watermark_jobs.update({ + where: { id: jobRowId }, + data: { state: 'failed', errorCode }, + }); + this.logger?.error({ jobId: jobRowId, errorCode }, '水印重生成任务中止'); + return 'failed'; + } + + /** + * 完成事务:job done + 有效版本推进 + album.updated outbox。乐观重试同 + * albums.service.update(并发相册更新败方整事务重跑;3 次仍败抛错交还 + * BullMQ 重试——照片处理幂等,完成事务随重投递收敛)。乱序完成的旧任务 + * (有效版本已更高)跳过推进、只置 done。 + */ + private async completeJob( + jobRowId: string, + albumId: string, + configVersion: number, + config: StoredWatermarkConfig, + ): Promise { + let outcome: 'done' | 'retry' = 'retry'; + for (let attempt = 0; attempt < 3 && outcome === 'retry'; attempt++) { + outcome = await this.prisma.$transaction(async (tx) => { + await tx.watermark_jobs.updateMany({ + where: { id: jobRowId, state: 'running' }, + data: { state: 'done', errorCode: null }, + }); + const album = await tx.albums.findUnique({ + where: { id: albumId }, + select: { entityVersion: true, watermarkEffectiveVersion: true }, + }); + if (album === null || album.watermarkEffectiveVersion >= configVersion) { + return 'done' as const; + } + const claimed = await tx.albums.updateMany({ + where: { + id: albumId, + entityVersion: album.entityVersion, + watermarkEffectiveVersion: album.watermarkEffectiveVersion, + }, + data: { + watermarkEffectiveVersion: configVersion, + watermarkEnabled: config.enabled, + entityVersion: { increment: 1 }, + }, + }); + if (claimed.count === 0) { + return 'retry' as const; + } + const updated = await tx.albums.findUniqueOrThrow({ + where: { id: albumId }, + }); + const translations = await tx.album_translations.findMany({ + where: { albumId }, + orderBy: { locale: 'asc' }, + }); + await tx.outbox_events.create({ + data: { + type: 'album.updated', + albumId, + entityVersion: updated.entityVersion, + payload: buildAlbumUpdatedPayloadCore( + updated, + translations, + ) as unknown as Prisma.InputJsonValue, + }, + }); + return 'done' as const; + }); + } + if (outcome !== 'done') { + throw new Error(`watermark: 完成事务乐观重试耗尽 jobId=${jobRowId}`); + } + } + + /** 流式拉取原图并聚合为单缓冲(同 publish.service.readOriginal 约定)。 */ + private async readOriginal(objectKey: string): Promise { + const stream = (await this.storage.getStream(objectKey)) as Readable; + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array)); + } + return Buffer.concat(chunks); + } +} diff --git a/services/worker/src/watermark.spec.ts b/services/worker/src/watermark.spec.ts new file mode 100644 index 0000000..beaaa2b --- /dev/null +++ b/services/worker/src/watermark.spec.ts @@ -0,0 +1,131 @@ +// services/worker/src/watermark.spec.ts —— composeWatermarkBrowse 单测(P4a +// Task 3,真实 sharp 缓冲;fixture 测试内生成,无二进制入库)。断言口径对应 +// 实现注释:与 browse 变体同参基线(rotate + ≤1920 withoutEnlargement + q82) +// ——enabled=false 输出与 runPipeline().browse **字节一致**;开启文字后输出 +// 必须与基线不同(水印本身即差异来源,e2e 的「水印图 ≠ 浏览图」断言依据)。 +// CJK 字形可用性是环境属性(librsvg 走系统字体链):此处硬断言只到「非空且 +// 有差异」,另以名义层宽度测量做**字体探针**(console.log 供验收记录——CJK +// 宽度显著大于同长度 Latin 缩窄串即字形已渲染;缺字体时为替代框,G5 登记)。 +import sharp from 'sharp'; +import { + normalizeWatermarkConfig, + type StoredWatermarkConfig, +} from '@photo/contracts'; +import { runPipeline } from './pipeline'; +import { composeWatermarkBrowse } from './watermark'; + +/** 生成带噪点纹理的 JPEG(纯色基线上合成水印差异更稳:JPEG 区域差可观测)。 */ +async function makeTexturedJpeg(width: number, height: number): Promise { + const svg = Buffer.from( + `` + + Array.from( + { length: 24 }, + (_, i) => + ``, + ).join('') + + '', + ); + return sharp(svg).jpeg({ quality: 85 }).toBuffer(); +} + +let BASE_INPUT: Buffer; + +/** 快捷构造完整配置(缺省经 normalize 填齐——与落库形状一致)。 */ +function config( + overrides: Omit, 'text'> & { + text?: Partial; + } = {}, +): StoredWatermarkConfig { + const base = normalizeWatermarkConfig({ enabled: true, text: { content: 'Photo Live' } }); + return { + ...base, + ...overrides, + text: { ...base.text, ...overrides.text }, + }; +} + +/** 名义文字层渲染探针(与实现 textSvg 同构;仅测量,不作硬断言)。 */ +async function probeLayerWidth(content: string, fontSize = 24): Promise { + const svg = Buffer.from( + `` + + `${content}`, + ); + const trimmed = await sharp(await sharp(svg).png().toBuffer()).trim().toBuffer(); + return (await sharp(trimmed).metadata()).width ?? 0; +} + +describe('composeWatermarkBrowse', () => { + beforeAll(async () => { + BASE_INPUT = await makeTexturedJpeg(1200, 800); + }); + + test('enabled=false:与 runPipeline 的 browse 变体字节一致(同参基线)', async () => { + const clean = await composeWatermarkBrowse(BASE_INPUT, config({ enabled: false })); + const pipeline = await runPipeline(BASE_INPUT); + expect(clean.width).toBe(pipeline.browseWidth); + expect(clean.height).toBe(pipeline.browseHeight); + expect(clean.buffer.equals(pipeline.browse)).toBe(true); + }); + + test('开启文字:非空 jpeg、尺寸同基线、字节与基线不同(水印即差异)', async () => { + const out = await composeWatermarkBrowse(BASE_INPUT, config()); + expect(out.buffer.length).toBeGreaterThan(0); + const meta = await sharp(out.buffer).metadata(); + expect(meta.format).toBe('jpeg'); + expect(out.width).toBe(1200); // withoutEnlargement:小图不放大 + expect(out.height).toBe(800); + const pipeline = await runPipeline(BASE_INPUT); + expect(out.buffer.equals(pipeline.browse)).toBe(false); + }); + + test('位置语义:topLeft 与 bottomRight 输出不同;center 亦不同', async () => { + const topLeft = await composeWatermarkBrowse(BASE_INPUT, config({ position: 'topLeft' })); + const bottomRight = await composeWatermarkBrowse(BASE_INPUT, config({ position: 'bottomRight' })); + const center = await composeWatermarkBrowse(BASE_INPUT, config({ position: 'center' })); + expect(topLeft.buffer.equals(bottomRight.buffer)).toBe(false); + expect(topLeft.buffer.equals(center.buffer)).toBe(false); + }); + + test('不透明度/尺寸参数生效:opacity 90 vs 10、scale 10% vs 40% 输出不同', async () => { + const opaque = await composeWatermarkBrowse(BASE_INPUT, config({ opacity: 90 })); + const faint = await composeWatermarkBrowse(BASE_INPUT, config({ opacity: 10 })); + expect(opaque.buffer.equals(faint.buffer)).toBe(false); + + const small = await composeWatermarkBrowse(BASE_INPUT, config({ scalePercent: 10 })); + const large = await composeWatermarkBrowse(BASE_INPUT, config({ scalePercent: 40 })); + expect(small.buffer.equals(large.buffer)).toBe(false); + }); + + test('CJK 内容:非空且与基线/英文水印均不同(字体探针记录名义层宽度)', async () => { + const cjk = await composeWatermarkBrowse( + BASE_INPUT, + config({ text: { content: '照片直播 水印' } }), + ); + const english = await composeWatermarkBrowse(BASE_INPUT, config()); + const pipeline = await runPipeline(BASE_INPUT); + expect(cjk.buffer.length).toBeGreaterThan(0); + expect(cjk.buffer.equals(pipeline.browse)).toBe(false); + expect(cjk.buffer.equals(english.buffer)).toBe(false); + + // 字体探针(信息性,不作断言):CJK 全宽字形应显著宽于同字符数 Latin + // 窄字串;宽度过小/为零 = 环境缺 CJK 字体(渲染为空/替代框),G5 登记。 + const cjkWidth = await probeLayerWidth('照片直播水印'); + const latinWidth = await probeLayerWidth('iiiiii'); + // eslint-disable-next-line no-console + console.log( + `[watermark font probe] CJK 名义层宽=${cjkWidth}px, Latin(i×6) 名义层宽=${latinWidth}px` + + (cjkWidth > latinWidth ? '(CJK 字形已渲染)' : '(疑似缺 CJK 字体——英文兜底)'), + ); + expect(cjkWidth).toBeGreaterThan(0); + }); + + test('XML 特殊字符内容:正常渲染不注入(输出仍与基线不同)', async () => { + const out = await composeWatermarkBrowse( + BASE_INPUT, + config({ text: { content: '&"quote"' } }), + ); + const pipeline = await runPipeline(BASE_INPUT); + expect(out.buffer.length).toBeGreaterThan(0); + expect(out.buffer.equals(pipeline.browse)).toBe(false); + }); +}); diff --git a/services/worker/src/watermark.ts b/services/worker/src/watermark.ts new file mode 100644 index 0000000..09b38d3 --- /dev/null +++ b/services/worker/src/watermark.ts @@ -0,0 +1,167 @@ +// services/worker/src/watermark.ts —— 水印合成流水线(P4a Task 3)。 +// 实现方式登记(技术方案 7.2 / 计划执行注意):sharp 无原生文字渲染,文字水印 +// 经 librsvg 渲染 SVG `` 层(font-family="sans-serif" 走系统字体链;CJK +// 字形可用性取决于运行环境字体——e2e 以「输出非空且与无水印图不同」断言, +// 精确字形不可断言,环境缺 CJK 字体时渲染为替代框/空白,英文水印为兜底, +// G5 字体事项登记)。合成流程: +// ① 名义字号渲染文字 SVG → PNG → .trim() 得紧凑文字层(对字形宽度不做任何 +// 假设——CJK 全宽 / Latin 变宽都由 trim 后实测尺寸消化); +// ② 文字层等比缩放到目标宽度 = 浏览图宽 × scalePercent%(scalePercent 语义 +// 「水印尺寸相对浏览图宽度的百分比」); +// ③ 与浏览图基线(rotate + resize ≤1920,同 pipeline.ts 规格)composite, +// 位置按 config.position + 2% 边距,输出 JPEG q82(与 browse 变体同参, +// 便于字节级对比断言)。 +// 不透明度经 SVG fill-opacity 承载(缩放不改变相对透明度)。 +import sharp from 'sharp'; +import type { StoredWatermarkConfig, WatermarkPosition } from '@photo/contracts'; +import { BROWSE_MAX_WIDTH } from './pipeline'; + +/** 文字层与浏览图边缘的边距(浏览图宽的 2%,至少 4px;center 无边距)。 */ +const MARGIN_RATIO = 0.02; +const MIN_MARGIN_PX = 4; + +/** XML 属性/文本转义(用户可控的 content 进入 SVG 标记,防注入)。 */ +function escapeXml(value: string): string { + return value + .replaceAll('&', '&') + .replaceAll('<', '<') + .replaceAll('>', '>') + .replaceAll('"', '"') + .replaceAll("'", '''); +} + +/** + * 名义渲染:fontSize 即 SVG font-size(px @72dpi)。画布按「最宽可能」预估 + * (每字符 2×fontSize + 两侧 padding)——后续 trim 收紧,宁可画大不裁字。 + */ +function textSvg(config: StoredWatermarkConfig): string { + const fontSize = config.text.fontSize; + const content = escapeXml(config.text.content); + const canvasWidth = Math.ceil(content.length * fontSize * 2 + fontSize * 2); + const canvasHeight = Math.ceil(fontSize * 3); + return ( + `` + + `${content}` + + `` + ); +} + +/** 渲染并裁紧文字层(透明底 PNG)。空内容产生零宽层 → 调用方视为无水印。 */ +async function renderTextLayer(config: StoredWatermarkConfig): Promise { + const png = await sharp(Buffer.from(textSvg(config), 'utf8')) + .png() + .toBuffer(); + return sharp(png).trim().toBuffer(); +} + +/** composite 左上角偏移(position + 边距;层尺寸可能超过基线时收钳)。 */ +function layerOffset( + position: WatermarkPosition, + baseWidth: number, + baseHeight: number, + layerWidth: number, + layerHeight: number, +): { left: number; top: number } { + const margin = Math.max( + MIN_MARGIN_PX, + Math.round(baseWidth * MARGIN_RATIO), + ); + const clamp = (value: number, max: number): number => + Math.max(0, Math.min(value, max)); + const right = baseWidth - layerWidth - margin; + const bottom = baseHeight - layerHeight - margin; + switch (position) { + case 'topLeft': + return { left: clamp(margin, baseWidth), top: clamp(margin, baseHeight) }; + case 'topRight': + return { left: clamp(right, baseWidth), top: clamp(margin, baseHeight) }; + case 'bottomLeft': + return { + left: clamp(margin, baseWidth), + top: clamp(bottom, baseHeight), + }; + case 'bottomRight': + return { left: clamp(right, baseWidth), top: clamp(bottom, baseHeight) }; + case 'center': + return { + left: clamp(Math.round((baseWidth - layerWidth) / 2), baseWidth), + top: clamp(Math.round((baseHeight - layerHeight) / 2), baseHeight), + }; + } +} + +export interface WatermarkComposeResult { + buffer: Buffer; + width: number; + height: number; +} + +/** + * 原图 → 带水印浏览图(发布路径与历史重生成共用)。基线与 pipeline.ts 的 + * browse 变体同参(rotate + ≤1920 withoutEnlargement + JPEG q82),仅多一层 + * 文字合成——与 browse 变体的字节差异即水印本身(e2e 断言依据)。 + * enabled=false 或空文字层:不合成,直接产出与 browse 同参的干净图(调用方 + * 负责「关闭 = 不产变体」的语义,此返回仅防御性兜底)。 + */ +export async function composeWatermarkBrowse( + input: Buffer, + config: StoredWatermarkConfig, +): Promise { + // 目标尺寸解析析计算(同 pipeline.ts 的 photos.width/height 口径):sharp 的 + // .metadata() 返回**输入**尺寸而非 resize 后尺寸,管线实例上取 meta 会拿到 + // 缩放前值;withoutEnlargement 下 outW = min(w, 1920) 精确,outH 按比例取整 + // (±1px 对层偏移无影响——边距/收钳兜住;精确尺寸以最终输出 metadata 为准)。 + const meta = await sharp(input).metadata(); + const swap = (meta.orientation ?? 1) >= 5; + const srcWidth = (swap ? meta.height : meta.width) ?? 0; + const srcHeight = (swap ? meta.width : meta.height) ?? 0; + if (srcWidth <= 0 || srcHeight <= 0) { + throw new Error('watermark: 无法解析原图尺寸'); + } + const baseWidth = Math.min(srcWidth, BROWSE_MAX_WIDTH); + const baseHeight = Math.max(1, Math.round((srcHeight * baseWidth) / srcWidth)); + + let pipeline = sharp(input) + .rotate() + .resize({ width: BROWSE_MAX_WIDTH, withoutEnlargement: true }); + if (config.enabled && config.text.content !== '') { + const nominal = await renderTextLayer(config); + const nominalMeta = await sharp(nominal).metadata(); + const nominalWidth = nominalMeta.width ?? 0; + const nominalHeight = nominalMeta.height ?? 0; + if (nominalWidth > 0 && nominalHeight > 0) { + // 目标宽度 = 浏览图宽 × scalePercent%(min 1px;层不过基线——scalePercent + // 上限 50%,只剩边距空间,天然满足)。 + const targetWidth = Math.max( + 1, + Math.min( + Math.round((baseWidth * config.scalePercent) / 100), + baseWidth, + ), + ); + const layer = await sharp(nominal) + .resize({ width: targetWidth }) + .png() + .toBuffer(); + const layerMeta = await sharp(layer).metadata(); + const { left, top } = layerOffset( + config.position, + baseWidth, + baseHeight, + layerMeta.width ?? targetWidth, + layerMeta.height ?? 1, + ); + pipeline = pipeline.composite([{ input: layer, left, top }]); + } + } + + const buffer = await pipeline.jpeg({ quality: 82 }).toBuffer(); + const outMeta = await sharp(buffer).metadata(); + return { + buffer, + width: outMeta.width ?? baseWidth, + height: outMeta.height ?? baseHeight, + }; +} diff --git a/services/worker/test/reset.ts b/services/worker/test/reset.ts index a68a7af..bacee2f 100644 --- a/services/worker/test/reset.ts +++ b/services/worker/test/reset.ts @@ -19,6 +19,9 @@ export const RESET_TABLES: readonly string[] = [ 'photo_variants', 'upload_sessions', 'photos', + // 水印域(P4a Task 3,同 api/test/reset.ts)。 + 'watermark_jobs', + 'watermark_configs', 'category_translations', 'categories', 'album_assets', diff --git a/services/worker/test/watermark.e2e-spec.ts b/services/worker/test/watermark.e2e-spec.ts new file mode 100644 index 0000000..463320d --- /dev/null +++ b/services/worker/test/watermark.e2e-spec.ts @@ -0,0 +1,468 @@ +// services/worker/test/watermark.e2e-spec.ts —— 水印历史重生成 e2e(P4a Task 3, +// 先写后实现)。真实 MySQL(直种 album/photos/watermark_configs/watermark_jobs, +// 形状对齐 api WatermarkService.create 的落库结果)+ 真实 MinIO(原图/变体对象) +// + 真实 Redis(BullMQ)。分层驱动同 publish.e2e-spec:核心断言直驱 +// WatermarkRegenService,一组真实入队用例覆盖 Worker 包装。覆盖: +// ① 重生成(enabled):逐张产水印变体(对象 + 行 version=configVersion)、 +// progress→total、job done、**有效版本推进**(watermarkEffectiveVersion/ +// watermarkEnabled + entityVersion+1)+ outbox album.updated(载荷水印位); +// ② A20:原图对象键/字节零触碰;browse 变体行不动; +// ③ 失败注入:单张原图被删 → failed=1、job 仍 done(部分成功)、失败张无水印行 +// (cn 保留旧图=无行回落 browse)、旧图对象拷贝到目标版本键(ov 模式键 +// 存在性保障——A20 失败照片保留旧图); +// ④ 关闭版重生成:水印行逐张删除(对象保留)、有效位 enabled=false、outbox +// 载荷 enabled=false + 新版本; +// ⑤ 任务级中止:配置行缺失 → state=failed + WATERMARK_CONFIG_MISSING、照片 +// 零处理、有效位不动; +// ⑥ 真实 BullMQ Worker:api 生产者同参入队 → 消费 → done;幽灵投递 skipped。 +import { randomBytes } from 'node:crypto'; +import { Readable } from 'node:stream'; +import { DeleteObjectCommand, S3Client } from '@aws-sdk/client-s3'; +import { Queue, type Worker } from 'bullmq'; +import IORedis from 'ioredis'; +import type { Logger } from 'pino'; +import sharp from 'sharp'; +import { PrismaClient } from '@photo/db-cn'; +import type { Prisma } from '@photo/db-cn'; +import { QUEUE_WATERMARK, normalizeWatermarkConfig } from '@photo/contracts'; +import { + createStorageFromEnv, + watermarkVariantObjectKey, + type MinioStorageAdapter, + type StoragePort, +} from '@photo/storage'; +import { + WATERMARK_REGEN_JOB_NAME, + WATERMARK_JOB_ATTEMPTS, + WATERMARK_JOB_BACKOFF, + createWatermarkRegenWorker, + type WatermarkRegenJobData, + type WatermarkRegenWorkerOptions, +} from '../src/watermark.processor'; +import { + WatermarkRegenService, + WATERMARK_ERROR_CONFIG_MISSING, + type WatermarkJobOutcome, +} from '../src/watermark.service'; +import { resetDb } from './reset'; + +jest.setTimeout(120_000); + +/** 轮询等待:e2e 对 BullMQ 异步路径的断言前置。 */ +async function waitFor( + check: () => Promise, + timeoutMs = 30_000, + stepMs = 100, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (await check()) return; + if (Date.now() > deadline) { + throw new Error(`waitFor 超时(${timeoutMs}ms)`); + } + await new Promise((resolve) => setTimeout(resolve, stepMs)); + } +} + +function cuidLike(alias: string): string { + return `c${Date.now().toString(36)}${alias}${randomBytes(6).toString('hex')}`; +} + +/** 测试内生成真实 JPEG(带纹理,水印差异可观测)。 */ +async function makeJpeg(width: number, height: number): Promise { + const svg = Buffer.from( + `` + + Array.from( + { length: 12 }, + (_, i) => + ``, + ).join('') + + '', + ); + return sharp(svg).jpeg({ quality: 85 }).toBuffer(); +} + +const KEY_ROOT = 'photo-live/dev'; + +/** 流式读取对象为单缓冲(与 worker readOriginal 同约定)。 */ +async function readObject(storage: StoragePort, key: string): Promise { + const stream = (await storage.getStream(key)) as Readable; + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array)); + } + return Buffer.concat(chunks); +} + +/** S3 协议直删单对象(StoragePort 无删除——YAGNI,仅测试隔离/注入用)。 */ +async function deleteObject(key: string): Promise { + const client = 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 client.send(new DeleteObjectCommand({ Bucket: process.env.S3_BUCKET, Key: key })); + } finally { + await client.destroy(); + } +} + +describe('watermark regen (e2e)', () => { + let prisma: PrismaClient; + let storage: MinioStorageAdapter; + let service: WatermarkRegenService; + let queue: Queue; + let queueConnection: IORedis; + + /** 直种 album(水印位可调,形状对齐 api create 提交后的状态)。 */ + const seedAlbum = async ( + alias: string, + watermark: { version: number; effective: number; enabled: boolean }, + ): Promise => { + const suffix = randomBytes(4).toString('hex'); + return ( + await prisma.albums.create({ + data: { + publicToken: `pt-${alias}-${suffix}`, + ownerId: `u-${alias}`, + startAt: new Date(), + defaultLocale: 'zh-CN', + watermarkVersion: watermark.version, + watermarkEffectiveVersion: watermark.effective, + watermarkEnabled: watermark.enabled, + }, + select: { id: true }, + }) + ).id; + }; + + /** 直种已发布照片(原图对象 + browse/thumb 变体行与对象——发布后形态)。 */ + const seedPublishedPhoto = async ( + albumId: string, + alias: string, + options: { withOriginalObject?: boolean } = {}, + ): Promise<{ + photoId: string; + objectKey: string; + browseKey: string; + original: Buffer; + browse: Buffer; + }> => { + const photoId = cuidLike(alias); + const objectKey = `${KEY_ROOT}/albums/${albumId}/originals/${photoId}.jpg`; + const browseKey = `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/browse.jpg`; + const thumbKey = `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/thumb.jpg`; + const original = await makeJpeg(900, 600); + if (options.withOriginalObject !== false) { + await storage.put(objectKey, original, 'image/jpeg'); + } + const browse = await sharp(original) + .resize({ width: 1920, withoutEnlargement: true }) + .jpeg({ quality: 82 }) + .toBuffer(); + const thumb = await sharp(original) + .resize({ width: 480, withoutEnlargement: true }) + .jpeg({ quality: 75 }) + .toBuffer(); + await storage.put(browseKey, browse, 'image/jpeg'); + await storage.put(thumbKey, thumb, 'image/jpeg'); + await prisma.photos.create({ + data: { + id: photoId, + albumId, + uploaderId: `u-${alias}`, + objectKey, + bytes: original.length, + processingState: 'ready', + visibility: 'published', + publishedAt: new Date(), + width: 900, + height: 600, + }, + }); + await prisma.photo_variants.createMany({ + data: [ + { photoId, variant: 'thumb', objectKey: thumbKey, bytes: thumb.length, version: 1, width: 480, height: 320 }, + { photoId, variant: 'browse', objectKey: browseKey, bytes: browse.length, version: 1, width: 900, height: 600 }, + ], + }); + return { photoId, objectKey, browseKey, original, browse }; + }; + + /** 直种配置行 + 任务行(形状对齐 api WatermarkService.create 落库结果)。 */ + const seedWatermarkJob = async ( + albumId: string, + configVersion: number, + enabled: boolean, + total: number, + ): Promise => { + await prisma.watermark_configs.create({ + data: { + albumId, + configVersion, + config: normalizeWatermarkConfig({ + enabled, + ...(enabled ? { text: { content: `WM v${configVersion}` } } : {}), + }) as unknown as Prisma.InputJsonValue, + }, + }); + return ( + await prisma.watermark_jobs.create({ + data: { albumId, configVersion, state: 'pending', total }, + select: { id: true }, + }) + ).id; + }; + + beforeAll(async () => { + await resetDb(); + prisma = new PrismaClient(); + storage = createStorageFromEnv(); + service = new WatermarkRegenService(prisma, storage); + queueConnection = new IORedis( + process.env.REDIS_URL_CN ?? 'redis://localhost:6379', + { maxRetriesPerRequest: null }, + ); + queueConnection.on('error', () => undefined); + queue = new Queue(QUEUE_WATERMARK, { connection: queueConnection }); + }); + + afterAll(async () => { + await queue.obliterate({ force: true }).catch(() => undefined); + await queue.close().catch(() => undefined); + await queueConnection.quit().catch(() => undefined); + await prisma.$disconnect(); + }); + + test('重生成(enabled):变体/进度/有效位推进/outbox,A20 原图与 browse 零触碰', async () => { + const albumId = await seedAlbum('wm1', { version: 1, effective: 0, enabled: false }); + const photos = [ + await seedPublishedPhoto(albumId, 'wm1a'), + await seedPublishedPhoto(albumId, 'wm1b'), + await seedPublishedPhoto(albumId, 'wm1c'), + ]; + const jobId = await seedWatermarkJob(albumId, 1, true, photos.length); + + const outcome = await service.processJob(jobId); + expect(outcome).toBe('done'); + + const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } }); + expect(job.state).toBe('done'); + expect(job.progress).toBe(3); + expect(job.total).toBe(3); + expect(job.failed).toBe(0); + expect(job.errorCode).toBeNull(); + + // 逐张:水印行(version=1)+ 对象存在且为非空 jpeg;与 browse 字节不同。 + for (const { photoId, objectKey, original } of photos) { + const row = await prisma.photo_variants.findUnique({ + where: { photoId_variant: { photoId, variant: 'watermark' } }, + }); + expect(row?.version).toBe(1); + expect(row?.objectKey).toBe(watermarkVariantObjectKey(albumId, photoId, 1)); + const wmObject = await readObject(storage, row!.objectKey); + expect(wmObject.length).toBeGreaterThan(0); + expect((await sharp(wmObject).metadata()).format).toBe('jpeg'); + const browseRow = await prisma.photo_variants.findUniqueOrThrow({ + where: { photoId_variant: { photoId, variant: 'browse' } }, + }); + expect(browseRow.version).toBe(1); // browse 变体行零触碰 + expect(wmObject.equals(await readObject(storage, browseRow.objectKey))).toBe(false); + // A20:originals 键与字节零触碰。 + expect(await readObject(storage, objectKey)).toEqual(original); + } + + // 有效位推进 + 实体版本 +1。 + const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkEffectiveVersion).toBe(1); + expect(album.watermarkEnabled).toBe(true); + expect(album.watermarkVersion).toBe(1); + expect(album.entityVersion).toBe(2); // 种子 1 → 推进后 2 + + // outbox album.updated:载荷水印位 = 推进后的有效状态。 + const event = await prisma.outbox_events.findFirstOrThrow({ + where: { albumId, type: 'album.updated' }, + }); + expect(event.entityVersion).toBe(2); + expect(event.priority).toBe('normal'); + expect(event.payload).toMatchObject({ + watermarkEnabled: true, + watermarkConfigVersion: 1, + }); + }); + + test('失败注入:单张原图缺失 → failed=1 仍 done,失败张无行 + 旧图拷贝到目标键', async () => { + const albumId = await seedAlbum('wm2', { version: 1, effective: 0, enabled: false }); + const ok = await seedPublishedPhoto(albumId, 'wm2a'); + const broken = await seedPublishedPhoto(albumId, 'wm2b'); + await deleteObject(broken.objectKey); // 模拟历史原图对象丢失(DB 行保留) + expect(await storage.head(broken.objectKey)).toBeNull(); + + const jobId = await seedWatermarkJob(albumId, 1, true, 2); + const outcome = await service.processJob(jobId); + expect(outcome).toBe('done'); + + const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } }); + expect(job.state).toBe('done'); + expect(job.failed).toBe(1); + expect(job.progress).toBe(2); + + // 成功张:水印行/对象齐备。 + const okRow = await prisma.photo_variants.findUnique({ + where: { photoId_variant: { photoId: ok.photoId, variant: 'watermark' } }, + }); + expect(okRow?.version).toBe(1); + + // 失败张:cn 无水印行(保留旧图=该张此前无水印,回落 browse);ov 模式键 + // 保障——旧图(browse 对象)已拷贝到目标版本键,字节一致。 + const brokenRow = await prisma.photo_variants.findUnique({ + where: { photoId_variant: { photoId: broken.photoId, variant: 'watermark' } }, + }); + expect(brokenRow).toBeNull(); + const fallbackKey = watermarkVariantObjectKey(albumId, broken.photoId, 1); + const fallbackObject = await readObject(storage, fallbackKey); + expect(fallbackObject.equals(broken.browse)).toBe(true); + + // 部分成功仍推进有效位(done 即切换;失败张 ov 拿旧图、cn 回落 browse)。 + const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkEffectiveVersion).toBe(1); + expect(album.watermarkEnabled).toBe(true); + }); + + test('关闭版重生成:水印行删除(对象保留)、有效位 enabled=false、outbox 同步', async () => { + const albumId = await seedAlbum('wm3', { version: 2, effective: 0, enabled: false }); + const photo = await seedPublishedPhoto(albumId, 'wm3a'); + // 前置:v1 开启版先行生效(行在 v1 键、有效位推进)。 + expect(await service.processJob(await seedWatermarkJob(albumId, 1, true, 1))).toBe('done'); + const v1Key = watermarkVariantObjectKey(albumId, photo.photoId, 1); + const v1Row = await prisma.photo_variants.findUniqueOrThrow({ + where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } }, + }); + expect(v1Row.objectKey).toBe(v1Key); + + // 关闭版(v2 enabled=false)重生成:删行,对象保留。 + const jobId = await seedWatermarkJob(albumId, 2, false, 1); + expect(await service.processJob(jobId)).toBe('done'); + + expect( + await prisma.photo_variants.findUnique({ + where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } }, + }), + ).toBeNull(); + expect(await storage.head(v1Key)).not.toBeNull(); // 对象保留(历史可溯) + + const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkEffectiveVersion).toBe(2); + expect(album.watermarkEnabled).toBe(false); + expect(album.watermarkVersion).toBe(2); + + const events = await prisma.outbox_events.findMany({ + where: { albumId, type: 'album.updated' }, + orderBy: { entityVersion: 'asc' }, + }); + expect(events.at(-1)?.payload).toMatchObject({ + watermarkEnabled: false, + watermarkConfigVersion: 2, + }); + }); + + test('任务级中止:配置行缺失 → failed + WATERMARK_CONFIG_MISSING,零处理零推进', async () => { + const albumId = await seedAlbum('wm4', { version: 1, effective: 0, enabled: false }); + const photo = await seedPublishedPhoto(albumId, 'wm4a'); + // 不种 watermark_configs —— 任务指向的版本无配置。 + const jobId = ( + await prisma.watermark_jobs.create({ + data: { albumId, configVersion: 1, state: 'pending', total: 1 }, + select: { id: true }, + }) + ).id; + + expect(await service.processJob(jobId)).toBe('failed'); + const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } }); + expect(job.state).toBe('failed'); + expect(job.errorCode).toBe(WATERMARK_ERROR_CONFIG_MISSING); + expect(job.progress).toBe(0); + expect( + await prisma.photo_variants.findUnique({ + where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } }, + }), + ).toBeNull(); + const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkEffectiveVersion).toBe(0); + expect(album.watermarkEnabled).toBe(false); + expect(album.entityVersion).toBe(1); // 未推进 + }); + + describe('真实 BullMQ Worker', () => { + let worker: Worker; + let workerConnection: IORedis; + + beforeAll(async () => { + workerConnection = new IORedis( + process.env.REDIS_URL_CN ?? 'redis://localhost:6379', + { maxRetriesPerRequest: null }, + ); + workerConnection.on('error', () => undefined); + const options: WatermarkRegenWorkerOptions = { + prisma, + storage, + connection: workerConnection, + logger: console as unknown as Logger, + }; + worker = createWatermarkRegenWorker(options); + }); + + afterAll(async () => { + await worker.close(); + await workerConnection.quit().catch(() => undefined); + }); + + /** 与 api 生产者同参入队(jobId=行 id 去重 + attempts:3 指数退避 1s 起)。 */ + const enqueue = (jobId: string) => + queue.add(WATERMARK_REGEN_JOB_NAME, { jobId }, { + jobId, + attempts: WATERMARK_JOB_ATTEMPTS, + backoff: WATERMARK_JOB_BACKOFF, + }); + + test('真实入队 → 消费 → done(有效位推进)', async () => { + const albumId = await seedAlbum('wm5', { version: 1, effective: 0, enabled: false }); + const photo = await seedPublishedPhoto(albumId, 'wm5a'); + const jobId = await seedWatermarkJob(albumId, 1, true, 1); + await enqueue(jobId); + + await waitFor(async () => + (await prisma.watermark_jobs.findUnique({ where: { id: jobId } }))?.state === 'done', + ); + const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } }); + expect(job.progress).toBe(1); + expect(job.failed).toBe(0); + const row = await prisma.photo_variants.findUnique({ + where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } }, + }); + expect(row?.version).toBe(1); + expect( + await storage.head(watermarkVariantObjectKey(albumId, photo.photoId, 1)), + ).not.toBeNull(); + const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } }); + expect(album.watermarkEnabled).toBe(true); + expect(album.watermarkEffectiveVersion).toBe(1); + }); + + test('行不存在的幽灵投递 → skipped,不产生任何行/对象痕迹', async () => { + const rowsBefore = await prisma.watermark_jobs.count(); + const ghost = cuidLike('ghost'); + await enqueue(ghost); + await waitFor(async () => { + const job = await queue.getJob(ghost); + return job?.returnvalue === 'skipped'; + }); + expect(await prisma.watermark_jobs.count()).toBe(rowsBefore); + }); + }); +});