From 955bea49b5a942b8071312e91d78db3dc7655410 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 18 Sep 2026 18:45:43 +0800 Subject: [PATCH] feat(sync): stat snapshot return path with version replacement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2 Task 9 统计回传（ov→cn）： - cn 迁移：albums 增 viewCountReplica/statsVersion/statsSyncedAt、 photos 增 statsVersion（like_count 已有副本位；无 VARCHAR 新列不涉校对手决） - cn 接收端 POST /v1/internal/sync/stat-snapshots（InternalSyncGuard）： 整批单事务，逐快照 updateMany WHERE statsVersion < statVersion —— 仅更高 statVersion 以权威 total 整体替换副本（绝不相加），重复/乱序幂等忽略； accepted 只数实际应用条数；views 应用同时推进 statsSyncedAt（A27 展示） - ov 回传发送器 services/sync/stat-return.ts：聚合 outbox_stat_events 未发送 行按（kind, photoId|albumId）取 MAX(statVersion) 折叠突发 → HMAC 推送 → 200 后按计数器标记截至该版本全部源行 sentAt；失败留待下轮重发；main.ts 第三循环 STAT_RETURN_INTERVAL_MS（默认 60s，env 可调） - 契约 AlbumSummary + statsSyncedAt（string|null readOnly，+gen）；listAlbums/ getAlbum 经 toSummaryResponse 暴露（管理 UI 列留 P4a） - e2e（services/sync stat-return.e2e-spec，7 例）：逐次点赞回传替换式收敛 （2→5→10 非累加）、浏览窗口去重回传 + statsSyncedAt、重复回传 accepted=0 值不动（A27 不重复累计）、乱序高版本替换（total 可下调）+ 低版本忽略、 发送端折叠只发最高版本且被折叠行随批确认、管理列表 likeCount/statsSyncedAt 展示、新端点 401/400 守卫 Co-Authored-By: Claude Code --- .env.example | 4 +- packages/contracts/openapi.yaml | 9 + packages/contracts/src/generated/openapi.d.ts | 7 + .../migration.sql | 12 + packages/db-cn/prisma/schema.prisma | 11 +- services/api/src/albums/albums.service.ts | 3 + services/api/src/sync/dto.ts | 58 ++- services/api/src/sync/sync.controller.ts | 15 +- services/api/src/sync/sync.service.ts | 54 ++ services/api/test/albums.e2e-spec.ts | 1 + services/sync/src/main.ts | 62 ++- services/sync/src/stat-return.ts | 241 +++++++++ services/sync/test/stat-return.e2e-spec.ts | 473 ++++++++++++++++++ 13 files changed, 933 insertions(+), 17 deletions(-) create mode 100644 packages/db-cn/prisma/migrations/20260918102542_stat_replica_columns/migration.sql create mode 100644 services/sync/src/stat-return.ts create mode 100644 services/sync/test/stat-return.e2e-spec.ts diff --git a/.env.example b/.env.example index d89d342..a7ef3e7 100644 --- a/.env.example +++ b/.env.example @@ -34,8 +34,8 @@ CN_API_URL=http://127.0.0.1:8080 # 以下为 P2 决策初值（D07/D11）的 env 覆盖键：默认值见 # packages/contracts/src/constants.ts（HEARTBEAT_MS 见 src/realtime.ts）， # 此处登记供运维可见与调整；SYNC_BATCH_SIZE/SYNC_PUSH_INTERVAL_MS 自 Task 3 -# 起、SYNC_RECONCILE_INTERVAL_MS 自 Task 4 起被 services/sync 读取，其余为 -# 前瞻登记（Task 7+ 逐个接线）。 +# 起、SYNC_RECONCILE_INTERVAL_MS 自 Task 4 起、STAT_RETURN_INTERVAL_MS 自 +# Task 9 起被 services/sync 读取，其余为前瞻登记（Task 7+ 逐个接线）。 SYNC_BATCH_SIZE=100 SYNC_PUSH_INTERVAL_MS=2000 SYNC_RECONCILE_INTERVAL_MS=300000 diff --git a/packages/contracts/openapi.yaml b/packages/contracts/openapi.yaml index fa59e35..863edfd 100644 --- a/packages/contracts/openapi.yaml +++ b/packages/contracts/openapi.yaml @@ -1790,6 +1790,7 @@ components: - translations - createdAt - updatedAt + - statsSyncedAt properties: id: type: string @@ -1858,6 +1859,14 @@ components: format: date-time readOnly: true description: 最近更新时间。 + statsSyncedAt: + type: [string, 'null'] + format: date-time + readOnly: true + description: | + 统计副本最近同步时间（P2 A27：ov→cn 统计快照在本相册的应用时刻）； + null 表示尚无回传应用（viewCountReplica 副本位仍为初始 0）。照片级 + 点赞副本（PhotoRecord.likeCount）的同步时间不经相册行表达。 AlbumList: allOf: diff --git a/packages/contracts/src/generated/openapi.d.ts b/packages/contracts/src/generated/openapi.d.ts index 77f28b6..7fa7d3d 100644 --- a/packages/contracts/src/generated/openapi.d.ts +++ b/packages/contracts/src/generated/openapi.d.ts @@ -908,6 +908,13 @@ export interface components { * @description 最近更新时间。 */ readonly updatedAt: string; + /** + * Format: date-time + * @description 统计副本最近同步时间（P2 A27：ov→cn 统计快照在本相册的应用时刻）； + * null 表示尚无回传应用（viewCountReplica 副本位仍为初始 0）。照片级 + * 点赞副本（PhotoRecord.likeCount）的同步时间不经相册行表达。 + */ + readonly statsSyncedAt: string | null; }; AlbumList: components["schemas"]["CursorPageMeta"] & { items: components["schemas"]["AlbumSummary"][]; diff --git a/packages/db-cn/prisma/migrations/20260918102542_stat_replica_columns/migration.sql b/packages/db-cn/prisma/migrations/20260918102542_stat_replica_columns/migration.sql new file mode 100644 index 0000000..b396200 --- /dev/null +++ b/packages/db-cn/prisma/migrations/20260918102542_stat_replica_columns/migration.sql @@ -0,0 +1,12 @@ +-- P2 Task 9：cn 侧统计回传副本列（ov→cn stat-snapshots 接收位）。 +-- albums.viewCountReplica/statsVersion/statsSyncedAt：相册浏览数替换式副本 + +-- 回传替换版本（仅更高 statVersion 替换，绝不相加）+ 最近应用时刻（A27 展示 +-- 同步时间）；photos.statsVersion：likeCount 副本的替换版本比较位 +-- （photos.like_count 已存在，P1 起即为副本预留位）。全列带默认值，存量行 +-- （副本未同步态 = 0/NULL）与 P1 流程不受影响。无 VARCHAR 新列，不涉校对手决。 +ALTER TABLE `albums` ADD COLUMN `statsSyncedAt` DATETIME(6) NULL, + ADD COLUMN `statsVersion` INTEGER NOT NULL DEFAULT 0, + ADD COLUMN `viewCountReplica` INTEGER NOT NULL DEFAULT 0; + +-- AlterTable +ALTER TABLE `photos` ADD COLUMN `statsVersion` INTEGER NOT NULL DEFAULT 0; diff --git a/packages/db-cn/prisma/schema.prisma b/packages/db-cn/prisma/schema.prisma index 21b11b9..657c348 100644 --- a/packages/db-cn/prisma/schema.prisma +++ b/packages/db-cn/prisma/schema.prisma @@ -67,6 +67,12 @@ model albums { originalDownloadEnabled Boolean @default(false) watermarkVersion Int @default(0) entityVersion Int @default(1) // P2 Task 3：album.updated 事件版本（同 photos 模式） + // P2 Task 9 统计回传副本位（ov→cn）：viewCountReplica 为 ov album_stats + // 权威值的替换式副本；statsVersion 是回传替换版本（仅更高 statVersion 替换， + // 绝不相加）；statsSyncedAt 记最近一次**应用**时刻（A27 同步时间展示）。 + viewCountReplica Int @default(0) + statsVersion Int @default(0) + statsSyncedAt DateTime? @db.DateTime(6) createdAt DateTime @default(now()) @db.DateTime(6) updatedAt DateTime @updatedAt @db.DateTime(6) @@ -153,7 +159,10 @@ model photos { processingState String @default("pending_upload") // pending_upload|uploaded|processing|ready|processing_failed errorCode String? @db.VarChar(64) visibility String @default("unpublished") // unpublished|published|hidden|deleted - likeCount Int @default(0) // P2 起为统计副本 + likeCount Int @default(0) // P2 起为统计副本（ov likeTotal 替换式回传） + // P2 Task 9：回传替换版本（ov photos_ov.countVersion 的 cn 侧镜像比较位， + // 仅更高 statVersion 替换 likeCount，重复/乱序回传幂等）。 + statsVersion Int @default(0) entityVersion Int @default(1) @@index([albumId, visibility, publishedAt, id]) @@index([albumId, uploaderId, uploadedAt]) diff --git a/services/api/src/albums/albums.service.ts b/services/api/src/albums/albums.service.ts index d238c52..fb68560 100644 --- a/services/api/src/albums/albums.service.ts +++ b/services/api/src/albums/albums.service.ts @@ -111,6 +111,9 @@ function toSummaryResponse( translations: translations.map(toTranslationResponse), createdAt: album.createdAt.toISOString(), updatedAt: album.updatedAt.toISOString(), + // P2 Task 9（A27 同步时间展示）：统计副本最近回传应用时刻；null=尚无回传。 + statsSyncedAt: + album.statsSyncedAt === null ? null : album.statsSyncedAt.toISOString(), }; } diff --git a/services/api/src/sync/dto.ts b/services/api/src/sync/dto.ts index 42bb988..2f4d187 100644 --- a/services/api/src/sync/dto.ts +++ b/services/api/src/sync/dto.ts @@ -1,14 +1,21 @@ // services/api/src/sync/dto.ts —— 内部同步端点请求 DTO（P2 Task 3 -// bootstrap/reconcile；Task 4 增 digest）。 -// 契约权威：openapi.yaml [internal]（syncBootstrap/syncReconcile/syncDigest）。 -// 经全局 ValidationPipe（whitelist）剥离未知字段；校验失败 400 VALIDATION_FAILED。 +// bootstrap/reconcile；Task 4 增 digest；Task 9 增 stat-snapshots）。 +// 契约权威：openapi.yaml [internal]（syncBootstrap/syncReconcile/syncDigest/ +// syncStatSnapshots）。经全局 ValidationPipe（whitelist）剥离未知字段；校验 +// 失败 400 VALIDATION_FAILED。 +import { Type } from 'class-transformer'; import { + ArrayMaxSize, + ArrayMinSize, + IsDateString, IsIn, IsInt, IsOptional, IsString, Max, Min, + ValidateIf, + ValidateNested, } from 'class-validator'; /** 契约 SyncEntityType（slot-state 预留——与 openapi enum 同步）。 */ @@ -52,3 +59,48 @@ export class DigestRequestDto { @IsString() albumId?: string; } + +/** 契约 SyncStatKind（与 openapi enum 同步）。 */ +export const SYNC_STAT_KINDS = ['likes', 'views'] as const; + +/** + * 契约 StatSnapshot（Task 9 ov→cn 统计快照单元）。kind=likes 时 photoId 必填 + * （ValidateIf 条件校验，违反 → 400）；kind=views 时 photoId 即使携带也仅作 + * 透传字段，接收端按 albumId 应用（相册级快照无照片位）。 + */ +export class StatSnapshotDto { + // 仅 kind=likes 时校验（条件为 true 时 undefined 也过不了 IsString → 400， + // 即契约「likes 时 photoId 必填」）；kind=views 时跳过、接收端忽略该字段。 + @ValidateIf((o: StatSnapshotDto) => o.kind === 'likes') + @IsString() + photoId?: string; + + @IsString() + albumId!: string; + + @IsIn(SYNC_STAT_KINDS) + kind!: (typeof SYNC_STAT_KINDS)[number]; + + @IsInt() + @Min(0) + total!: number; + + @IsInt() + @Min(1) + statVersion!: number; + + @IsString() + source!: string; + + @IsDateString() + confirmedAt!: string; +} + +/** 契约 StatSnapshotRequest（Task 9）：批次上限对齐 SYNC_BATCH_SIZE 量级。 */ +export class StatSnapshotRequestDto { + @ArrayMinSize(1) + @ArrayMaxSize(100) + @ValidateNested({ each: true }) + @Type(() => StatSnapshotDto) + snapshots!: StatSnapshotDto[]; +} diff --git a/services/api/src/sync/sync.controller.ts b/services/api/src/sync/sync.controller.ts index 8138ec7..cb76f34 100644 --- a/services/api/src/sync/sync.controller.ts +++ b/services/api/src/sync/sync.controller.ts @@ -1,7 +1,8 @@ // services/api/src/sync/sync.controller.ts —— cn 侧内部同步端点（P2 Task 3： -// bootstrap/reconcile；Task 4 增 digest）。契约权威：openapi.yaml [internal] -// （POST /internal/sync/bootstrap、POST /internal/sync/reconcile、 -// POST /internal/sync/digest）。鉴权：InternalSyncGuard（HMAC 原始请求体 +// bootstrap/reconcile；Task 4 增 digest；Task 9 增 stat-snapshots）。契约权威： +// openapi.yaml [internal]（POST /internal/sync/bootstrap、 +// POST /internal/sync/reconcile、POST /internal/sync/digest、 +// POST /internal/sync/stat-snapshots）。鉴权：InternalSyncGuard（HMAC 原始请求体 // 签名 + 时间戳 ±300s，密钥 SYNC_INTERNAL_KEY 仅 env）。仅 cn 入口装配 // （app-cn.module）；ov 入口的 content-batches 在 sync-inbox 模块（同守卫）。 import { Body, Controller, HttpCode, Post, UseGuards } from '@nestjs/common'; @@ -10,6 +11,7 @@ import { BootstrapRequestDto, DigestRequestDto, ReconcileRequestDto, + StatSnapshotRequestDto, } from './dto'; import { SyncService } from './sync.service'; @@ -38,4 +40,11 @@ export class SyncController { digest(@Body() dto: DigestRequestDto) { return this.sync.digest(dto.albumId); } + + /** 统计快照回传（Task 9 ov→cn）：版本替换式副本更新（见 service 设计说明）。 */ + @Post('stat-snapshots') + @HttpCode(200) + statSnapshots(@Body() dto: StatSnapshotRequestDto) { + return this.sync.receiveStatSnapshots(dto.snapshots); + } } diff --git a/services/api/src/sync/sync.service.ts b/services/api/src/sync/sync.service.ts index 9008825..31c1d6f 100644 --- a/services/api/src/sync/sync.service.ts +++ b/services/api/src/sync/sync.service.ts @@ -30,6 +30,7 @@ import { Injectable } from '@nestjs/common'; import type { components } from '@photo/contracts'; import type { Prisma } from '@photo/db-cn'; import { PrismaService } from '../prisma/prisma.service'; +import type { StatSnapshotDto } from './dto'; import { type AlbumCategoryInput, buildAlbumUpdatedPayload, @@ -41,6 +42,7 @@ import { export type BootstrapResponse = components['schemas']['BootstrapResponse']; export type ReconcileResponse = components['schemas']['ReconcileResponse']; export type DigestResponse = components['schemas']['DigestResponse']; +export type StatSnapshotResponse = components['schemas']['StatSnapshotResponse']; export type SyncEventWire = components['schemas']['SyncEvent']; /** 线格式 schema 版本（契约 SyncEvent.schemaVersion 当前为 1）。 */ @@ -502,6 +504,58 @@ export class SyncService { }; } + /** + * 统计快照回传接收（P2 Task 9，ov→cn）：整批单事务，逐快照做**版本替换**—— + * 仅当 cn 侧 statsVersion < snapshot.statVersion 时以权威 total 整体替换副本 + * （绝不相加），重复回传（同版本）与乱序（更低版本）均幂等忽略。这是 A27 + * 「不重复累计」的接收端锚点：total 恒为 ov 权威快照值（非增量），即使未来 + * 出现下调修正（total 变小），更高版本也照常替换——版本是唯一权威。 + * + * 副本位：likes → photos.likeCount（比较位 photos.statsVersion）； + * views → albums.viewCountReplica（比较位 albums.statsVersion， + * 同时推进 statsSyncedAt=本批应用时刻，A27 展示同步时间；likes 快照的 + * photoId 在 cn 必然存在（照片源于 cn），不存在/更旧版本 → 该条不计入 + * accepted，不报错——与「旧版本忽略不计」同一幂等口径）。 + * accepted 只数**实际应用**（新版本生效）的条数（契约：旧版本被忽略的不计）。 + */ + async receiveStatSnapshots( + snapshots: StatSnapshotDto[], + ): Promise { + const syncedAt = new Date(); + return this.prisma.$transaction(async (tx) => { + let accepted = 0; + for (const snapshot of snapshots) { + if (snapshot.kind === 'likes' && snapshot.photoId !== undefined) { + const applied = await tx.photos.updateMany({ + where: { + id: snapshot.photoId, + statsVersion: { lt: snapshot.statVersion }, + }, + data: { + likeCount: snapshot.total, + statsVersion: snapshot.statVersion, + }, + }); + accepted += applied.count; + } else if (snapshot.kind === 'views') { + const applied = await tx.albums.updateMany({ + where: { + id: snapshot.albumId, + statsVersion: { lt: snapshot.statVersion }, + }, + data: { + viewCountReplica: snapshot.total, + statsVersion: snapshot.statVersion, + statsSyncedAt: syncedAt, + }, + }); + accepted += applied.count; + } + } + return { accepted }; + }); + } + /** 相册 id 集合 → 分类快照输入（分类行 + 翻译行配对；升序确定性由构建器排）。 */ private async categoryInputsByAlbumIds( albumIds: string[], diff --git a/services/api/test/albums.e2e-spec.ts b/services/api/test/albums.e2e-spec.ts index b5b97b4..537ab2a 100644 --- a/services/api/test/albums.e2e-spec.ts +++ b/services/api/test/albums.e2e-spec.ts @@ -37,6 +37,7 @@ const ALBUM_SUMMARY_KEYS = [ 'publicEnabled', 'publicToken', 'startAt', + 'statsSyncedAt', 'status', 'timezone', 'translations', diff --git a/services/sync/src/main.ts b/services/sync/src/main.ts index 6b0181e..8b9ddce 100644 --- a/services/sync/src/main.ts +++ b/services/sync/src/main.ts @@ -1,21 +1,27 @@ // services/sync/src/main.ts —— sync 进程入口（P2 Task 3 发送器 + Task 4 对账 -// 调度）。构建产物 dist/main.js 由根 dev:sync / 生产启动命令运行；env 一律来自 -// 进程环境（根 dotenv-cli 注入），此处不读文件。 +// 调度 + Task 9 统计回传）。构建产物 dist/main.js 由根 dev:sync / 生产启动 +// 命令运行；env 一律来自进程环境（根 dotenv-cli 注入），此处不读文件。 // -// 两个独立循环： +// 三个独立循环： // · 发送循环：每 SYNC_PUSH_INTERVAL_MS 轮询 outbox（SYNC_BATCH_SIZE 一批， // high 优先——见 sender.ts 排序说明；满批立即追下一轮，积压排空不被轮询 // 间隔拖慢）。失败退避：连续失败按间隔 ×2^n 指数递增、60s 封顶，成功归零。 // · 对账循环（Task 4）：每 SYNC_RECONCILE_INTERVAL_MS 一轮 SyncReconciler. // runOnce（当前态清扫重放 + digest 比对触发 bootstrap，见 reconciler.ts）。 // 失败仅记日志、下轮再试（分钟级节奏，无需退避——幂等安全）。 +// · 统计回传循环（Task 9）：每 STAT_RETURN_INTERVAL_MS 一轮 StatReturnSender. +// runReturnOnce（ov outbox_stat_events 聚合 → cn stat-snapshots 版本替换， +// 见 stat-return.ts）。失败语义同对账：留待下轮重试（cn 侧版本比较保证 +// 幂等），分钟级节奏不另设退避。 // -// 双库连接：cn 库（outbox 取数）+ ov 库（副本计数直读）。优雅退出： -// SIGINT/SIGTERM 停两循环（等待在飞一轮，fetch 超时已封顶 30s）后断开两连接。 +// 双库连接：cn 库（outbox 取数）+ ov 库（副本计数直读/回传队列取数）。 +// 优雅退出：SIGINT/SIGTERM 停全部循环（等待在飞一轮，fetch 超时已封顶 30s） +// 后断开两连接。 import { pino } from 'pino'; import { PrismaClient } from '@photo/db-cn'; import { PrismaClient as OvPrismaClient } from '@photo/db-ov'; import { + STAT_RETURN_INTERVAL_MS, SYNC_BATCH_SIZE, SYNC_INTERNAL_KEY_ENV, SYNC_PUSH_INTERVAL_MS, @@ -23,6 +29,7 @@ import { } from '@photo/contracts'; import { OutboxSender } from './sender'; import { SyncReconciler } from './reconciler'; +import { StatReturnSender } from './stat-return'; /** 失败退避封顶（对齐 SYNC_RECONCILE_INTERVAL_MS 量级：分钟级封顶足够）。 */ const MAX_BACKOFF_MS = 60_000; @@ -75,6 +82,11 @@ async function main(): Promise { 'SYNC_RECONCILE_INTERVAL_MS', SYNC_RECONCILE_INTERVAL_MS, ); + const statReturnIntervalMs = readIntEnv( + logger, + 'STAT_RETURN_INTERVAL_MS', + STAT_RETURN_INTERVAL_MS, + ); const cnApiUrl = process.env.CN_API_URL ?? 'http://127.0.0.1:8080'; const ovApiUrl = process.env.OV_API_URL ?? 'http://127.0.0.1:8081'; const internalKey = process.env[SYNC_INTERNAL_KEY_ENV] ?? ''; @@ -102,9 +114,23 @@ async function main(): Promise { timeoutMs: REQUEST_TIMEOUT_MS, logger, }); + const statReturn = new StatReturnSender({ + prismaOv, + cnApiUrl, + internalKey, + timeoutMs: REQUEST_TIMEOUT_MS, + logger, + }); logger.info( - { cnApiUrl, ovApiUrl, intervalMs, batchSize, reconcileIntervalMs }, - 'sync sender + reconciler started', + { + cnApiUrl, + ovApiUrl, + intervalMs, + batchSize, + reconcileIntervalMs, + statReturnIntervalMs, + }, + 'sync sender + reconciler + stat-return started', ); let stopping = false; @@ -153,7 +179,27 @@ async function main(): Promise { await sleepStoppable(reconcileIntervalMs); } }; - const inFlight: Promise[] = [senderLoop(), reconcileLoop()]; + const statReturnLoop = async (): Promise => { + while (!stopping) { + try { + const stats = await statReturn.runReturnOnce(); + if (stats.picked > 0) { + logger.info(stats, 'stat snapshots returned'); + } + } catch (err) { + logger.error( + { err: err instanceof Error ? err.message : String(err) }, + 'stat return cycle failed（未确认行留待下轮重发，幂等安全）', + ); + } + await sleepStoppable(statReturnIntervalMs); + } + }; + const inFlight: Promise[] = [ + senderLoop(), + reconcileLoop(), + statReturnLoop(), + ]; const shutdown = (signal: string): void => { if (stopping) return; diff --git a/services/sync/src/stat-return.ts b/services/sync/src/stat-return.ts new file mode 100644 index 0000000..35b3c0c --- /dev/null +++ b/services/sync/src/stat-return.ts @@ -0,0 +1,241 @@ +// services/sync/src/stat-return.ts —— ov→cn 统计快照回传发送器（P2 Task 9）。 +// 单轮 runReturnOnce()：读 ov 库 outbox_stat_events 未发送行 → 按 +// （kind, photoId|albumId）聚合取 **MAX(statVersion)** 快照（点赞/浏览突发 +// 折叠为每计数器一条权威值——中间版本对 cn 无意义，cn 只做版本替换）→ +// HMAC 签名推送 cn 入口 POST /v1/internal/sync/stat-snapshots → 200 后把该 +// 计数器 **截至该版本** 的全部源行标 sentAt（含被折叠的低版本行——它们已被 +// 更高版本覆盖，重发永无可能改变 cn 状态）。 +// +// 失败语义（同 sender）：任一环节抛错则本轮已推送批的确认照常落库、未推送 +// 行留待下轮（sentAt 为空即重发）；cn 侧「仅更高 statVersion 替换」保证重试 +// 幂等——「cn 已应用、ov 未收到 200」的重复回传在 cn 全部判旧（accepted=0）， +// A27 不重复累计。cn 侧 like_count/viewCountReplica 是替换式副本，本发送器 +// 从不读 cn 库（数据面单向，与 reconciler 的 cn 经 HTTP 原则一致）。 +import type { PrismaClient } from '@photo/db-ov'; +import type { components } from '@photo/contracts'; +import { + SYNC_SIGNATURE_HEADER, + SYNC_TIMESTAMP_HEADER, + signSyncBody, +} from '@photo/contracts'; +import type { Logger } from 'pino'; + +/** openapi StatSnapshot（线格式）。 */ +export type StatSnapshotWire = components['schemas']['StatSnapshot']; + +/** stat-snapshots 响应（openapi StatSnapshotResponse）。 */ +interface StatSnapshotResponse { + accepted: number; +} + +/** outbox_stat_events 行（发送器只读关心的列）。 */ +type StatEventRow = { + id: string; + kind: string; + albumId: string; + photoId: string | null; + total: number; + statVersion: number; + createdAt: Date; +}; + +export interface StatReturnSenderOptions { + /** ov 库客户端（回传队列取数 + sentAt 回写）。 */ + prismaOv: PrismaClient; + /** cn 入口基址（如 http://127.0.0.1:8080，不含路径）；端点路径本模块拼接。 */ + cnApiUrl: string; + /** HMAC 密钥（与接收端 InternalSyncGuard 同一 SYNC_INTERNAL_KEY 值）。 */ + internalKey: string; + /** 单批快照上限（聚合后；默认 100，与 SYNC_BATCH_SIZE 量级对齐）。 */ + batchSize?: number; + /** 单次 HTTP 超时毫秒（默认 30s）。 */ + timeoutMs?: number; + logger?: Logger; +} + +/** 单轮 runReturnOnce 的观测口径（main 循环打点；测试用于断言折叠/幂等）。 */ +export interface StatReturnStats { + /** 取到的未发送源行数（聚合前）。 */ + picked: number; + /** 聚合后实际发送的快照条数（=Σ 各批快照数）。 */ + snapshots: number; + /** cn 实际应用（新版本生效）的快照条数；重复回传恒 0（A27 幂等锚）。 */ + accepted: number; +} + +const DEFAULT_BATCH_SIZE = 100; +const DEFAULT_TIMEOUT_MS = 30_000; + +/** cn 侧 stat-snapshots 端点路径（全局前缀 v1 + InternalSyncGuard 域）。 */ +const STAT_SNAPSHOTS_PATH = '/v1/internal/sync/stat-snapshots'; + +/** 快照来源标识（契约 StatSnapshot.source；本发送器唯一取值）。 */ +const SOURCE = 'ov'; + +/** 聚合键：likes 按照片、views 按相册（与 outbox_stat_events 写入口径一致）。 */ +function collapseKey(row: StatEventRow): string { + return row.kind === 'likes' + ? `likes:${row.photoId ?? ''}` + : `views:${row.albumId}`; +} + +export class StatReturnSender { + private readonly batchSize: number; + private readonly timeoutMs: number; + private readonly endpoint: string; + + constructor(private readonly options: StatReturnSenderOptions) { + this.batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE; + this.timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS; + this.endpoint = `${options.cnApiUrl.replace(/\/+$/, '')}${STAT_SNAPSHOTS_PATH}`; + } + + /** 单轮回传：取未发送行 → 聚合折叠 → 分批签名推送 → 按批回写 sentAt。 */ + async runReturnOnce(): Promise { + const rows = await this.options.prismaOv.outbox_stat_events.findMany({ + where: { sentAt: null }, + orderBy: [{ createdAt: 'asc' }, { id: 'asc' }], + select: { + id: true, + kind: true, + albumId: true, + photoId: true, + total: true, + statVersion: true, + createdAt: true, + }, + }); + if (rows.length === 0) { + return { picked: 0, snapshots: 0, accepted: 0 }; + } + + // 折叠突发：同计数器只发最高版本（total 恒为该版本权威快照值）。 + // statVersion 随每次点赞/计入单调 +1（photos_ov.countVersion / + // album_stats.statVersion），同键并列版本理论不出现；防御口径取「并列时 + // 保留后见行」（createdAt 序在后）——同版本同 total，无语义差。 + const collapsed = new Map(); + for (const row of rows) { + const key = collapseKey(row); + const existing = collapsed.get(key); + if ( + existing === undefined || + row.statVersion >= existing.statVersion + ) { + collapsed.set(key, row); + } + } + const latest = [...collapsed.values()]; + + let accepted = 0; + for (let offset = 0; offset < latest.length; offset += this.batchSize) { + const chunk = latest.slice(offset, offset + this.batchSize); + const response = await this.push(chunk); + accepted += response.accepted; + await this.markSent(chunk); + } + this.options.logger?.debug( + { picked: rows.length, snapshots: latest.length, accepted }, + 'stat snapshots returned', + ); + return { picked: rows.length, snapshots: latest.length, accepted }; + } + + /** + * 回写确认：对每个聚合后的快照，把该计数器**截至其版本**的未发送源行全部 + * 标 sentAt（含被折叠的低版本行）。取数与回写之间新落库的行版本必然更高 + * （单调自增），不会被误标；cn 200 = 整批已受理（应用或幂等判旧）。 + */ + private async markSent(chunk: StatEventRow[]): Promise { + const sentAt = new Date(); + for (const row of chunk) { + if (row.kind === 'likes' && row.photoId !== null) { + await this.options.prismaOv.outbox_stat_events.updateMany({ + where: { + kind: 'likes', + photoId: row.photoId, + statVersion: { lte: row.statVersion }, + sentAt: null, + }, + data: { sentAt }, + }); + } else if (row.kind === 'views') { + await this.options.prismaOv.outbox_stat_events.updateMany({ + where: { + kind: 'views', + albumId: row.albumId, + statVersion: { lte: row.statVersion }, + sentAt: null, + }, + data: { sentAt }, + }); + } + } + } + + /** + * 签名推送：同一 JSON.stringify 产物既作签名输入也作请求体（rawBody 验签 + * 字节一致，与 sender.push / reconciler.signedPost 同约定）。非 2xx / 响应 + * 形状异常均抛错（调用方留待下轮重试）。 + */ + private async push( + snapshots: StatEventRow[], + ): Promise { + const body = { + snapshots: snapshots.map((row): StatSnapshotWire => { + if (row.kind === 'likes' && row.photoId !== null) { + return { + photoId: row.photoId, + albumId: row.albumId, + kind: 'likes', + total: row.total, + statVersion: row.statVersion, + source: SOURCE, + confirmedAt: row.createdAt.toISOString(), + }; + } + return { + albumId: row.albumId, + kind: 'views', + total: row.total, + statVersion: row.statVersion, + source: SOURCE, + confirmedAt: row.createdAt.toISOString(), + }; + }), + }; + const rawBody = JSON.stringify(body); + const timestamp = Math.floor(Date.now() / 1000).toString(); + let response: Response; + try { + response = await fetch(this.endpoint, { + method: 'POST', + headers: { + 'content-type': 'application/json', + [SYNC_TIMESTAMP_HEADER]: timestamp, + [SYNC_SIGNATURE_HEADER]: signSyncBody(this.options.internalKey, rawBody), + }, + body: rawBody, + signal: AbortSignal.timeout(this.timeoutMs), + }); + } catch (err) { + throw new Error( + `stat-snapshots 请求失败（${this.endpoint}）：${err instanceof Error ? err.message : String(err)}`, + ); + } + if (!response.ok) { + const text = await response.text().catch(() => ''); + throw new Error( + `stat-snapshots 非 2xx（${this.endpoint}）：status=${response.status} body=${text.slice(0, 500)}`, + ); + } + const data: unknown = await response.json(); + if ( + typeof data !== 'object' || + data === null || + typeof (data as StatSnapshotResponse).accepted !== 'number' + ) { + throw new Error('stat-snapshots 响应形状异常（accepted 缺失）'); + } + return data as StatSnapshotResponse; + } +} diff --git a/services/sync/test/stat-return.e2e-spec.ts b/services/sync/test/stat-return.e2e-spec.ts new file mode 100644 index 0000000..b4078cd --- /dev/null +++ b/services/sync/test/stat-return.e2e-spec.ts @@ -0,0 +1,473 @@ +// services/sync/test/stat-return.e2e-spec.ts —— P2 Task 9 统计回传（ov→cn） +// E2E。复用 sync.e2e-spec.ts 的进程内双入口 harness（cn/ov 真实 HTTP + 真实 +// 双 MySQL + MinIO + HMAC 全链路字节一致），被测对象为生产 StatReturnSender +// （runReturnOnce 直驱，指向 cn 入口真实 URL）+ cn 侧 stat-snapshots 接收端。 +// +// 覆盖（A27 仿真级 + Task 9 全部工作项）： +// ① 点赞逐次回传：3 次点赞、每次回传一轮 → cn likeCount 逐次被**权威快照 +// 替换**（2→5→10），绝非增量累加（累加口径会是 2+5+10=17）；statsVersion +// 随 statVersion 推进；源行全部确认； +// ② 浏览回传：首访 counted + 窗口内刷新不计 → 恰 1 条 views 快照 → cn +// viewCountReplica=1、statsSyncedAt 落位（A27 展示同步时间）； +// ③ 重复回传幂等（A27 不重复累计）：清 sentAt 模拟「cn 已应用、ov 未收到 +// 200」重发 → 聚合折叠后同版本全判旧 accepted=0，cn 值分毫不动； +// ④ 乱序快照：接收端直推更高 statVersion 但更低 total（版本权威，可表达 +// 下调修正）→ 替换生效；更低 statVersion 旧快照 → accepted=0 忽略； +// 发送端聚合：同相册 v5/v3 两行未发送 → 只发 v5、两行（含被折叠的 v3） +// 全部确认； +// ⑤ cn 管理展示：listAlbumPhotos.likeCount = 副本现值；listAlbums 条目 +// statsSyncedAt 非空（A27 同步时间展示，P4a 前仅 API 暴露）； +// ⑥ 新端点守卫/校验：缺签名 401 SYNC_AUTH_FAILED；likes 缺 photoId → +// 400 VALIDATION_FAILED（契约「likes 时 photoId 必填」）。 +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 { ErrorCodes, 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'; +// AppOvModule 含 realtime 网关：进程内 ov 入口须与生产 main-ov 一致安装契约 +// WS 适配器（同 sync.e2e-spec 约定）。 +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 { StatReturnSender } from '../src/stat-return'; + +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'; + +const STAT_SNAPSHOTS_PATH = '/v1/internal/sync/stat-snapshots'; + +/** 签名 POST（直驱 cn 接收端构造乱序/越权场景；可省签名头）。 */ +async function signedPost( + base: string, + path: string, + body: unknown, + omitSignature = false, +): Promise { + const raw = JSON.stringify(body); + const headers: Record = { + 'content-type': 'application/json', + 'x-sync-timestamp': String(Math.floor(Date.now() / 1000)), + }; + if (!omitSignature) { + headers['x-sync-signature'] = signSyncBody(SYNC_KEY, raw); + } + return fetch(`${base}${path}`, { method: 'POST', headers, body: raw }); +} + +/** 测试内生成真实 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(); +} + +/** 线格式快照（直推场景手工构造）。 */ +interface WireSnapshot { + photoId?: string; + albumId: string; + kind: 'likes' | 'views'; + total: number; + statVersion: number; + source: string; + confirmedAt: string; +} + +describe('stat return ov→cn (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 statReturn: StatReturnSender; + let adminToken: string; + + let albumId: string; + let publicToken: string; + let photoId: string; + + /** 排空内容发送器（真实管线把相册/照片同步到 ov 副本）。 */ + 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 仍有未投递事件'); + } + + /** ov 公开点赞（真实入口，批次幂等键每调用唯一）。 */ + async function like( + sessionId: string, + batchId: string, + delta: number, + ): Promise<{ total: number; countVersion: number }> { + const res = await request(ovBase) + .post(`/v1/public/albums/${publicToken}/photos/${photoId}/likes`) + .send({ sessionId, batchId, delta }); + expect(res.status).toBe(200); + const body = res.body as { + total: number; + countVersion: number; + batchAccepted: boolean; + }; + expect(body.batchAccepted).toBe(true); + // 只取断言口径字段（响应含 batchAccepted，直接 toEqual 整体会因多属性失败）。 + return { total: body.total, countVersion: body.countVersion }; + } + + /** ov 公开浏览上报（真实入口）。 */ + async function visit( + visitorId: string, + requestId: string, + ): Promise<{ viewCount: number; counted: boolean }> { + const res = await request(ovBase) + .post(`/v1/public/albums/${publicToken}/visits`) + .send({ visitorId, requestId }); + expect(res.status).toBe(200); + return res.body as { viewCount: number; counted: boolean }; + } + + /** cn 照片副本现值断言口径。 */ + async function cnPhotoStats(): Promise<{ + likeCount: number; + statsVersion: number; + }> { + const row = await prismaCn.photos.findUniqueOrThrow({ + where: { id: photoId }, + select: { likeCount: true, statsVersion: true }, + }); + return row; + } + + /** cn 相册统计副本现值断言口径。 */ + async function cnAlbumStats(): Promise<{ + viewCountReplica: number; + statsVersion: number; + statsSyncedAt: Date | null; + }> { + const row = await prismaCn.albums.findUniqueOrThrow({ + where: { id: albumId }, + select: { + viewCountReplica: true, + statsVersion: true, + statsSyncedAt: true, + }, + }); + return row; + } + + 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, + }); + statReturn = new StatReturnSender({ + prismaOv, + cnApiUrl: cnBase, + 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; + + // 建相册 + 上传发布一张照片，经真实同步管线到 ov 副本。 + const album = await request(cnBase) + .post('/v1/albums') + .set('Authorization', `Bearer ${adminToken}`) + .send({ + translations: [ + { locale: 'zh-CN', title: '统计回传晚会' }, + { locale: 'en', title: 'Stat Return Gala' }, + ], + defaultLocale: 'zh-CN', + startAt: '2026-10-01T10:00:00.000Z', + }); + expect(album.status).toBe(201); + albumId = album.body.id as string; + publicToken = album.body.publicToken as string; + + const jpeg = await makeJpeg(960, 640, { r: 90, g: 160, b: 90 }); + const init = await request(cnBase) + .post('/v1/uploads') + .set('Authorization', `Bearer ${adminToken}`) + .send({ + albumId, + clientTaskId: 'stat-return-a', + filename: 'stat-return.jpg', + bytes: jpeg.length, + contentType: 'image/jpeg', + }); + expect(init.status).toBe(201); + const { uploadId, credential } = init.body as { + uploadId: string; + credential: { url: string; method: string; headers: Record }; + }; + photoId = (init.body as { photoId: string }).photoId; + 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'); + await drainSender(); + expect( + await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId }, + select: { visibility: true }, + }), + ).toMatchObject({ visibility: 'published' }); + }); + + afterAll(async () => { + await cnApp.close(); + await ovApp.close(); + await prismaCn.$disconnect(); + await prismaOv.$disconnect(); + }); + + it('① 逐次点赞回传：cn likeCount 被权威快照替换（2→5→10），绝非累加', async () => { + // 三次点赞、每次独立回传一轮——多轮回传是「替换而非累加」的判别性场景 + // （单轮回传时 total==Σdelta 不可判别；累加口径最终会到 2+5+10=17）。 + const first = await like('stat-ret-s1', 'stat-ret-b1', 2); + expect(first).toEqual({ total: 2, countVersion: 1 }); + let ret = await statReturn.runReturnOnce(); + expect(ret).toEqual({ picked: 1, snapshots: 1, accepted: 1 }); + expect(await cnPhotoStats()).toEqual({ likeCount: 2, statsVersion: 1 }); + + const second = await like('stat-ret-s2', 'stat-ret-b2', 3); + expect(second).toEqual({ total: 5, countVersion: 2 }); + ret = await statReturn.runReturnOnce(); + expect(ret).toEqual({ picked: 1, snapshots: 1, accepted: 1 }); + expect(await cnPhotoStats()).toEqual({ likeCount: 5, statsVersion: 2 }); + + const third = await like('stat-ret-s3', 'stat-ret-b3', 5); + expect(third).toEqual({ total: 10, countVersion: 3 }); + ret = await statReturn.runReturnOnce(); + expect(ret).toEqual({ picked: 1, snapshots: 1, accepted: 1 }); + // 最终一致 = ov 权威 total（10），不是快照之和（2+5+10）也不是 delta 之和 + // 的二次累加——A27「绝不相加」。 + expect(await cnPhotoStats()).toEqual({ likeCount: 10, statsVersion: 3 }); + + // 源行全部确认（sentAt 非空）。 + expect( + await prismaOv.outbox_stat_events.count({ where: { sentAt: null } }), + ).toBe(0); + expect( + await prismaOv.outbox_stat_events.count({ where: { kind: 'likes' } }), + ).toBe(3); + }); + + it('② 浏览回传：窗口去重后恰 1 条 views 快照 → viewCountReplica/statsSyncedAt', async () => { + const first = await visit('stat-ret-visitor-1', 'stat-ret-req-1'); + expect(first).toEqual({ viewCount: 1, counted: true }); + // 窗口内刷新（新 requestId、同 visitor）：不计入，也不产生新快照行。 + const refresh = await visit('stat-ret-visitor-1', 'stat-ret-req-2'); + expect(refresh).toEqual({ viewCount: 1, counted: false }); + + expect( + await prismaOv.outbox_stat_events.count({ where: { sentAt: null } }), + ).toBe(1); + + const ret = await statReturn.runReturnOnce(); + expect(ret).toEqual({ picked: 1, snapshots: 1, accepted: 1 }); + + const album = await cnAlbumStats(); + expect(album.viewCountReplica).toBe(1); + expect(album.statsVersion).toBe(1); + // A27：展示同步时间——首份 views 快照应用即落位。 + expect(album.statsSyncedAt).not.toBeNull(); + }); + + it('③ 重复回传幂等：重发已应用快照 accepted=0，cn 值分毫不动（A27）', async () => { + // 模拟「cn 已应用、ov 未收到 200」：清 sentAt 让回传器重发同一批。 + await prismaOv.outbox_stat_events.updateMany({ data: { sentAt: null } }); + const ret = await statReturn.runReturnOnce(); + // 4 源行（3 likes + 1 views）折叠为 2 快照；cn 侧同版本全判旧。 + expect(ret).toEqual({ picked: 4, snapshots: 2, accepted: 0 }); + + expect(await cnPhotoStats()).toEqual({ likeCount: 10, statsVersion: 3 }); + const album = await cnAlbumStats(); + expect(album.viewCountReplica).toBe(1); + expect(album.statsVersion).toBe(1); + expect(album.statsSyncedAt).not.toBeNull(); + // 重发确认照常落库（200 = 受理，应用与判旧都是终态）。 + expect( + await prismaOv.outbox_stat_events.count({ where: { sentAt: null } }), + ).toBe(0); + }); + + it('④ 乱序快照：接收端仅更高 statVersion 替换（版本权威，total 可下调）', async () => { + // 更高版本 + 更低 total：替换生效——证明权威是 statVersion 而非 total + // （未来下调修正可表达），也再次否定「max(total) 口径」。 + const higher = await signedPost(cnBase, STAT_SNAPSHOTS_PATH, { + snapshots: [ + { + photoId, + albumId, + kind: 'likes', + total: 7, + statVersion: 50, + source: 'ov', + confirmedAt: new Date().toISOString(), + }, + ], + }); + expect(higher.status).toBe(200); + expect(await higher.json()).toEqual({ accepted: 1 }); + expect(await cnPhotoStats()).toEqual({ likeCount: 7, statsVersion: 50 }); + + // 更低版本旧快照（无论 total 多大）：忽略。 + const lower = await signedPost(cnBase, STAT_SNAPSHOTS_PATH, { + snapshots: [ + { + photoId, + albumId, + kind: 'likes', + total: 999, + statVersion: 30, + source: 'ov', + confirmedAt: new Date().toISOString(), + }, + ], + }); + expect(lower.status).toBe(200); + expect(await lower.json()).toEqual({ accepted: 0 }); + expect(await cnPhotoStats()).toEqual({ likeCount: 7, statsVersion: 50 }); + }); + + it('④b 发送端聚合：同相册 v5/v3 未发送行 → 只发 v5，两行全确认', async () => { + // 手工构造乱序源行：v5 先插、v3 后插（createdAt 更晚）——聚合必须按 + // statVersion 取 max，不受插入顺序影响；v3 属「被更高版本覆盖」行， + // 随批确认（重发永无语义）。 + await prismaOv.outbox_stat_events.create({ + data: { kind: 'views', albumId, total: 42, statVersion: 5 }, + }); + await prismaOv.outbox_stat_events.create({ + data: { kind: 'views', albumId, total: 99, statVersion: 3 }, + }); + + const ret = await statReturn.runReturnOnce(); + expect(ret).toEqual({ picked: 2, snapshots: 1, accepted: 1 }); + + const album = await cnAlbumStats(); + expect(album.viewCountReplica).toBe(42); + expect(album.statsVersion).toBe(5); + expect( + await prismaOv.outbox_stat_events.count({ where: { sentAt: null } }), + ).toBe(0); + }); + + it('⑤ cn 管理展示：listAlbumPhotos.likeCount=副本现值；listAlbums.statsSyncedAt', async () => { + const photos = await request(cnBase) + .get(`/v1/albums/${albumId}/photos`) + .set('Authorization', `Bearer ${adminToken}`); + expect(photos.status).toBe(200); + const item = ( + photos.body.items as Array<{ photoId: string; likeCount: number }> + ).find((p) => p.photoId === photoId); + expect(item?.likeCount).toBe(7); + + const albums = await request(cnBase) + .get('/v1/albums') + .set('Authorization', `Bearer ${adminToken}`); + expect(albums.status).toBe(200); + const summary = ( + albums.body.items as Array<{ + id: string; + statsSyncedAt: string | null; + }> + ).find((a) => a.id === albumId); + expect(summary).toBeDefined(); + expect(summary?.statsSyncedAt).not.toBeNull(); + expect(Date.parse(summary?.statsSyncedAt as string)).not.toBeNaN(); + }); + + it('⑥ 新端点守卫/校验：缺签名 401 SYNC_AUTH_FAILED；likes 缺 photoId 400', async () => { + const snapshot: WireSnapshot = { + albumId, + kind: 'views', + total: 1, + statVersion: 1, + source: 'ov', + confirmedAt: new Date().toISOString(), + }; + const unauth = await signedPost( + cnBase, + STAT_SNAPSHOTS_PATH, + { snapshots: [snapshot] }, + true, + ); + expect(unauth.status).toBe(401); + expect(((await unauth.json()) as { code: string }).code).toBe( + ErrorCodes.SYNC_AUTH_FAILED, + ); + + const malformed = await signedPost(cnBase, STAT_SNAPSHOTS_PATH, { + snapshots: [ + { albumId, kind: 'likes', total: 1, statVersion: 1, source: 'ov', confirmedAt: new Date().toISOString() }, + ], + }); + expect(malformed.status).toBe(400); + expect(((await malformed.json()) as { code: string }).code).toBe( + ErrorCodes.VALIDATION_FAILED, + ); + }); +});