photo-live/services/api/src/albums/watermark.service.ts

// 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<WatermarkJobCreatedResponse> {
    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<WatermarkJobsStatusResponse> {
    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, status: true },
    });
    if (album === null || album.status === 'deleted') {
      // 已删除与不存在同码 404（软删除语义，deleteAlbum）。
      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(),
      })),
    };
  }
}