// 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 副本据此切换模式键）。
// P4a Task 5 检查点：任务完成后评估用量阈值告警（水印变体行切换改变存储；
// 评估自开事务且失败只记日志——任务 done 是既成事实，见 evaluateUsageSafe）。
import { Readable } from 'node:stream';
import { evaluateUsageSafe } from '@photo/db-cn';
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);
    // P4a Task 5 检查点：完成事务提交后评估（变体行已切换，存储口径随之变化）。
    await evaluateUsageSafe(this.prisma, (err) =>
      this.logger?.warn(
        {
          jobId: jobRowId,
          err: err instanceof Error ? err.message : String(err),
        },
        '用量告警评估失败（水印任务完成后）',
      ),
    );
    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);
  }
}