diff --git a/packages/db-ov/prisma/schema.prisma b/packages/db-ov/prisma/schema.prisma index e39d45a..c79f048 100644 --- a/packages/db-ov/prisma/schema.prisma +++ b/packages/db-ov/prisma/schema.prisma @@ -88,7 +88,7 @@ model photos_ov { // 缺失暂存 / 待补应用 / 待对账),receivedAt 供积压监控与重试排序。 model sync_inbox { eventId String @id - entityType String @db.VarChar(32) // album|category|photo|asset|slot-state + entityType String @db.VarChar(32) // album|category|photo|asset|slot-state|ad entityId String @db.VarChar(64) albumId String? @db.VarChar(64) entityVersion Int @@ -185,7 +185,8 @@ model embed_slots_ov { // —— 副本域:ads_ov(cn ads 的展示投影,2026-09-28)—— // 无墓碑——ad.deleted 墓碑与 enabled=false 快照都表达为「删行」(公开不可见); -// createdAt 为 cn 原值(排序 tie-break,非本机生成)。 +// createdAt 为 cn 原值(排序 tie-break,非本机生成);updatedAt 为 ov 应用时刻 +// (载荷无该字段,非 cn 原值——迁移头注释此列过度声明,以本注释为准)。 model ads_ov { adId String @id @db.VarChar(32) imageObjectKey String @db.VarChar(512) diff --git a/services/api/src/sync-inbox/dto.ts b/services/api/src/sync-inbox/dto.ts index e4fb2a2..9fc955f 100644 --- a/services/api/src/sync-inbox/dto.ts +++ b/services/api/src/sync-inbox/dto.ts @@ -27,6 +27,7 @@ export const SYNC_ENTITY_TYPES = [ 'photo', 'asset', 'slot-state', + 'ad', ] as const; /** 契约 SyncPriority。 */ diff --git a/services/api/src/sync-inbox/sync-inbox.service.ts b/services/api/src/sync-inbox/sync-inbox.service.ts index d92ea79..c326bb7 100644 --- a/services/api/src/sync-inbox/sync-inbox.service.ts +++ b/services/api/src/sync-inbox/sync-inbox.service.ts @@ -11,7 +11,8 @@ // (同事件重投递走 ① 的 eventId 幂等锚,根本到不了版本比较); // ③ 依赖缺失暂存:非 album 事件在 albums_ov 无所属相册行时保持 appliedAt // null(暂存不 applied,不投影);album 事件应用后按 albumId 回扫暂存 -// 队列补应用(相册到达即解锁该相册全部暂存事件); +// 队列补应用(相册到达即解锁该相册全部暂存事件);slot-state/ad 豁免 +// (albumPublicToken 弱关联 / 无相册依赖,见 decideAndApply ③ 注); // ④ 应用投影(版本不低于已应用):album → albums_ov upsert(不触碰 // bannerAssetKey/emergencyDisabled——分别归 asset.updated / 紧急停用通道 // Task 5;categories 仅当载荷携带全量清单(bootstrap/reconcile 快照事件) @@ -23,6 +24,8 @@ // 即撤下语义);asset → banner 投影 albums_ov.bannerAssetKey(cover/logo // 仅推进版本,ov schema 无位);slot-state(P4b Task 2)→ embed_slots_ov // 全量快照直投(albumPublicToken 弱关联相册、无依赖暂存,见 ③ 注); +// ad(广告位 2026-09-28)→ ads_ov 全量快照直投,墓碑或 enabled=false 均 +// 删行(disabled=公开不可见=无投影); // ⑤ sync_state upsert appliedVersion(watermark=false 实体行;水位行预留)。 // ⑥ 实时转发(Task 7):photo 事件**应用事务提交后**发布 OV_PHOTO_APPLIED // 进程内事件 → realtime 网关向相册房间广播契约 photo.* 消息(stale/ @@ -35,6 +38,7 @@ import { Injectable } from '@nestjs/common'; import { EventEmitter2 } from '@nestjs/event-emitter'; import type { components } from '@photo/contracts'; import type { + AdUpdatedPayload, AlbumUpdatedPayload, AssetUpdatedPayload, CategoryDeletedPayload, @@ -173,8 +177,14 @@ export class SyncInboxService { // ③ 依赖检查:非 album 事件须所属相册副本已存在,否则暂存。slot-state // 豁免:展示位副本以 albumPublicToken 弱关联相册(公开读时按 albums_ov // 现查,缺席=404 与相册下线同形),且解绑事件 albumId 为空串——相册行的 - // 到达顺序对副本正确性无影响,暂存反而会让解绑事件无限积压。 - if (event.entityType !== 'album' && event.entityType !== 'slot-state') { + // 到达顺序对副本正确性无影响,暂存反而会让解绑事件无限积压。ad 豁免: + // ad 无相册依赖(albumId 恒空串,绑定关系在载荷 albumIds),直接应用 + // ——不豁免则空串在 albums_ov 永远查不到,事件被永久暂存且无解锁路径。 + if ( + event.entityType !== 'album' && + event.entityType !== 'slot-state' && + event.entityType !== 'ad' + ) { const album = await tx.albums_ov.findUnique({ where: { albumId: event.albumId }, select: { albumId: true }, @@ -436,6 +446,40 @@ export class SyncInboxService { }); return undefined; } + // ad(广告位 2026-09-28):全量快照直投 ads_ov;墓碑或 enabled=false 快照 + // 均表达为删行(disabled=公开不可见=无投影,设计 §2.2/§4)。albumIds + // 整体替换;createdAt 取 cn 原值(排序 tie-break)。 + if (event.entityType === 'ad') { + const p = event.payload as unknown as AdUpdatedPayload; + if (event.tombstone === true || p.enabled === false) { + await tx.ads_ov.deleteMany({ where: { adId: event.entityId } }); + } else { + await tx.ads_ov.upsert({ + where: { adId: event.entityId }, + create: { + adId: event.entityId, + imageObjectKey: p.imageObjectKey, + linkUrl: p.linkUrl, + sortOrder: p.sortOrder, + albumIds: p.albumIds as unknown as Prisma.InputJsonValue, + entityVersion: event.entityVersion, + createdAt: new Date(p.createdAt), + // ov 应用时刻(载荷无 updatedAt——投影行该列非 cn 原值)。 + updatedAt: new Date(), + }, + update: { + imageObjectKey: p.imageObjectKey, + linkUrl: p.linkUrl, + sortOrder: p.sortOrder, + albumIds: p.albumIds as unknown as Prisma.InputJsonValue, + entityVersion: event.entityVersion, + createdAt: new Date(p.createdAt), + updatedAt: new Date(), + }, + }); + } + return undefined; + } return undefined; } diff --git a/services/api/test/ads.e2e-spec.ts b/services/api/test/ads.e2e-spec.ts new file mode 100644 index 0000000..821c9fb --- /dev/null +++ b/services/api/test/ads.e2e-spec.ts @@ -0,0 +1,271 @@ +// 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); + }); +}); diff --git a/services/api/test/reset-ov.ts b/services/api/test/reset-ov.ts index 6b3fbfd..b917b87 100644 --- a/services/api/test/reset-ov.ts +++ b/services/api/test/reset-ov.ts @@ -21,10 +21,11 @@ export const RESET_TABLES_OV: readonly string[] = [ // 同步账本(Task 4 inbox 消费)。 'sync_inbox', 'sync_state', - // 副本域(photos 先于 albums;展示位副本相册无关,随副本域一并清)。 + // 副本域(photos 先于 albums;展示位/广告位副本相册无关,随副本域一并清)。 'photos_ov', 'albums_ov', 'embed_slots_ov', + 'ads_ov', ]; /** TRUNCATE photo_ov 中 RESET_TABLES_OV 里**实际存在**的表(空库安全)。 */ diff --git a/services/sync/src/sender.ts b/services/sync/src/sender.ts index 34f937a..5f30d49 100644 --- a/services/sync/src/sender.ts +++ b/services/sync/src/sender.ts @@ -188,9 +188,10 @@ export class OutboxSender { /** * outbox 行 → SyncEvent 线格式。entityId 按实体类型取自不同位置: - * album → albumId 列;photo → photoId 列;category/asset/slot-state → 载荷内 - * 的 categoryId/assetId/slotId(outbox 表无此三列)。映射缺料抛错(fail-fast: - * 本轮不推、不标记,错误日志暴露;cn 侧发射器按构造保证不缺料)。 + * album → albumId 列;photo → photoId 列;category/asset/slot-state/ad → + * 载荷内的 categoryId/assetId/slotId/adId(outbox 表无此四列)。映射缺料 + * 抛错(fail-fast:本轮不推、不标记,错误日志暴露;cn 侧发射器按构造保证 + * 不缺料)。 */ private toWireEvent(row: OutboxRow): SyncEventWire { const type = row.type as OutboxEvent['type']; @@ -209,6 +210,11 @@ export class OutboxSender { throw new Error(`outbox 事件 ${row.id}(${type})载荷缺 slotId,无法推导 entityId`); } entityId = payload.slotId; + } else if (entityType === 'ad') { + if (typeof payload.adId !== 'string') { + throw new Error(`outbox 事件 ${row.id}(${type})载荷缺 adId,无法推导 entityId`); + } + entityId = payload.adId; } else if (typeof payload.categoryId === 'string') { entityId = payload.categoryId; } else if (typeof payload.assetId === 'string') {