photo-live/services/worker/src/publish.service.ts

// services/worker/src/publish.service.ts —— 发布事务（Task 21，「上传即公开」的心脏）。
// 处理成功 = 单个 Prisma 事务内原子完成三件事：
//   ① photo_variants 两条 upsert（thumb/browse）；
//   ② photos 条件更新（WHERE processingState IN ('uploaded','processing_failed')）：
//      ready + published + publishedAt + 尺寸 + entityVersion+1（每次产生事件的
//      状态迁移自增；失败重试成功也切 published——上传即公开语义）；
//   ③ outbox_events 写入 photo.published（payload = 契约 PhotoPublishedPayload，
//      entityVersion 与 ② 提交后一致；P2 消费按 eventId 去重、entityVersion 单调）。
// 条件更新匹配 0 行（并发/重复投递已处理过）→ 幂等 no-op（'already'），不再
// bump entityVersion、不再补 outbox——事务即幂等边界。
// 变体对象先于事务写入 MinIO（键含 version 段，内容确定，重复写无害）；流式拉取
// 原图整缓冲驻内存（50MB 上限内可接受，P5 再考虑流式）。
// P4a Task 5 检查点：发布事务提交后评估用量阈值告警（新增照片存储可能超阈；
// 评估自开事务且失败只记日志——发布是既成事实，见 evaluateUsageSafe）。
import { Readable } from 'node:stream';
import { evaluateUsageSafe } from '@photo/db-cn';
import type { Prisma, PrismaClient } from '@photo/db-cn';
import type { StoragePort } from '@photo/storage';
import {
  variantsObjectKey,
  watermarkVariantObjectKey,
} from '@photo/storage';
import {
  parseStoredWatermarkConfig,
  type PhotoPublishedPayload,
  type StoredWatermarkConfig,
} from '@photo/contracts';
import type { Logger } from 'pino';
import { runPipeline } from './pipeline';
import { composeWatermarkBrowse, type WatermarkComposeResult } from './watermark';

/** 变体键版本段（对应 photo_variants.version；水印/规格升级时递增）。 */
const VARIANT_VERSION = 1;

/** 可处理态：首次上传（uploaded）与失败重试（processing_failed，终态仍 published）。 */
const PROCESSABLE_STATES = ['uploaded', 'processing_failed'] as const;

export type ProcessOutcome =
  | 'published' // 本次处理并切 published（含失败重试成功）
  | 'already' // 重复投递：已 ready（或并发赢家已处理），幂等 no-op
  | 'skipped'; // 照片不存在 / 状态不可处理（pending_upload、processing 等）

/** 发布路径的水印产出现状（P4a Task 3，见 resolveWatermarkTargets 注释）。 */
interface PublishWatermarkTarget {
  version: number;
  config: StoredWatermarkConfig;
  out: WatermarkComposeResult;
}

export class PublishService {
  constructor(
    private readonly prisma: PrismaClient,
    private readonly storage: StoragePort,
    private readonly logger?: Logger,
  ) {}

  /**
   * 处理一张照片：拉原图 → sharp 流水线 → 上传变体 → 原子发布事务。
   * PipelineError/存储错误原样抛出（processor 交给 BullMQ 重试，3 次耗尽置
   * processing_failed）。不抛错路径绝不留下半发布状态。
   */
  async processPhoto(photoId: string): Promise<ProcessOutcome> {
    const photo = await this.prisma.photos.findUnique({
      where: { id: photoId },
      select: {
        albumId: true,
        categoryId: true,
        objectKey: true,
        processingState: true,
      },
    });
    if (photo === null) {
      this.logger?.warn({ photoId }, '照片不存在，跳过处理');
      return 'skipped';
    }
    if (photo.processingState === 'ready') {
      // 快路径：已发布（重复投递）。真正的幂等保障仍是事务内条件更新——
      // 并发窗口内两方都读到 uploaded 时，恰一方 claim 成功。
      return 'already';
    }
    if (!(PROCESSABLE_STATES as readonly string[]).includes(photo.processingState)) {
      this.logger?.warn(
        { photoId, processingState: photo.processingState },
        '照片状态不可处理，跳过',
      );
      return 'skipped';
    }

    const input = await this.readOriginal(photo.objectKey);
    const pipeline = await runPipeline(input); // 失败抛 PipelineError → 重试
    const thumbKey = variantsObjectKey(photo.albumId, photoId, VARIANT_VERSION, 'thumb');
    const browseKey = variantsObjectKey(photo.albumId, photoId, VARIANT_VERSION, 'browse');
    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 → 已被处理，幂等返回。
      const claimed = await tx.photos.updateMany({
        where: { id: photoId, processingState: { in: [...PROCESSABLE_STATES] } },
        data: {
          processingState: 'ready',
          visibility: 'published',
          publishedAt,
          width: pipeline.width,
          height: pipeline.height,
          errorCode: null, // processing_failed 重试成功后清除错误码
          entityVersion: { increment: 1 },
        },
      });
      if (claimed.count === 0) {
        return 'already' as const;
      }
      const claimedPhoto = await tx.photos.findUniqueOrThrow({
        where: { id: photoId },
        select: { entityVersion: true },
      });
      await tx.photo_variants.upsert({
        where: { photoId_variant: { photoId, variant: 'thumb' } },
        update: {
          objectKey: thumbKey,
          bytes: pipeline.thumb.length,
          version: VARIANT_VERSION,
          width: pipeline.thumbWidth,
          height: pipeline.thumbHeight,
        },
        create: {
          photoId,
          variant: 'thumb',
          objectKey: thumbKey,
          bytes: pipeline.thumb.length,
          version: VARIANT_VERSION,
          width: pipeline.thumbWidth,
          height: pipeline.thumbHeight,
        },
      });
      await tx.photo_variants.upsert({
        where: { photoId_variant: { photoId, variant: 'browse' } },
        update: {
          objectKey: browseKey,
          bytes: pipeline.browse.length,
          version: VARIANT_VERSION,
          width: pipeline.browseWidth,
          height: pipeline.browseHeight,
        },
        create: {
          photoId,
          variant: 'browse',
          objectKey: browseKey,
          bytes: pipeline.browse.length,
          version: VARIANT_VERSION,
          width: pipeline.browseWidth,
          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,
        categoryId: photo.categoryId,
        objectKey: photo.objectKey,
        thumbKey,
        browseKey,
        width: pipeline.width,
        height: pipeline.height,
        publishedAt: publishedAt.toISOString(),
        entityVersion: claimedPhoto.entityVersion,
      };
      await tx.outbox_events.create({
        data: {
          type: 'photo.published',
          albumId: photo.albumId,
          photoId,
          entityVersion: claimedPhoto.entityVersion,
          // 契约接口无 Prisma Json 输入所需的索引签名，结构化拷贝落 Json 列。
          payload: payload as unknown as Prisma.InputJsonValue,
          occurredAt: publishedAt,
        },
      });
      return 'published' as const;
    });
    this.logger?.info({ photoId, outcome }, '照片处理完成');
    if (outcome === 'published') {
      // P4a Task 5 检查点（仅真实发布路径；already/skipped 无存储变化）。
      await evaluateUsageSafe(this.prisma, (err) =>
        this.logger?.warn(
          { photoId, err: err instanceof Error ? err.message : String(err) },
          '用量告警评估失败（发布后）',
        ),
      );
    }
    return outcome;
  }

  /** 流式拉取原图并聚合为单缓冲（S3 Body → chunks → concat）。 */
  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);
  }

  /**
   * 发布路径的水印变体产出现状（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;
  }
}