// services/worker/test/watermark.e2e-spec.ts —— 水印历史重生成 e2e（P4a Task 3，
// 先写后实现）。真实 MySQL（直种 album/photos/watermark_configs/watermark_jobs，
// 形状对齐 api WatermarkService.create 的落库结果）+ 真实 MinIO（原图/变体对象）
// + 真实 Redis（BullMQ）。分层驱动同 publish.e2e-spec：核心断言直驱
// WatermarkRegenService，一组真实入队用例覆盖 Worker 包装。覆盖：
//   ① 重生成（enabled）：逐张产水印变体（对象 + 行 version=configVersion）、
//      progress→total、job done、**有效版本推进**（watermarkEffectiveVersion/
//      watermarkEnabled + entityVersion+1）+ outbox album.updated（载荷水印位）；
//   ② A20：原图对象键/字节零触碰；browse 变体行不动；
//   ③ 失败注入：单张原图被删 → failed=1、job 仍 done（部分成功）、失败张无水印行
//      （cn 保留旧图=无行回落 browse）、旧图对象拷贝到目标版本键（ov 模式键
//      存在性保障——A20 失败照片保留旧图）；
//   ④ 关闭版重生成：水印行逐张删除（对象保留）、有效位 enabled=false、outbox
//      载荷 enabled=false + 新版本；
//   ⑤ 任务级中止：配置行缺失 → state=failed + WATERMARK_CONFIG_MISSING、照片
//      零处理、有效位不动；
//   ⑥ 真实 BullMQ Worker：api 生产者同参入队 → 消费 → done；幽灵投递 skipped。
import { randomBytes } from 'node:crypto';
import { Readable } from 'node:stream';
import { DeleteObjectCommand, S3Client } from '@aws-sdk/client-s3';
import { Queue, type Worker } from 'bullmq';
import IORedis from 'ioredis';
import type { Logger } from 'pino';
import sharp from 'sharp';
import { PrismaClient } from '@photo/db-cn';
import type { Prisma } from '@photo/db-cn';
import { QUEUE_WATERMARK, normalizeWatermarkConfig } from '@photo/contracts';
import {
  createStorageFromEnv,
  watermarkVariantObjectKey,
  type MinioStorageAdapter,
  type StoragePort,
} from '@photo/storage';
import {
  WATERMARK_REGEN_JOB_NAME,
  WATERMARK_JOB_ATTEMPTS,
  WATERMARK_JOB_BACKOFF,
  createWatermarkRegenWorker,
  type WatermarkRegenJobData,
  type WatermarkRegenWorkerOptions,
} from '../src/watermark.processor';
import {
  WatermarkRegenService,
  WATERMARK_ERROR_CONFIG_MISSING,
  type WatermarkJobOutcome,
} from '../src/watermark.service';
import { resetDb } from './reset';

jest.setTimeout(120_000);

/** 轮询等待：e2e 对 BullMQ 异步路径的断言前置。 */
async function waitFor(
  check: () => Promise<boolean>,
  timeoutMs = 30_000,
  stepMs = 100,
): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  for (;;) {
    if (await check()) return;
    if (Date.now() > deadline) {
      throw new Error(`waitFor 超时（${timeoutMs}ms）`);
    }
    await new Promise((resolve) => setTimeout(resolve, stepMs));
  }
}

function cuidLike(alias: string): string {
  return `c${Date.now().toString(36)}${alias}${randomBytes(6).toString('hex')}`;
}

/** 测试内生成真实 JPEG（带纹理，水印差异可观测）。 */
async function makeJpeg(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: 12 },
        (_, i) =>
          `<rect x="${(i * 83) % width}" y="${(i * 47) % height}" width="${50 + i * 9}" height="${35 + i * 7}" fill="#${(0x50 + i * 7).toString(16).padStart(2, '0')}6030" fill-opacity="0.85"/>`,
      ).join('') +
      '</svg>',
  );
  return sharp(svg).jpeg({ quality: 85 }).toBuffer();
}

const KEY_ROOT = 'photo-live/dev';

/** 流式读取对象为单缓冲（与 worker readOriginal 同约定）。 */
async function readObject(storage: StoragePort, key: string): Promise<Buffer> {
  const stream = (await storage.getStream(key)) 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);
}

/** S3 协议直删单对象（StoragePort 无删除——YAGNI，仅测试隔离/注入用）。 */
async function deleteObject(key: string): Promise<void> {
  const client = new S3Client({
    endpoint: process.env.S3_ENDPOINT,
    region: process.env.S3_REGION,
    credentials: {
      accessKeyId: process.env.S3_ACCESS_KEY as string,
      secretAccessKey: process.env.S3_SECRET_KEY as string,
    },
    forcePathStyle: true,
  });
  try {
    await client.send(new DeleteObjectCommand({ Bucket: process.env.S3_BUCKET, Key: key }));
  } finally {
    await client.destroy();
  }
}

describe('watermark regen (e2e)', () => {
  let prisma: PrismaClient;
  let storage: MinioStorageAdapter;
  let service: WatermarkRegenService;
  let queue: Queue;
  let queueConnection: IORedis;

  /** 直种 album（水印位可调，形状对齐 api create 提交后的状态）。 */
  const seedAlbum = async (
    alias: string,
    watermark: { version: number; effective: number; enabled: boolean },
  ): Promise<string> => {
    const suffix = randomBytes(4).toString('hex');
    return (
      await prisma.albums.create({
        data: {
          publicToken: `pt-${alias}-${suffix}`,
          ownerId: `u-${alias}`,
          startAt: new Date(),
          defaultLocale: 'zh-CN',
          watermarkVersion: watermark.version,
          watermarkEffectiveVersion: watermark.effective,
          watermarkEnabled: watermark.enabled,
        },
        select: { id: true },
      })
    ).id;
  };

  /** 直种已发布照片（原图对象 + browse/thumb 变体行与对象——发布后形态）。 */
  const seedPublishedPhoto = async (
    albumId: string,
    alias: string,
    options: { withOriginalObject?: boolean } = {},
  ): Promise<{
    photoId: string;
    objectKey: string;
    browseKey: string;
    original: Buffer;
    browse: Buffer;
  }> => {
    const photoId = cuidLike(alias);
    const objectKey = `${KEY_ROOT}/albums/${albumId}/originals/${photoId}.jpg`;
    const browseKey = `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/browse.jpg`;
    const thumbKey = `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/thumb.jpg`;
    const original = await makeJpeg(900, 600);
    if (options.withOriginalObject !== false) {
      await storage.put(objectKey, original, 'image/jpeg');
    }
    const browse = await sharp(original)
      .resize({ width: 1920, withoutEnlargement: true })
      .jpeg({ quality: 82 })
      .toBuffer();
    const thumb = await sharp(original)
      .resize({ width: 480, withoutEnlargement: true })
      .jpeg({ quality: 75 })
      .toBuffer();
    await storage.put(browseKey, browse, 'image/jpeg');
    await storage.put(thumbKey, thumb, 'image/jpeg');
    await prisma.photos.create({
      data: {
        id: photoId,
        albumId,
        uploaderId: `u-${alias}`,
        objectKey,
        bytes: original.length,
        processingState: 'ready',
        visibility: 'published',
        publishedAt: new Date(),
        width: 900,
        height: 600,
      },
    });
    await prisma.photo_variants.createMany({
      data: [
        { photoId, variant: 'thumb', objectKey: thumbKey, bytes: thumb.length, version: 1, width: 480, height: 320 },
        { photoId, variant: 'browse', objectKey: browseKey, bytes: browse.length, version: 1, width: 900, height: 600 },
      ],
    });
    return { photoId, objectKey, browseKey, original, browse };
  };

  /** 直种配置行 + 任务行（形状对齐 api WatermarkService.create 落库结果）。 */
  const seedWatermarkJob = async (
    albumId: string,
    configVersion: number,
    enabled: boolean,
    total: number,
  ): Promise<string> => {
    await prisma.watermark_configs.create({
      data: {
        albumId,
        configVersion,
        config: normalizeWatermarkConfig({
          enabled,
          ...(enabled ? { text: { content: `WM v${configVersion}` } } : {}),
        }) as unknown as Prisma.InputJsonValue,
      },
    });
    return (
      await prisma.watermark_jobs.create({
        data: { albumId, configVersion, state: 'pending', total },
        select: { id: true },
      })
    ).id;
  };

  beforeAll(async () => {
    await resetDb();
    prisma = new PrismaClient();
    storage = createStorageFromEnv();
    service = new WatermarkRegenService(prisma, storage);
    queueConnection = new IORedis(
      process.env.REDIS_URL_CN ?? 'redis://localhost:6379',
      { maxRetriesPerRequest: null },
    );
    queueConnection.on('error', () => undefined);
    queue = new Queue(QUEUE_WATERMARK, { connection: queueConnection });
  });

  afterAll(async () => {
    await queue.obliterate({ force: true }).catch(() => undefined);
    await queue.close().catch(() => undefined);
    await queueConnection.quit().catch(() => undefined);
    await prisma.$disconnect();
  });

  test('重生成（enabled）：变体/进度/有效位推进/outbox，A20 原图与 browse 零触碰', async () => {
    const albumId = await seedAlbum('wm1', { version: 1, effective: 0, enabled: false });
    const photos = [
      await seedPublishedPhoto(albumId, 'wm1a'),
      await seedPublishedPhoto(albumId, 'wm1b'),
      await seedPublishedPhoto(albumId, 'wm1c'),
    ];
    const jobId = await seedWatermarkJob(albumId, 1, true, photos.length);

    const outcome = await service.processJob(jobId);
    expect(outcome).toBe('done');

    const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
    expect(job.state).toBe('done');
    expect(job.progress).toBe(3);
    expect(job.total).toBe(3);
    expect(job.failed).toBe(0);
    expect(job.errorCode).toBeNull();

    // 逐张：水印行（version=1）+ 对象存在且为非空 jpeg；与 browse 字节不同。
    for (const { photoId, objectKey, original } of photos) {
      const row = await prisma.photo_variants.findUnique({
        where: { photoId_variant: { photoId, variant: 'watermark' } },
      });
      expect(row?.version).toBe(1);
      expect(row?.objectKey).toBe(watermarkVariantObjectKey(albumId, photoId, 1));
      const wmObject = await readObject(storage, row!.objectKey);
      expect(wmObject.length).toBeGreaterThan(0);
      expect((await sharp(wmObject).metadata()).format).toBe('jpeg');
      const browseRow = await prisma.photo_variants.findUniqueOrThrow({
        where: { photoId_variant: { photoId, variant: 'browse' } },
      });
      expect(browseRow.version).toBe(1); // browse 变体行零触碰
      expect(wmObject.equals(await readObject(storage, browseRow.objectKey))).toBe(false);
      // A20：originals 键与字节零触碰。
      expect(await readObject(storage, objectKey)).toEqual(original);
    }

    // 有效位推进 + 实体版本 +1。
    const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
    expect(album.watermarkEffectiveVersion).toBe(1);
    expect(album.watermarkEnabled).toBe(true);
    expect(album.watermarkVersion).toBe(1);
    expect(album.entityVersion).toBe(2); // 种子 1 → 推进后 2

    // outbox album.updated：载荷水印位 = 推进后的有效状态。
    const event = await prisma.outbox_events.findFirstOrThrow({
      where: { albumId, type: 'album.updated' },
    });
    expect(event.entityVersion).toBe(2);
    expect(event.priority).toBe('normal');
    expect(event.payload).toMatchObject({
      watermarkEnabled: true,
      watermarkConfigVersion: 1,
    });
  });

  test('失败注入：单张原图缺失 → failed=1 仍 done，失败张无行 + 旧图拷贝到目标键', async () => {
    const albumId = await seedAlbum('wm2', { version: 1, effective: 0, enabled: false });
    const ok = await seedPublishedPhoto(albumId, 'wm2a');
    const broken = await seedPublishedPhoto(albumId, 'wm2b');
    await deleteObject(broken.objectKey); // 模拟历史原图对象丢失（DB 行保留）
    expect(await storage.head(broken.objectKey)).toBeNull();

    const jobId = await seedWatermarkJob(albumId, 1, true, 2);
    const outcome = await service.processJob(jobId);
    expect(outcome).toBe('done');

    const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
    expect(job.state).toBe('done');
    expect(job.failed).toBe(1);
    expect(job.progress).toBe(2);

    // 成功张：水印行/对象齐备。
    const okRow = await prisma.photo_variants.findUnique({
      where: { photoId_variant: { photoId: ok.photoId, variant: 'watermark' } },
    });
    expect(okRow?.version).toBe(1);

    // 失败张：cn 无水印行（保留旧图=该张此前无水印，回落 browse）；ov 模式键
    // 保障——旧图（browse 对象）已拷贝到目标版本键，字节一致。
    const brokenRow = await prisma.photo_variants.findUnique({
      where: { photoId_variant: { photoId: broken.photoId, variant: 'watermark' } },
    });
    expect(brokenRow).toBeNull();
    const fallbackKey = watermarkVariantObjectKey(albumId, broken.photoId, 1);
    const fallbackObject = await readObject(storage, fallbackKey);
    expect(fallbackObject.equals(broken.browse)).toBe(true);

    // 部分成功仍推进有效位（done 即切换；失败张 ov 拿旧图、cn 回落 browse）。
    const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
    expect(album.watermarkEffectiveVersion).toBe(1);
    expect(album.watermarkEnabled).toBe(true);
  });

  test('关闭版重生成：水印行删除（对象保留）、有效位 enabled=false、outbox 同步', async () => {
    const albumId = await seedAlbum('wm3', { version: 2, effective: 0, enabled: false });
    const photo = await seedPublishedPhoto(albumId, 'wm3a');
    // 前置：v1 开启版先行生效（行在 v1 键、有效位推进）。
    expect(await service.processJob(await seedWatermarkJob(albumId, 1, true, 1))).toBe('done');
    const v1Key = watermarkVariantObjectKey(albumId, photo.photoId, 1);
    const v1Row = await prisma.photo_variants.findUniqueOrThrow({
      where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
    });
    expect(v1Row.objectKey).toBe(v1Key);

    // 关闭版（v2 enabled=false）重生成：删行，对象保留。
    const jobId = await seedWatermarkJob(albumId, 2, false, 1);
    expect(await service.processJob(jobId)).toBe('done');

    expect(
      await prisma.photo_variants.findUnique({
        where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
      }),
    ).toBeNull();
    expect(await storage.head(v1Key)).not.toBeNull(); // 对象保留（历史可溯）

    const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
    expect(album.watermarkEffectiveVersion).toBe(2);
    expect(album.watermarkEnabled).toBe(false);
    expect(album.watermarkVersion).toBe(2);

    const events = await prisma.outbox_events.findMany({
      where: { albumId, type: 'album.updated' },
      orderBy: { entityVersion: 'asc' },
    });
    expect(events.at(-1)?.payload).toMatchObject({
      watermarkEnabled: false,
      watermarkConfigVersion: 2,
    });
  });

  test('任务级中止：配置行缺失 → failed + WATERMARK_CONFIG_MISSING，零处理零推进', async () => {
    const albumId = await seedAlbum('wm4', { version: 1, effective: 0, enabled: false });
    const photo = await seedPublishedPhoto(albumId, 'wm4a');
    // 不种 watermark_configs —— 任务指向的版本无配置。
    const jobId = (
      await prisma.watermark_jobs.create({
        data: { albumId, configVersion: 1, state: 'pending', total: 1 },
        select: { id: true },
      })
    ).id;

    expect(await service.processJob(jobId)).toBe('failed');
    const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
    expect(job.state).toBe('failed');
    expect(job.errorCode).toBe(WATERMARK_ERROR_CONFIG_MISSING);
    expect(job.progress).toBe(0);
    expect(
      await prisma.photo_variants.findUnique({
        where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
      }),
    ).toBeNull();
    const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
    expect(album.watermarkEffectiveVersion).toBe(0);
    expect(album.watermarkEnabled).toBe(false);
    expect(album.entityVersion).toBe(1); // 未推进
  });

  describe('真实 BullMQ Worker', () => {
    let worker: Worker<WatermarkRegenJobData, WatermarkJobOutcome, string>;
    let workerConnection: IORedis;

    beforeAll(async () => {
      workerConnection = new IORedis(
        process.env.REDIS_URL_CN ?? 'redis://localhost:6379',
        { maxRetriesPerRequest: null },
      );
      workerConnection.on('error', () => undefined);
      const options: WatermarkRegenWorkerOptions = {
        prisma,
        storage,
        connection: workerConnection,
        logger: console as unknown as Logger,
      };
      worker = createWatermarkRegenWorker(options);
    });

    afterAll(async () => {
      await worker.close();
      await workerConnection.quit().catch(() => undefined);
    });

    /** 与 api 生产者同参入队（jobId=行 id 去重 + attempts:3 指数退避 1s 起）。 */
    const enqueue = (jobId: string) =>
      queue.add(WATERMARK_REGEN_JOB_NAME, { jobId }, {
        jobId,
        attempts: WATERMARK_JOB_ATTEMPTS,
        backoff: WATERMARK_JOB_BACKOFF,
      });

    test('真实入队 → 消费 → done（有效位推进）', async () => {
      const albumId = await seedAlbum('wm5', { version: 1, effective: 0, enabled: false });
      const photo = await seedPublishedPhoto(albumId, 'wm5a');
      const jobId = await seedWatermarkJob(albumId, 1, true, 1);
      await enqueue(jobId);

      await waitFor(async () =>
        (await prisma.watermark_jobs.findUnique({ where: { id: jobId } }))?.state === 'done',
      );
      const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
      expect(job.progress).toBe(1);
      expect(job.failed).toBe(0);
      const row = await prisma.photo_variants.findUnique({
        where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
      });
      expect(row?.version).toBe(1);
      expect(
        await storage.head(watermarkVariantObjectKey(albumId, photo.photoId, 1)),
      ).not.toBeNull();
      const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
      expect(album.watermarkEnabled).toBe(true);
      expect(album.watermarkEffectiveVersion).toBe(1);
    });

    test('行不存在的幽灵投递 → skipped，不产生任何行/对象痕迹', async () => {
      const rowsBefore = await prisma.watermark_jobs.count();
      const ghost = cuidLike('ghost');
      await enqueue(ghost);
      await waitFor(async () => {
        const job = await queue.getJob(ghost);
        return job?.returnvalue === 'skipped';
      });
      expect(await prisma.watermark_jobs.count()).toBe(rowsBefore);
    });
  });
});