// 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 无新增未投递。 // ⑤ (ads-carousel Task 8)cn 对账 'ad' 清扫:直种两行 ads(全局+定向)→ // 签名 POST /internal/sync/reconcile {entityType:'ad'} → 2 条当前态 // wire 事件(albumId 恒空串、载荷=全量快照、确定性 recon:ad:{id}:{v} // 合成 eventId,sortOrder 升序确定性排序);随后 runOnce 走生产清扫环 // (SWEEP_ENTITY_TYPES 含 'ad')→ ads_ov 投影就位——错过的 ad.* 推送 // 由此获得 digest 之外的自愈通道(digest 相册维度覆盖不了 ads)。 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 { SYNC_SIGNATURE_HEADER, SYNC_TIMESTAMP_HEADER, signSyncBody, } from '@photo/contracts'; 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 { return sharp({ create: { width, height, channels: 3, background: rgb } }) .jpeg({ quality: 85 }) .toBuffer(); } /** reconcile 清单线格式事件(openapi SyncEvent 的断言形状,⑤ 用)。 */ interface ReconcileListedEvent { eventId: string; schemaVersion: number; entityType: string; entityId: string; albumId: string; entityVersion: number; tombstone: boolean; priority: string; payload: Record; } 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 { 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 { 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 }; }; 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 { 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); } /** * 签名 POST 到 cn 内部同步端点(⑤ 直拉 reconcile 清单用;HMAC 原始字节 * 签名同 sync.e2e-spec 的 signedPost 约定——同一 JSON.stringify 产物既作 * 签名输入也作请求体)。 */ async function signedPost(path: string, body: unknown): Promise { const raw = JSON.stringify(body); return fetch(`${cnBase}${path}`, { method: 'POST', headers: { 'content-type': 'application/json', [SYNC_TIMESTAMP_HEADER]: String(Math.floor(Date.now() / 1000)), [SYNC_SIGNATURE_HEADER]: signSyncBody(SYNC_KEY, raw), }, body: raw, }); } beforeAll(async () => { expect(SYNC_KEY).not.toBe(''); await resetDb(); await resetOvDb(); cnApp = await NestFactory.create(AppCnModule, { rawBody: true, }); configureApp(cnApp); await cnApp.listen(0); cnBase = await cnApp.getUrl(); ovApp = await NestFactory.create(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); }); // ── ads-carousel Task 8:cn 对账 'ad' 清扫(reconcile 清单直拉)── it('⑤ ad 清扫:两行 ads(全局+定向)→ 2 条当前态事件(albumId 空串/全量载荷)', async () => { // 直种 cn ads 当前态权威表(reconcile 数据源,不经 CRUD 全链路):一全局 // (albumIds 缺省=SQL NULL)+ 一定向(绑定 beforeAll 相册);sortOrder 拉开 // 以断言清单确定性排序。 const globalAd = await prismaCn.ads.create({ data: { id: 'recon-ad-global', imageObjectKey: 'photo-live/dev/ads/recon-ad-global/v1.jpg', linkUrl: 'https://sponsor.example/global', sortOrder: 10, }, }); const targetedAd = await prismaCn.ads.create({ data: { id: 'recon-ad-targeted', imageObjectKey: 'photo-live/dev/ads/recon-ad-targeted/v1.png', linkUrl: 'https://sponsor.example/targeted', sortOrder: 20, albumIds: [albumId], }, }); const res = await signedPost('/v1/internal/sync/reconcile', { entityType: 'ad', sinceVersion: 0, limit: 100, }); expect(res.status).toBe(200); const body = (await res.json()) as { events: ReconcileListedEvent[] }; expect(body.events).toHaveLength(2); // sortOrder 升序(与 ads list 同口径的全量确定性排序)。 expect(body.events.map((e) => e.entityId)).toEqual([ 'recon-ad-global', 'recon-ad-targeted', ]); // 全量 wire 形状(openapi SyncEvent):确定性合成 eventId recon:ad:{id}:{v} // ——与真实 outbox eventId(cuid)永不冲突,重放幂等锚。 expect(body.events[0]).toEqual({ eventId: 'recon:ad:recon-ad-global:1', schemaVersion: 1, entityType: 'ad', entityId: 'recon-ad-global', // 广告无相册归属:albumId 恒空串(与 outbox 写入口径一致,设计 §4)。 albumId: '', entityVersion: 1, tombstone: false, priority: 'normal', payload: { adId: 'recon-ad-global', imageObjectKey: 'photo-live/dev/ads/recon-ad-global/v1.jpg', linkUrl: 'https://sponsor.example/global', enabled: true, sortOrder: 10, albumIds: null, createdAt: globalAd.createdAt.toISOString(), entityVersion: 1, }, }); // 定向广告:绑定关系在载荷 albumIds(信封 albumId 仍空串)。 expect(body.events[1]).toEqual({ eventId: 'recon:ad:recon-ad-targeted:1', schemaVersion: 1, entityType: 'ad', entityId: 'recon-ad-targeted', albumId: '', entityVersion: 1, tombstone: false, priority: 'normal', payload: { adId: 'recon-ad-targeted', imageObjectKey: 'photo-live/dev/ads/recon-ad-targeted/v1.png', linkUrl: 'https://sponsor.example/targeted', enabled: true, sortOrder: 20, albumIds: [albumId], createdAt: targetedAd.createdAt.toISOString(), entityVersion: 1, }, }); // 生产清扫环贯通:runOnce 阶段一逐类拉清单(含 'ad')→ 签名推送 ov → // ads_ov 投影就位(模拟「ad.* outbox 事件全部投后即失」——清扫当前态 // 快照重放即补齐,ov 侧 eventId 幂等 + 版本单调承担判重)。 const stats = await reconciler.runOnce(); // album v5 + photo×2 v2 + ad×2(category ③ 已删、asset 无行)。 expect(stats.sweptEvents).toBe(5); // 仅 ad 两条为新合成 eventId(album/photo 的 recon 事件 ①-④ 已入账判重)。 expect(stats.replayApplied).toBe(2); // digest 相册维度不受 ads 影响:稳态无 bootstrap。 expect(stats.mismatchedAlbums).toEqual([]); expect( await prismaOv.ads_ov.findUnique({ where: { adId: 'recon-ad-global' } }), ).toMatchObject({ adId: 'recon-ad-global', imageObjectKey: 'photo-live/dev/ads/recon-ad-global/v1.jpg', linkUrl: 'https://sponsor.example/global', sortOrder: 10, albumIds: null, entityVersion: 1, }); expect( await prismaOv.ads_ov.findUnique({ where: { adId: 'recon-ad-targeted' } }), ).toMatchObject({ adId: 'recon-ad-targeted', sortOrder: 20, albumIds: [albumId], entityVersion: 1, }); }); });