photo-live/services/sync/test/reconcile.e2e-spec.ts

// services/sync/test/reconcile.e2e-spec.ts —— P2 Task 4 对账调度 E2E。
// 复用 sync.e2e-spec.ts 的进程内双入口 harness（cn/ov 真实 HTTP + 真实双
// MySQL + HMAC），被测对象换成生产 SyncReconciler 实例（runOnce 直驱）。
//
// 覆盖（任务验收场景）：
//   ① 漏事件追赶：ov 停在 v1、cn 已到 v5 且 outbox 事件「投后即失」→
//      runOnce 当前态清扫 → ov 副本追到 v5；
//   ② digest 修复缺失副本行：直接删 photos_ov 行（不 bump 任何版本）→
//      digest 计数不一致 → 自动触发该相册 bootstrap → 发送器投递 → 行恢复；
//   ③ digest 修复错过的分类墓碑（Task 3 登记边界）：cn 删分类且墓碑 outbox
//      行被移除（对账无法为已删行重放）→ digest categoryCount 不一致 →
//      bootstrap 的 album.updated 携带分类全量清单 → ov categories 整体替换
//      收敛为空；
//   ④ 稳态空转：一切一致时 runOnce 无 bootstrap 触发、重放全部判重/判旧
//      （applied=0）、outbox 无新增未投递。
import { NestFactory } from '@nestjs/core';
import { type NestExpressApplication } from '@nestjs/platform-express';
import request from 'supertest';
import sharp from 'sharp';
import { PrismaClient } from '@photo/db-cn';
import { PrismaClient as OvPrismaClient } from '@photo/db-ov';
import {
  createStorageFromEnv,
  type MinioStorageAdapter,
} from '@photo/storage';
import { AppCnModule } from '../../api/src/app-cn.module';
import { AppOvModule } from '../../api/src/app-ov.module';
import { seedAdmin } from '../../api/src/scripts/seed-admin';
import { configureApp } from '../../api/src/shared/configure-app';
// Task 7 起 AppOvModule 含 realtime 网关：进程内 ov 入口须与生产 main-ov
// 一致安装契约 WS 适配器（否则 Nest 落到默认 socket.io 驱动 → listen 即报错）。
import { ContractWsAdapter } from '../../api/src/realtime/contract-ws.adapter';
import { PublishService } from '../../worker/src/publish.service';
import { resetDb } from '../../api/test/reset';
import { resetOvDb } from '../../api/test/reset-ov';
import { OutboxSender } from '../src/sender';
import { SyncReconciler } from '../src/reconciler';

jest.setTimeout(180_000);

const ADMIN_ACCOUNT = process.env.ADMIN_ACCOUNT ?? 'admin';
const ADMIN_PASSWORD = process.env.ADMIN_PASSWORD ?? 'admin-secret-change-me';
const SYNC_KEY = process.env.SYNC_INTERNAL_KEY ?? 'dev-sync-secret-change-me';

/** 测试内生成真实 JPEG（同 sync.e2e-spec；无二进制入库）。 */
async function makeJpeg(
  width: number,
  height: number,
  rgb: { r: number; g: number; b: number },
): Promise<Buffer> {
  return sharp({ create: { width, height, channels: 3, background: rgb } })
    .jpeg({ quality: 85 })
    .toBuffer();
}

describe('sync reconcile + digest healing (e2e)', () => {
  let cnApp: NestExpressApplication;
  let ovApp: NestExpressApplication;
  let cnBase: string;
  let ovBase: string;
  let prismaCn: PrismaClient;
  let prismaOv: OvPrismaClient;
  let storage: MinioStorageAdapter;
  let sender: OutboxSender;
  let reconciler: SyncReconciler;
  let adminToken: string;

  let albumId: string;
  let categoryId: string;
  let photo1Id: string;
  let photo2Id: string;

  /** 排空发送器（对账触发的 bootstrap 事件由它投递，职责分离）。 */
  async function drainSender(): Promise<void> {
    for (let round = 0; round < 20; round++) {
      const stats = await sender.runOnce();
      if (stats.picked === 0) {
        return;
      }
    }
    throw new Error('drainSender 超时：20 轮后 outbox 仍有未投递事件');
  }

  /** 上传全链路：init → presign PUT 实写 MinIO → complete → 直驱发布事务。 */
  async function uploadAndPublish(
    clientTaskId: string,
    jpeg: Buffer,
  ): Promise<string> {
    const init = await request(cnBase)
      .post('/v1/uploads')
      .set('Authorization', `Bearer ${adminToken}`)
      .send({
        albumId,
        clientTaskId,
        filename: 'reconcile.jpg',
        bytes: jpeg.length,
        contentType: 'image/jpeg',
        categoryId,
      });
    expect(init.status).toBe(201);
    const { uploadId, photoId, credential } = init.body as {
      uploadId: string;
      photoId: string;
      credential: { url: string; method: string; headers: Record<string, string> };
    };
    const put = await fetch(credential.url, {
      method: credential.method,
      headers: credential.headers,
      body: new Uint8Array(jpeg),
    });
    expect(put.ok).toBe(true);
    const done = await request(cnBase)
      .post(`/v1/uploads/${uploadId}/complete`)
      .set('Authorization', `Bearer ${adminToken}`);
    expect(done.status).toBe(200);
    expect(await new PublishService(prismaCn, storage).processPhoto(photoId)).toBe(
      'published',
    );
    return photoId;
  }

  /** 相册开关翻转一次（publicEnabled 当前值取反；版本 +1、产生 album.updated）。 */
  async function togglePublicEnabled(): Promise<void> {
    const current = await prismaCn.albums.findUniqueOrThrow({
      where: { id: albumId },
      select: { publicEnabled: true },
    });
    const res = await request(cnBase)
      .patch(`/v1/albums/${albumId}`)
      .set('Authorization', `Bearer ${adminToken}`)
      .send({ publicEnabled: !current.publicEnabled });
    expect(res.status).toBe(200);
  }

  beforeAll(async () => {
    expect(SYNC_KEY).not.toBe('');
    await resetDb();
    await resetOvDb();

    cnApp = await NestFactory.create<NestExpressApplication>(AppCnModule, {
      rawBody: true,
    });
    configureApp(cnApp);
    await cnApp.listen(0);
    cnBase = await cnApp.getUrl();

    ovApp = await NestFactory.create<NestExpressApplication>(AppOvModule, {
      rawBody: true,
    });
    ovApp.useWebSocketAdapter(new ContractWsAdapter(ovApp));
    configureApp(ovApp);
    await ovApp.listen(0);
    ovBase = await ovApp.getUrl();

    prismaCn = new PrismaClient();
    prismaOv = new OvPrismaClient();
    storage = createStorageFromEnv();
    sender = new OutboxSender({
      prisma: prismaCn,
      ovApiUrl: ovBase,
      internalKey: SYNC_KEY,
    });
    reconciler = new SyncReconciler({
      prismaOv,
      cnApiUrl: cnBase,
      ovApiUrl: ovBase,
      internalKey: SYNC_KEY,
    });

    await seedAdmin(prismaCn, ADMIN_ACCOUNT, ADMIN_PASSWORD);
    const login = await request(cnBase).post('/v1/auth/password').send({
      account: ADMIN_ACCOUNT,
      password: ADMIN_PASSWORD,
    });
    expect(login.status).toBe(200);
    adminToken = login.body.sessionToken as string;

    // 基线：相册（v1）+ 分类（v1）+ 两张发布照片（各 v2）→ ov 全量同步。
    const album = await request(cnBase)
      .post('/v1/albums')
      .set('Authorization', `Bearer ${adminToken}`)
      .send({
        translations: [
          { locale: 'zh-CN', title: '对账相册' },
          { locale: 'en', title: 'Reconcile Gala' },
        ],
        defaultLocale: 'zh-CN',
        startAt: '2026-10-01T10:00:00.000Z',
      });
    expect(album.status).toBe(201);
    albumId = album.body.id as string;

    const category = await request(cnBase)
      .post(`/v1/albums/${albumId}/categories`)
      .set('Authorization', `Bearer ${adminToken}`)
      .send({
        sortOrder: 0,
        translations: [
          { locale: 'zh-CN', name: '主舞台' },
          { locale: 'en', name: 'Main Stage' },
        ],
      });
    expect(category.status).toBe(201);
    categoryId = category.body.id as string;

    photo1Id = await uploadAndPublish(
      'recon-a',
      await makeJpeg(900, 600, { r: 170, g: 120, b: 70 }),
    );
    photo2Id = await uploadAndPublish(
      'recon-b',
      await makeJpeg(800, 600, { r: 70, g: 120, b: 170 }),
    );
    await drainSender();

    expect(
      await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId } }),
    ).toMatchObject({ entityVersion: 1, publicEnabled: true });
    expect(await prismaOv.photos_ov.count({ where: { albumId } })).toBe(2);
  });

  afterAll(async () => {
    await cnApp.close();
    await ovApp.close();
    await prismaCn.$disconnect();
    await prismaOv.$disconnect();
  });

  it('① 漏事件追赶：cn 到 v5、事件「投后即失」→ runOnce 清扫 → ov 追到 v5', async () => {
    // cn 连续 4 次开关翻转：v1 → v5。
    for (let i = 0; i < 4; i += 1) {
      await togglePublicEnabled();
    }
    expect(
      (
        await prismaCn.albums.findUniqueOrThrow({
          where: { id: albumId },
          select: { entityVersion: true, publicEnabled: true },
        })
      ),
    ).toMatchObject({ entityVersion: 5, publicEnabled: true });

    // 模拟「已确认投递但 ov 从未收到」（发送器停机 + outbox 清扫后的漏账）：
    // 未投递事件全部伪造 deliveredAt —— 发送器永不再推，只能靠对账补。
    await prismaCn.outbox_events.updateMany({
      where: { deliveredAt: null },
      data: { deliveredAt: new Date() },
    });

    const stats = await reconciler.runOnce();
    // 四类型当前态：album v5 + category v1 + photo×2 v2。
    expect(stats.sweptEvents).toBe(4);
    // album v5 > ov 已应用 v1 → 应用；其余同版本重断言（Task 4 语义）。
    expect(stats.replayApplied).toBe(4);
    expect(stats.mismatchedAlbums).toEqual([]);

    expect(
      await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId } }),
    ).toMatchObject({ entityVersion: 5, publicEnabled: true });
  });

  it('② digest 修复缺失副本行：删 photos_ov 行 → bootstrap 自动触发 → 行恢复', async () => {
    // ov 侧漂移：照片行直接丢失（不 bump 任何版本——清扫的确定性 eventId
    // 已判重，唯一能发现的是 digest 计数比对）。
    await prismaOv.photos_ov.delete({ where: { photoId: photo1Id } });
    expect(await prismaOv.photos_ov.count({ where: { albumId } })).toBe(1);

    const stats = await reconciler.runOnce();
    expect(stats.mismatchedAlbums).toEqual([albumId]);
    // bootstrap 已把该相册当前状态快照写回 outbox（发送器负责投递）。
    expect(
      await prismaCn.outbox_events.count({ where: { deliveredAt: null } }),
    ).toBeGreaterThan(0);

    await drainSender();

    // 同版本重断言语义（sync-inbox ②，Task 4）：bootstrap 新 eventId + 同
    // entityVersion → 重应用 → 缺失行按当前态重建（photo2 未受牵连）。
    const restored = await prismaOv.photos_ov.findUniqueOrThrow({
      where: { photoId: photo1Id },
    });
    expect(restored).toMatchObject({
      albumId,
      visibility: 'published',
      entityVersion: 2,
    });
    expect(
      await prismaOv.photos_ov.findUniqueOrThrow({
        where: { photoId: photo2Id },
      }),
    ).toMatchObject({ visibility: 'published', entityVersion: 2 });

    // 收敛：再跑一轮 digest 无不一致、无新 bootstrap。
    const steady = await reconciler.runOnce();
    expect(steady.mismatchedAlbums).toEqual([]);
    expect(
      await prismaCn.outbox_events.count({ where: { deliveredAt: null } }),
    ).toBe(0);
  });

  it('③ digest 修复错过的分类墓碑：cn 行已删且墓碑事件丢失 → 分类全量清单整体替换收敛', async () => {
    // cn 删除分类（tombstone outbox 事件产生），随后模拟墓碑「投后即失」：
    // 直接删除 outbox 行——reconcile 清扫无法为已不存在的行重放任何事件
    // （Task 3 登记边界），ov categories 残留陈旧元素。
    const remove = await request(cnBase)
      .delete(`/v1/categories/${categoryId}`)
      .set('Authorization', `Bearer ${adminToken}`);
    expect(remove.status).toBe(204);
    await prismaCn.outbox_events.deleteMany({
      where: { type: 'category.deleted', albumId },
    });
    let replica = await prismaOv.albums_ov.findUniqueOrThrow({
      where: { albumId },
    });
    expect((replica.categories as unknown[]).length).toBe(1);

    const stats = await reconciler.runOnce();
    expect(stats.mismatchedAlbums).toEqual([albumId]);
    await drainSender();

    // bootstrap 的 album.updated 携带分类全量清单（此刻为空）→ ov 整体替换。
    replica = await prismaOv.albums_ov.findUniqueOrThrow({
      where: { albumId },
    });
    expect(replica.categories).toEqual([]);
    // cn 侧分类确已删净（对账收敛到权威态，非仅表面计数一致）。
    expect(
      await prismaCn.categories.count({ where: { albumId } }),
    ).toBe(0);
  });

  it('④ 稳态空转：一致时无 bootstrap、重放全判重（applied=0）', async () => {
    const stats = await reconciler.runOnce();
    expect(stats).toMatchObject({
      digestAlbums: 1,
      mismatchedAlbums: [],
      ovOrphanAlbums: [],
      replayApplied: 0,
    });
    // 清扫仍会拉当前态（sweptEvents>0）但全部 eventId 判重——无副作用。
    expect(stats.sweptEvents).toBeGreaterThan(0);
    expect(
      await prismaCn.outbox_events.count({ where: { deliveredAt: null } }),
    ).toBe(0);
    expect(await prismaOv.photos_ov.count({ where: { albumId } })).toBe(2);
  });
});