feat(worker): watermark versioning with atomic history regeneration

Co-Authored-By: Claude Code <noreply@anthropic.com>

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<void> {
   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<void> {
     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<void> {
     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) {

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<PublishWatermarkTarget[]> {
+    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<number>();
+    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<number, StoredWatermarkConfig>();
+    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;
+  }
 }

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<WatermarkRegenJobData, WatermarkJobOutcome, string> {
+  const { prisma } = options;
+  const service = new WatermarkRegenService(prisma, options.storage, options.logger);
+
+  const processor = async (
+    job: Job<WatermarkRegenJobData>,
+  ): Promise<WatermarkJobOutcome> => {
+    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<WatermarkRegenJobData, WatermarkJobOutcome, string>(
+    QUEUE_WATERMARK,
+    processor,
+    {
+      connection: options.connection,
+      concurrency: options.concurrency ?? 1, // 相册级任务串行起步（多相册仍并行）
+    },
+  );
+}

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<WatermarkJobOutcome> {
+    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<void> {
+    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<void> {
+    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<WatermarkJobOutcome> {
+    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<void> {
+    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<Buffer> {
+    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);
+  }
+}

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<Buffer> {
+  const svg = Buffer.from(
+    `<svg xmlns="http://www.w3.org/2000/svg" width="${width}" height="${height}">` +
+      Array.from(
+        { length: 24 },
+        (_, i) =>
+          `<rect x="${(i * 97) % width}" y="${(i * 53) % height}" width="${40 + i * 7}" height="${30 + i * 5}" fill="#${(0x40 + i * 9).toString(16).padStart(2, '0')}7040" fill-opacity="0.8"/>`,
+      ).join('') +
+      '</svg>',
+  );
+  return sharp(svg).jpeg({ quality: 85 }).toBuffer();
+}
+
+let BASE_INPUT: Buffer;
+
+/** 快捷构造完整配置（缺省经 normalize 填齐——与落库形状一致）。 */
+function config(
+  overrides: Omit<Partial<StoredWatermarkConfig>, 'text'> & {
+    text?: Partial<StoredWatermarkConfig['text']>;
+  } = {},
+): 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<number> {
+  const svg = Buffer.from(
+    `<svg xmlns="http://www.w3.org/2000/svg" width="${content.length * fontSize * 2 + fontSize * 2}" height="${fontSize * 3}">` +
+      `<text x="${Math.ceil(fontSize / 2)}" y="${Math.ceil(fontSize * 1.5)}" font-family="sans-serif" font-size="${fontSize}" fill="#ffffff">${content}</text></svg>`,
+  );
+  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: '<b>&"quote"</b>' } }),
+    );
+    const pipeline = await runPipeline(BASE_INPUT);
+    expect(out.buffer.length).toBeGreaterThan(0);
+    expect(out.buffer.equals(pipeline.browse)).toBe(false);
+  });
+});

services/worker/src/watermark.ts

@@ -0,0 +1,167 @@
+// services/worker/src/watermark.ts —— 水印合成流水线（P4a Task 3）。
+// 实现方式登记（技术方案 7.2 / 计划执行注意）：sharp 无原生文字渲染，文字水印
+// 经 librsvg 渲染 SVG `<text>` 层（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('&', '&amp;')
+    .replaceAll('<', '&lt;')
+    .replaceAll('>', '&gt;')
+    .replaceAll('"', '&quot;')
+    .replaceAll("'", '&apos;');
+}
+
+/**
+ * 名义渲染：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 (
+    `<svg xmlns="http://www.w3.org/2000/svg" width="${canvasWidth}" height="${canvasHeight}">` +
+    `<text x="${Math.ceil(fontSize / 2)}" y="${Math.ceil(fontSize * 1.5)}" ` +
+    `font-family="sans-serif" font-size="${fontSize}" fill="#ffffff" ` +
+    `fill-opacity="${(config.opacity / 100).toFixed(3)}">${content}</text>` +
+    `</svg>`
+  );
+}
+
+/** 渲染并裁紧文字层（透明底 PNG）。空内容产生零宽层 → 调用方视为无水印。 */
+async function renderTextLayer(config: StoredWatermarkConfig): Promise<Buffer> {
+  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<WatermarkComposeResult> {
+  // 目标尺寸解析析计算（同 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,
+  };
+}