// services/api/test/ads.e2e-spec.ts —— 广告位 ov 消费链 e2e(ads-carousel // Task 5)。聚焦「事件→投影」直断:CRUD 端点 Task 7 才有,故手工以 // AdUpdatedPayload/AdDeletedPayload 构造 cn outbox_events 行(albumId 恒空串 // ——契约信封约定,绑定关系在载荷 albumIds),经生产 OutboxSender 真实 HTTP // 推送到测试内起的 ov 入口,断言 ads_ov 副本与 sync_state 版本账本——与生产 // 链路逐字节同源(embed.e2e-spec 模式:AppOvModule + drainSender)。 // // 覆盖(5 用例,链条递进共享同一 adId): // ① ad.updated 快照(enabled=true, albumIds=null)→ 副本行全字段 + 版本 // 账本就位; // ② v2(enabled=false)→ 删行(disabled=公开不可见=无投影,设计 §2.2/§4); // ③ v3(enabled=true, albumIds 数组)→ 同 adId upsert 复活(albumIds 整体 // 替换); // ④ 旧版本重放(v2 迟到于 v3 后,新 eventId)→ 行保持 v3(版本单调不回滚 // ——迟到的 disabled 快照不能删掉现行行); // ⑤ ad.deleted 墓碑 v4(high 优先级)→ 删行,账本收敛 v4。 // ④ 在 ⑤ 之前执行:stale 重放的断言基线是「行处于 v3 态」,墓碑(v4)之后 // 无法回到该基线(此后 v3 亦为 stale)。 import { NestFactory } from '@nestjs/core'; import { type NestExpressApplication } from '@nestjs/platform-express'; import { PrismaClient } from '@photo/db-cn'; import type { Prisma } from '@photo/db-cn'; import { PrismaClient as OvPrismaClient } from '@photo/db-ov'; import type { AdDeletedPayload, AdUpdatedPayload } from '@photo/contracts'; import { AppCnModule } from '../src/app-cn.module'; import { AppOvModule } from '../src/app-ov.module'; import { configureApp } from '../src/shared/configure-app'; import { ContractWsAdapter } from '../src/realtime/contract-ws.adapter'; import { resetDb } from './reset'; import { resetOvDb } from './reset-ov'; import { OutboxSender } from '../../sync/src/sender'; jest.setTimeout(120_000); const SYNC_KEY = process.env.SYNC_INTERNAL_KEY ?? 'dev-sync-secret-change-me'; /** 本 spec 全程操作的单个广告位 ID(链条用例共享;VarChar(32) 内)。 */ const AD_ID = 'ads-e2e-0001'; /** cn 原值创建时刻(ISO UTC)——排序 tie-break 位,断言原值回传。 */ const AD_CREATED_AT = '2026-09-28T08:00:00.000Z'; /** 副本行断言所需的最小结构(prismaOv.ads_ov 行的形状子集)。 */ interface AdsOvRow { adId: string; imageObjectKey: string; linkUrl: string; sortOrder: number; albumIds: unknown; entityVersion: number; createdAt: Date; updatedAt: Date; } describe('ads sync projection (e2e)', () => { let cnApp: NestExpressApplication; let ovApp: NestExpressApplication; let ovBase: string; let prismaCn: PrismaClient; let prismaOv: OvPrismaClient; let sender: OutboxSender; // outbox 行 occurredAt 基准 + 单调序号:保证发送器取批顺序(occurredAt // 升序)与用例推进顺序一致——④ 的迟到 v2 必须晚于 ③ 的 v3 被推送。 const occurredBase = Date.now() - 60_000; let seq = 0; /** ③ 落位的 v3 快照(④ stale 断言的基线)。 */ let v3Snapshot: AdUpdatedPayload; /** 排空发送器:循环 runOnce 直至无未投递事件(embed.e2e 同款实现)。 */ 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 仍有未投递事件'); } /** * 手工构造 outbox 行(绕过 CRUD——Task 7 才有):albumId 恒空串(ad.* 信封 * 约定);occurredAt 按序号递增保证批内顺序。返回 eventId(outbox 主键, * ④ 断言 stale 行 appliedAt 用)。 */ async function enqueueAdEvent( type: 'ad.updated' | 'ad.deleted', entityVersion: number, payload: AdUpdatedPayload | AdDeletedPayload, priority: 'normal' | 'high' = 'normal', ): Promise { const row = await prismaCn.outbox_events.create({ data: { type, albumId: '', entityVersion, priority, payload: payload as unknown as Prisma.InputJsonValue, occurredAt: new Date(occurredBase + seq * 1000), }, select: { id: true }, }); seq += 1; return row.id; } /** ad.updated 全量快照载荷(版本相关字段随 v 演进,可覆盖)。 */ function adSnapshot( v: number, overrides: Partial = {}, ): AdUpdatedPayload { return { adId: AD_ID, imageObjectKey: `photo-live/dev/ads/${AD_ID}/v${v}.jpg`, linkUrl: `https://example.com/ads?v=${v}`, enabled: true, sortOrder: v * 10, albumIds: null, createdAt: AD_CREATED_AT, entityVersion: v, ...overrides, }; } /** 断言副本行 === 快照全字段(updatedAt 为 ov 应用时刻,仅断言存在)。 */ function expectReplica(row: AdsOvRow | null, p: AdUpdatedPayload): void { expect(row).toMatchObject({ adId: AD_ID, imageObjectKey: p.imageObjectKey, linkUrl: p.linkUrl, sortOrder: p.sortOrder, albumIds: p.albumIds, entityVersion: p.entityVersion, createdAt: new Date(AD_CREATED_AT), updatedAt: expect.any(Date), }); } /** 断言版本账本收敛在指定版本。 */ async function expectAppliedVersion(v: number): Promise { const state = await prismaOv.sync_state.findUnique({ where: { entityType_entityId: { entityType: 'ad', entityId: AD_ID } }, }); expect(state?.appliedVersion).toBe(v); } beforeAll(async () => { expect(SYNC_KEY).not.toBe(''); await resetDb(); await resetOvDb(); // cn 入口仅作骨架就位(本 spec 不走 cn HTTP;outbox 行直插 cn 库)。 cnApp = await NestFactory.create(AppCnModule, { rawBody: true, }); configureApp(cnApp); await cnApp.listen(0); 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(); sender = new OutboxSender({ prisma: prismaCn, ovApiUrl: ovBase, internalKey: SYNC_KEY, }); }); afterAll(async () => { await cnApp.close(); await ovApp.close(); await prismaCn.$disconnect(); await prismaOv.$disconnect(); }); it('① ad.updated 快照(enabled=true, albumIds=null)→ 副本行 + 版本账本就位', async () => { const v1 = adSnapshot(1); await enqueueAdEvent('ad.updated', 1, v1); await drainSender(); const replica = await prismaOv.ads_ov.findUnique({ where: { adId: AD_ID }, }); expectReplica(replica, v1); // albumIds null(全局投放)→ SQL NULL,非 JSON null。 expect(replica?.albumIds).toBeNull(); const state = await prismaOv.sync_state.findUnique({ where: { entityType_entityId: { entityType: 'ad', entityId: AD_ID } }, }); expect(state).toMatchObject({ entityType: 'ad', entityId: AD_ID, appliedVersion: 1, watermark: false, }); }); it('② ad.updated v2(enabled=false)→ 行删除(disabled=无投影)', async () => { await enqueueAdEvent('ad.updated', 2, adSnapshot(2, { enabled: false })); await drainSender(); expect( await prismaOv.ads_ov.findUnique({ where: { adId: AD_ID } }), ).toBeNull(); await expectAppliedVersion(2); }); it('③ ad.updated v3(enabled=true, albumIds 数组)→ 行复活(同 adId upsert,albumIds 整体替换)', async () => { v3Snapshot = adSnapshot(3, { albumIds: ['ads-e2e-alb-1', 'ads-e2e-alb-2'], sortOrder: 35, linkUrl: 'https://example.com/ads?promo=v3', }); await enqueueAdEvent('ad.updated', 3, v3Snapshot); await drainSender(); const replica = await prismaOv.ads_ov.findUnique({ where: { adId: AD_ID }, }); expectReplica(replica, v3Snapshot); // 定向投放数组逐元素回传(JSON 列整体替换)。 expect(replica?.albumIds).toEqual(['ads-e2e-alb-1', 'ads-e2e-alb-2']); await expectAppliedVersion(3); }); it('④ 旧版本重放(v2 迟到于 v3 后,新 eventId)→ 行保持 v3(版本单调不回滚)', async () => { const staleEventId = await enqueueAdEvent( 'ad.updated', 2, adSnapshot(2, { enabled: false }), ); await drainSender(); // 行保持 v3 态:迟到的 disabled 快照(若被应用会删行)未生效。 const replica = await prismaOv.ads_ov.findUnique({ where: { adId: AD_ID }, }); expectReplica(replica, v3Snapshot); expect(replica?.albumIds).toEqual(['ads-e2e-alb-1', 'ads-e2e-alb-2']); await expectAppliedVersion(3); // stale 事件裁决即终(appliedAt 置位、不入积压)。 const staleRow = await prismaOv.sync_inbox.findUnique({ where: { eventId: staleEventId }, }); expect(staleRow?.appliedAt).not.toBeNull(); }); it('⑤ ad.deleted 墓碑 v4(high)→ 行删除,账本收敛 v4', async () => { await enqueueAdEvent( 'ad.deleted', 4, { adId: AD_ID, entityVersion: 4 }, 'high', ); await drainSender(); expect( await prismaOv.ads_ov.findUnique({ where: { adId: AD_ID } }), ).toBeNull(); await expectAppliedVersion(4); }); });