photo-live/services/worker/test/publish.e2e-spec.ts

// services/worker/test/publish.e2e-spec.ts —— Task 21 发布事务/补偿扫描 e2e（先写后实现）。
// 真实 MySQL（Prisma 直种 album/photos/photo_jobs）+ 真实 MinIO（storage.put 真
// JPEG 原图，变体 head 复核）+ 真实 Redis（BullMQ）。
// 分层驱动以规避 BullMQ 时序抖动：核心断言直接调 PublishService/sweepOnce；
// Worker 包装经一组真实入队用例覆盖（happy / 终败 / 跳过）。
// 覆盖：
//   ① 发布事务：变体对象 + photo_variants(2) + photos(ready/published/尺寸/
//      entityVersion+1) + outbox(photo.published，payload=契约字段全集) 单事务一致；
//   ② 重复处理幂等：无二次变体/事件，entityVersion 不再自增；
//   ③ 状态不可处理（pending_upload）→ 跳过；
//   ④ sweepOnce：到期 pending（nextRetryAt 过期 / NULL）认领入队，done 不再认领，
//      僵尸 active（updatedAt>5min）重入队；
//   ⑤ 真实 Worker：happy 路径 done；坏文件 3 次指数退避后 photo processing_failed
//      （visibility 仍 unpublished）+ job failed(attempt=3)；pending_upload 跳过置 done。
import { randomBytes } from 'node:crypto';
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 { PhotoPublishedPayload } from '@photo/contracts';
import {
  createStorageFromEnv,
  type MinioStorageAdapter,
} from '@photo/storage';
import {
  PROCESS_JOB_ATTEMPTS,
  PROCESS_JOB_BACKOFF,
  PHOTO_PROCESS_JOB_NAME,
  createPhotoProcessWorker,
  photoProcessJobId,
  type PhotoProcessJobData,
  type ProcessOutcome,
} from '../src/process.processor';
import { PublishService } from '../src/publish.service';
import { sweepOnce } from '../src/sweep';
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));
  }
}

/** cuid 形状 ID（对象键含 photoId，须先有 ID 再一次 insert 成形，同 api 约定）。 */
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> {
  return sharp({
    create: { width, height, channels: 3, background: { r: 180, g: 120, b: 60 } },
  })
    .jpeg({ quality: 85 })
    .toBuffer();
}

describe('publish (e2e)', () => {
  let prisma: PrismaClient;
  let storage: MinioStorageAdapter;
  let publish: PublishService;
  let queue: Queue;
  let queueConnection: IORedis;

  const KEY_ROOT = 'photo-live/dev';

  interface SeedResult {
    albumId: string;
    photoId: string;
    objectKey: string;
  }

  /** 直种 album + photo(+可选原图对象) + photo_jobs(pending)。 */
  async function seedPhoto(
    alias: string,
    options: {
      state?: string;
      jobState?: string;
      nextRetryAt?: Date | null;
      updatedAt?: Date;
      original?: Buffer | null; // null = 不写对象（或调用方自写）
    } = {},
  ): Promise<SeedResult> {
    const suffix = randomBytes(4).toString('hex');
    const albumId = (
      await prisma.albums.create({
        data: {
          publicToken: `pt-${alias}-${suffix}`,
          ownerId: `u-${alias}`,
          startAt: new Date(),
          defaultLocale: 'zh-CN',
        },
        select: { id: true },
      })
    ).id;
    const photoId = cuidLike(alias);
    const objectKey = `${KEY_ROOT}/albums/${albumId}/originals/${photoId}.jpg`;
    const original =
      options.original === undefined
        ? await makeJpeg(800, 500)
        : options.original;
    if (original !== null) {
      await storage.put(objectKey, original, 'image/jpeg');
    }
    await prisma.photos.create({
      data: {
        id: photoId,
        albumId,
        uploaderId: `u-${alias}`,
        objectKey,
        bytes: original?.length ?? 0,
        processingState: options.state ?? 'uploaded',
      },
    });
    await prisma.photo_jobs.create({
      data: {
        photoId,
        type: PHOTO_PROCESS_JOB_NAME,
        state: options.jobState ?? 'pending',
        nextRetryAt: options.nextRetryAt ?? null,
        ...(options.updatedAt !== undefined ? { updatedAt: options.updatedAt } : {}),
      },
    });
    return { albumId, photoId, objectKey };
  }

  /** 变体对象键（与 src/publish.service.ts 的 variantsObjectKey 同布局的断言字面量）。 */
  const variantKey = (albumId: string, photoId: string, variant: string) =>
    `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/${variant}.jpg`;

  beforeAll(async () => {
    await resetDb();
    prisma = new PrismaClient();
    storage = createStorageFromEnv();
    publish = new PublishService(prisma, storage);
    queueConnection = new IORedis(
      process.env.REDIS_URL_CN ?? 'redis://localhost:6379',
      { maxRetriesPerRequest: null },
    );
    queueConnection.on('error', () => undefined); // 测试环境不崩进程
    queue = new Queue('photo-process', { connection: queueConnection });
  });

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

  describe('PublishService.processPhoto（直接驱动）', () => {
    test('发布事务：变体上传 + 三表单事务一致（photos/photo_variants/outbox）', async () => {
      const { albumId, photoId, objectKey } = await seedPhoto('pub1');
      const outcome = await publish.processPhoto(photoId);
      expect(outcome).toBe('published');

      // MinIO：thumb/browse 真实存在（head 复核，jpeg + 非零字节）。
      const thumbHead = await storage.head(variantKey(albumId, photoId, 'thumb'));
      const browseHead = await storage.head(variantKey(albumId, photoId, 'browse'));
      expect(thumbHead).not.toBeNull();
      expect(browseHead).not.toBeNull();
      expect(thumbHead?.contentType).toBe('image/jpeg');
      expect(thumbHead?.bytes).toBeGreaterThan(0);
      expect(browseHead?.bytes).toBeGreaterThan(0);

      // photos：ready/published/publishedAt/尺寸/entityVersion 1→2。
      const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } });
      expect(photo.processingState).toBe('ready');
      expect(photo.visibility).toBe('published');
      expect(photo.publishedAt).not.toBeNull();
      expect(photo.width).toBe(800);
      expect(photo.height).toBe(500);
      expect(photo.entityVersion).toBe(2);
      expect(photo.errorCode).toBeNull();

      // photo_variants：两行，键/尺寸/版本一致（800x500 → thumb 480x300，
      // browse 不放大 800x500）。
      const variants = await prisma.photo_variants.findMany({
        where: { photoId },
        orderBy: { variant: 'asc' }, // browse < thumb（字典序）
      });
      expect(variants).toHaveLength(2);
      const [browseRow, thumbRow] = variants;
      expect(thumbRow).toMatchObject({
        variant: 'thumb',
        objectKey: variantKey(albumId, photoId, 'thumb'),
        version: 1,
        width: 480,
        height: 300,
      });
      expect(thumbRow.bytes).toBe(thumbHead?.bytes);
      expect(browseRow).toMatchObject({
        variant: 'browse',
        objectKey: variantKey(albumId, photoId, 'browse'),
        version: 1,
        width: 800,
        height: 500,
      });
      expect(browseRow.bytes).toBe(browseHead?.bytes);

      // outbox：类型/entityVersion 对齐 photo，payload = 契约 PhotoPublishedPayload 全集。
      const event = await prisma.outbox_events.findFirstOrThrow({
        where: { photoId },
      });
      expect(event.type).toBe('photo.published');
      expect(event.albumId).toBe(albumId);
      expect(event.entityVersion).toBe(2);
      expect(event.occurredAt.getTime()).toBe(photo.publishedAt?.getTime());
      const expectedPayload: PhotoPublishedPayload = {
        photoId,
        albumId,
        objectKey,
        thumbKey: variantKey(albumId, photoId, 'thumb'),
        browseKey: variantKey(albumId, photoId, 'browse'),
        width: 800,
        height: 500,
        publishedAt: (photo.publishedAt as Date).toISOString(),
        entityVersion: 2,
      };
      expect(event.payload).toEqual(expectedPayload);
      // 全库仅此一条事件（无重复/孤儿）。
      expect(await prisma.outbox_events.count()).toBe(1);
    });

    test('重复处理幂等：无二次变体/事件，entityVersion 不再自增', async () => {
      const { photoId } = await seedPhoto('pub2');
      expect(await publish.processPhoto(photoId)).toBe('published');
      const before = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } });

      const again = await publish.processPhoto(photoId);
      expect(again).toBe('already');

      const after = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } });
      expect(after.entityVersion).toBe(before.entityVersion);
      expect(after.publishedAt?.getTime()).toBe(before.publishedAt?.getTime());
      expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(2);
      expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(1);
    });

    test('状态不可处理（pending_upload）→ 跳过，不留任何处理痕迹', async () => {
      const { photoId } = await seedPhoto('pub3', { state: 'pending_upload' });
      expect(await publish.processPhoto(photoId)).toBe('skipped');
      const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } });
      expect(photo.processingState).toBe('pending_upload');
      expect(photo.visibility).toBe('unpublished');
      expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(0);
      expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(0);
    });
  });

  describe('sweepOnce（直接驱动，无 Worker 在跑）', () => {
    test('认领到期 pending（过期/NULL）并重排，done 不认领，僵尸 active 重排', async () => {
      // 前置：直驱用例留下的 pending job 行（照片其实已处理）先置 done——
      // 既贴近真实（api 侧 job 随处理闭环），也让本用例的认领计数可精确断言。
      await prisma.photo_jobs.updateMany({
        where: { state: 'pending' },
        data: { state: 'done' },
      });

      const past = await seedPhoto('sw-a', {
        state: 'pending_upload',
        nextRetryAt: new Date(Date.now() - 60_000),
      }); // ① nextRetryAt 过期
      const nullRetry = await seedPhoto('sw-b', { state: 'pending_upload' }); // ② NULL
      const done = await seedPhoto('sw-c', {
        state: 'pending_upload',
        jobState: 'done',
      }); // done：不再认领
      const stale = await seedPhoto('sw-d', {
        state: 'pending_upload',
        jobState: 'active',
        updatedAt: new Date(Date.now() - 6 * 60_000),
      }); // 僵尸 active（>5min）

      const result = await sweepOnce({ prisma, queue });
      expect(result.claimed).toBe(2);
      expect(result.enqueueFailures).toBe(0);
      expect(result.staleReEnqueued).toBe(1);

      // DB：pending→active（过期与 NULL 都认领）；done 不动；active 保持。
      const jobs = await prisma.photo_jobs.findMany({
        where: { photoId: { in: [past.photoId, nullRetry.photoId, done.photoId, stale.photoId] } },
      });
      const stateOf = (photoId: string) =>
        jobs.find((j) => j.photoId === photoId)?.state;
      expect(stateOf(past.photoId)).toBe('active');
      expect(stateOf(nullRetry.photoId)).toBe('active');
      expect(stateOf(done.photoId)).toBe('done');
      expect(stateOf(stale.photoId)).toBe('active');

      // BullMQ：认领/重排的 jobId 已入队；done 的 jobId 不存在。
      expect(await queue.getJob(photoProcessJobId(past.photoId))).toBeDefined();
      expect(await queue.getJob(photoProcessJobId(nullRetry.photoId))).toBeDefined();
      expect(await queue.getJob(photoProcessJobId(stale.photoId))).toBeDefined();
      expect(await queue.getJob(photoProcessJobId(done.photoId))).toBeUndefined();

      // 清空队列，后续真实 Worker 用例从干净状态开始。
      await queue.obliterate({ force: true });
    });
  });

  describe('真实 BullMQ Worker', () => {
    let worker: Worker<PhotoProcessJobData, ProcessOutcome, string>;
    let workerConnection: IORedis;
    const logger = console as unknown as Logger; // e2e 直观日志

    beforeAll(async () => {
      workerConnection = new IORedis(
        process.env.REDIS_URL_CN ?? 'redis://localhost:6379',
        { maxRetriesPerRequest: null },
      );
      workerConnection.on('error', () => undefined);
      worker = createPhotoProcessWorker({
        prisma,
        storage,
        connection: workerConnection,
        logger,
      });
    });

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

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

    test('happy：真实入队 → 处理 → 发布事务 + photo_jobs done', async () => {
      const { albumId, photoId } = await seedPhoto('wk-ok');
      await enqueue(photoId);

      await waitFor(async () =>
        (await prisma.photos.findUnique({ where: { id: photoId } }))?.processingState === 'ready',
      );
      const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } });
      expect(photo.visibility).toBe('published');
      expect(photo.entityVersion).toBe(2);
      expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(2);
      expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(1);
      expect(
        await storage.head(variantKey(albumId, photoId, 'thumb')),
      ).not.toBeNull();

      const job = await prisma.photo_jobs.findUniqueOrThrow({
        where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } },
      });
      expect(job.state).toBe('done');
      expect(job.attempt).toBe(1);
    });

    test('坏文件：3 次退避重试后终败，photo processing_failed 且不公开', async () => {
      const corrupt = Buffer.concat([
        Buffer.from([0xff, 0xd8, 0xff, 0xe0]),
        randomBytes(8192),
      ]);
      const { photoId } = await seedPhoto('wk-bad', { original: corrupt });
      await enqueue(photoId);

      await waitFor(
        async () =>
          (await prisma.photo_jobs.findUnique({
            where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } },
          }))?.state === 'failed',
        60_000,
      );

      const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } });
      expect(photo.processingState).toBe('processing_failed');
      expect(photo.errorCode).toBe('DECODE_FAILED');
      expect(photo.visibility).toBe('unpublished'); // 失败不公开
      expect(photo.publishedAt).toBeNull();

      const job = await prisma.photo_jobs.findUniqueOrThrow({
        where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } },
      });
      expect(job.state).toBe('failed');
      expect(job.errorCode).toBe('DECODE_FAILED');
      expect(job.attempt).toBe(3);

      expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(0);
      expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(0);
    });

    test('pending_upload 跳过：job 置 done，照片零处理', async () => {
      const { photoId } = await seedPhoto('wk-skip', { state: 'pending_upload' });
      await enqueue(photoId);

      await waitFor(
        async () =>
          (await prisma.photo_jobs.findUnique({
            where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } },
          }))?.state === 'done',
      );
      const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } });
      expect(photo.processingState).toBe('pending_upload');
      expect(photo.visibility).toBe('unpublished');
      expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(0);
      expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(0);
    });
  });
});