// 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)。 // // Task 6 追加段(cn HTTP):/v1/admin/ads/uploads 上传初始化—— // ⑥ admin 201:staging 键形态 + 凭证真实可用(PUT 实写 MinIO 后 head 可核, // 对照 uploads.e2e 的 presign 断言写法;无会话概念=不落任何 DB 行); // ⑦ 非 admin 403;匿名 401(入参合法); // ⑧ 超过 AD_UPLOAD_MAX_BYTES(2MB)400(服务层,报文含限额值); // contentType 非白名单 400(DTO @IsIn);扩展名不接受 400(服务层)。 import { randomBytes } from 'node:crypto'; import { NestFactory } from '@nestjs/core'; import { type NestExpressApplication } from '@nestjs/platform-express'; import request from 'supertest'; import { PrismaClient } from '@photo/db-cn'; import type { Prisma } from '@photo/db-cn'; import { PrismaClient as OvPrismaClient } from '@photo/db-ov'; import { AD_UPLOAD_MAX_BYTES, type AdDeletedPayload, type AdUpdatedPayload, ErrorCodes, } from '@photo/contracts'; import { createStorageFromEnv, type StoragePort } from '@photo/storage'; import { AppCnModule } from '../src/app-cn.module'; import { AppOvModule } from '../src/app-ov.module'; import { seedAdmin } from '../src/scripts/seed-admin'; import { hashSessionToken } from '../src/auth/session.guard'; 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'; const ADMIN_ACCOUNT = process.env.ADMIN_ACCOUNT ?? 'admin'; const ADMIN_PASSWORD = process.env.ADMIN_PASSWORD ?? 'admin-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 cnBase: string; let ovBase: string; let prismaCn: PrismaClient; let prismaOv: OvPrismaClient; let sender: OutboxSender; let storage: StoragePort; let adminToken: string; let staffToken: string; // 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 入口仅作骨架就位(①-⑤ 不走 cn HTTP;outbox 行直插 cn 库)——Task 6 // 追加段起复用同一实例走 /v1/admin/ads/uploads。 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(); sender = new OutboxSender({ prisma: prismaCn, ovApiUrl: ovBase, internalKey: SYNC_KEY, }); storage = createStorageFromEnv(); // Task 6 追加段身份:admin(seedAdmin + 真实登录)与 staff 对照账号 // (直插 users + sessions,embed.e2e 同款);匿名 = 不带 Authorization。 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; const staff = await prismaCn.users.create({ data: { account: 'ads-staff', role: 'staff', passwordHash: 'x', displayName: 'ads-staff', }, select: { id: true }, }); const staffSession = randomBytes(32).toString('hex'); await prismaCn.sessions.create({ data: { userId: staff.id, tokenHash: hashSessionToken(staffSession), expiresAt: new Date(Date.now() + 3_600_000), }, }); staffToken = staffSession; }); 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); }); // ── Task 6 追加段:/v1/admin/ads/uploads 上传初始化(cn HTTP,admin-only)── it('⑥ admin 初始化 → 201 staging 键 + 凭证(真实 PUT 实写 MinIO,head 可核)', async () => { const res = await request(cnBase) .post('/v1/admin/ads/uploads') .set('Authorization', `Bearer ${adminToken}`) .send({ filename: 'ad-banner.jpeg', bytes: 256, contentType: 'image/jpeg' }); expect(res.status).toBe(201); // 响应形状精确断言(契约 AdUploadInitResponse;无会话概念=无其他字段)。 expect(Object.keys(res.body).sort()).toEqual(['credential', 'objectKey']); expect(Object.keys(res.body.credential).sort()).toEqual( ['expiresAt', 'headers', 'method', 'mode', 'url'], ); expect(res.body.credential).toMatchObject({ mode: 'presigned', method: 'PUT', headers: { 'Content-Type': 'image/jpeg' }, }); expect(new Date(res.body.credential.expiresAt).getTime()).toBeGreaterThan( Date.now(), ); // staging 键形态:photo-live/dev/ads/staging/{cuid 25 字符}.{ext};jpeg 归一 // 为 jpg(扩展名白名单映射,同上传域口径)。 expect(res.body.objectKey).toMatch( /^photo-live\/dev\/ads\/staging\/c[0-9a-z]{24}\.jpg$/, ); // 凭证真实可用:按凭证 PUT 后 head 可见精确字节数与类型(fetch,Node 22)。 const put = await fetch(res.body.credential.url, { method: res.body.credential.method, headers: res.body.credential.headers, body: new Uint8Array(Buffer.alloc(256, 7)), }); expect(put.ok).toBe(true); expect(await storage.head(res.body.objectKey)).toMatchObject({ bytes: 256, contentType: 'image/jpeg', }); }); it('⑦ 非 admin 403;匿名 401(入参合法)', async () => { const body = { filename: 'ad.png', bytes: 100, contentType: 'image/png' }; const staff = await request(cnBase) .post('/v1/admin/ads/uploads') .set('Authorization', `Bearer ${staffToken}`) .send(body); expect(staff.status).toBe(403); expect(staff.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN); const anon = await request(cnBase).post('/v1/admin/ads/uploads').send(body); expect(anon.status).toBe(401); }); it('⑧ 超过 2MB / contentType 非白名单 / 扩展名不接受 → 400 VALIDATION_FAILED', async () => { const post = (payload: object) => request(cnBase) .post('/v1/admin/ads/uploads') .set('Authorization', `Bearer ${adminToken}`) .send(payload); const over = await post({ filename: 'big.jpg', bytes: AD_UPLOAD_MAX_BYTES + 1, contentType: 'image/jpeg', }); expect(over.status).toBe(400); expect(over.body.code).toBe(ErrorCodes.VALIDATION_FAILED); // 服务层限额报文含限额值(DTO 不设 @Max,限额唯一裁决点在服务层)。 expect(over.body.message).toContain(String(AD_UPLOAD_MAX_BYTES)); const badType = await post({ filename: 'a.gif', bytes: 100, contentType: 'image/gif', }); expect(badType.status).toBe(400); expect(badType.body.code).toBe(ErrorCodes.VALIDATION_FAILED); const badExt = await post({ filename: 'a.gif', bytes: 100, contentType: 'image/jpeg', }); expect(badExt.status).toBe(400); expect(badExt.body.code).toBe(ErrorCodes.VALIDATION_FAILED); const noExt = await post({ filename: 'noext', bytes: 100, contentType: 'image/jpeg', }); expect(noExt.status).toBe(400); expect(noExt.body.code).toBe(ErrorCodes.VALIDATION_FAILED); }); });