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 再考虑流式）。
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 type { Logger } from 'pino';
import { runPipeline } from './pipeline';

/** 对象键根（与 api src/uploads/support.ts 的 OBJECT_KEY_ROOT 一致：P1a 单环境）。 */
const OBJECT_KEY_ROOT = 'photo-live/dev';

/** 变体键版本段（对应 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 等）

export function variantsObjectKey(
  albumId: string,
  photoId: string,
  version: number,
  variant: 'thumb' | 'browse',
): string {
  return `${OBJECT_KEY_ROOT}/albums/${albumId}/variants/${photoId}/${version}/${variant}.jpg`;
}

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, 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');

    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,
        },
      });
      const payload: PhotoPublishedPayload = {
        photoId,
        albumId: photo.albumId,
        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 }, '照片处理完成');
    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);
  }
}