From 1f2a0c658fbac20244b1b7b187c111b06fae1e2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 18 Sep 2026 18:14:01 +0800 Subject: [PATCH] feat(ov): windowed dedup visit counting with retention cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 8 (P2): POST /public/albums/{token}/visits with visit_sessions window dedup (VISIT_WINDOW_MINUTES), requestId idempotent replay (echoes stored outcome; expired beyond VISIT_REQUEST_TTL_MINUTES rejected), album_stats viewCount/statVersion authority + outbox_stat_events views snapshots, WS stats.updated broadcast, visitorId+IP fixed-window rate limit (VISIT_WINDOW_REQUESTS), and ov CleanupModule scheduler (setInterval, CLEANUP_INTERVAL_MINUTES) deleting like_batches / visit_requests / visit_sessions past their windows — totals never touched. Contract prose aligned to replay semantics; constants VISIT_WINDOW_REQUESTS / CLEANUP_INTERVAL_MINUTES added. Co-Authored-By: Claude Code --- .env.example | 5 + packages/contracts/openapi.yaml | 13 +- packages/contracts/src/constants.ts | 4 + packages/contracts/src/generated/openapi.d.ts | 13 +- services/api/src/app-ov.module.ts | 9 +- services/api/src/cleanup/cleanup.module.ts | 16 + services/api/src/cleanup/cleanup.service.ts | 142 ++++ services/api/src/interactions/dto.ts | 24 +- .../src/interactions/interactions.module.ts | 13 +- .../api/src/interactions/visits.controller.ts | 34 + .../api/src/interactions/visits.service.ts | 319 +++++++ services/api/src/realtime/ov-events.ts | 17 +- services/api/src/realtime/realtime.gateway.ts | 13 + services/api/test/visits.e2e-spec.ts | 801 ++++++++++++++++++ 14 files changed, 1398 insertions(+), 25 deletions(-) create mode 100644 services/api/src/cleanup/cleanup.module.ts create mode 100644 services/api/src/cleanup/cleanup.service.ts create mode 100644 services/api/src/interactions/visits.controller.ts create mode 100644 services/api/src/interactions/visits.service.ts create mode 100644 services/api/test/visits.e2e-spec.ts diff --git a/.env.example b/.env.example index 745b986..d89d342 100644 --- a/.env.example +++ b/.env.example @@ -55,3 +55,8 @@ LIKE_SESSION_WINDOW_DELTA=500 LIKE_IP_WINDOW_DELTA=2000 LIKE_BURST_MERGE_MS=500 VISIT_REQUEST_TTL_MINUTES=35 +# P2 Task 8：浏览上报基础限速（visitorId+IP 固定窗，窗口=VISIT_WINDOW_MINUTES， +# 每窗请求数上限——去重本身为主缓解）与 ov 互动流水清理调度间隔（分钟，0=停用； +# 清理 like_batches/visit_requests/visit_sessions 超窗口行，累计总数永不清）。 +VISIT_WINDOW_REQUESTS=60 +CLEANUP_INTERVAL_MINUTES=60 diff --git a/packages/contracts/openapi.yaml b/packages/contracts/openapi.yaml index 8e26005..fa59e35 100644 --- a/packages/contracts/openapi.yaml +++ b/packages/contracts/openapi.yaml @@ -1202,12 +1202,14 @@ paths: 加载成功后上报一次（列表/图片/WS 等接口调用一律不触发计数）。 - requestId 为上报幂等键（建议 UUID），counted 确认前跨刷新/重试持久复用； - 重复 requestId 返回 counted=false 与权威 viewCount，不重复计数。 + 重复 requestId 幂等重放：counted 回显该 requestId 首次处理的结局 + （确认响应丢失后重试仍见 true），viewCount 恒为当前权威值，不重复计数。 - 去重窗口（VISIT_WINDOW_MINUTES，默认 30 分钟，D07）内同一 visitorId 不重复计数；窗口过后再上报按新访次计数。 - 过期 requestId（VISIT_REQUEST_TTL_MINUTES，默认 35 分钟）拒绝重放 - （counted=false），不得重新加计。 - - 基础限速：visitorId 滑窗，超限 429（RATE_LIMITED）；去重本身为主缓解。 + （counted=false，无论原结局），不得重新加计。 + - 基础限速：visitorId+来源 IP 固定窗（VISIT_WINDOW_REQUESTS，默认 + 60/窗），超限 429（RATE_LIMITED）；去重本身为主缓解。 - 相册非公开/紧急停用 404（ALBUM_NOT_PUBLIC）。 security: [] requestBody: @@ -2666,8 +2668,9 @@ components: type: boolean readOnly: true description: | - 本次是否实际计数：true=新访次；false=去重窗口内重复、重复 - requestId 或过期 requestId 重放（均不重复计数）。 + 本次是否计入：true=新访次，或幂等重放已确认计入的 requestId + （回显原结局）；false=去重窗口内重复或过期 requestId 重放 + （均不重复计数）。 # ── 内部同步 ── diff --git a/packages/contracts/src/constants.ts b/packages/contracts/src/constants.ts index 43af7cd..4fbfdc0 100644 --- a/packages/contracts/src/constants.ts +++ b/packages/contracts/src/constants.ts @@ -28,3 +28,7 @@ export const LIKE_IP_WINDOW_DELTA = 2000; export const LIKE_BURST_MERGE_MS = 500; /** 浏览上报 requestId 保留分钟：过期 requestId 拒绝重放（≥ 去重窗口）。 */ export const VISIT_REQUEST_TTL_MINUTES = 35; +/** 浏览上报基础限速：visitorId+IP 固定窗（=去重窗口）内请求数上限，超限 429。 */ +export const VISIT_WINDOW_REQUESTS = 60; +/** ov 互动流水清理调度间隔分钟（like_batches/visit_requests/visit_sessions）。 */ +export const CLEANUP_INTERVAL_MINUTES = 60; diff --git a/packages/contracts/src/generated/openapi.d.ts b/packages/contracts/src/generated/openapi.d.ts index 2f15184..77f28b6 100644 --- a/packages/contracts/src/generated/openapi.d.ts +++ b/packages/contracts/src/generated/openapi.d.ts @@ -554,12 +554,14 @@ export interface paths { * 加载成功后上报一次（列表/图片/WS 等接口调用一律不触发计数）。 * * - requestId 为上报幂等键（建议 UUID），counted 确认前跨刷新/重试持久复用； - * 重复 requestId 返回 counted=false 与权威 viewCount，不重复计数。 + * 重复 requestId 幂等重放：counted 回显该 requestId 首次处理的结局 + * （确认响应丢失后重试仍见 true），viewCount 恒为当前权威值，不重复计数。 * - 去重窗口（VISIT_WINDOW_MINUTES，默认 30 分钟，D07）内同一 visitorId * 不重复计数；窗口过后再上报按新访次计数。 * - 过期 requestId（VISIT_REQUEST_TTL_MINUTES，默认 35 分钟）拒绝重放 - * （counted=false），不得重新加计。 - * - 基础限速：visitorId 滑窗，超限 429（RATE_LIMITED）；去重本身为主缓解。 + * （counted=false，无论原结局），不得重新加计。 + * - 基础限速：visitorId+来源 IP 固定窗（VISIT_WINDOW_REQUESTS，默认 + * 60/窗），超限 429（RATE_LIMITED）；去重本身为主缓解。 * - 相册非公开/紧急停用 404（ALBUM_NOT_PUBLIC）。 */ post: operations["reportAlbumVisit"]; @@ -1334,8 +1336,9 @@ export interface components { */ readonly viewCount: number; /** - * @description 本次是否实际计数：true=新访次；false=去重窗口内重复、重复 - * requestId 或过期 requestId 重放（均不重复计数）。 + * @description 本次是否计入：true=新访次，或幂等重放已确认计入的 requestId + * （回显原结局）；false=去重窗口内重复或过期 requestId 重放 + * （均不重复计数）。 */ readonly counted: boolean; }; diff --git a/services/api/src/app-ov.module.ts b/services/api/src/app-ov.module.ts index cf229e9..c638ad1 100644 --- a/services/api/src/app-ov.module.ts +++ b/services/api/src/app-ov.module.ts @@ -7,8 +7,9 @@ // （public-ov：读 photos_ov 副本的同形契约端点 + hot 排序）；Task 7 起加入 // interactions（点赞批次幂等 + Redis ov 限速）与 realtime（WS 网关：契约帧 // 分发见 main-ov 的 ContractWsAdapter），EventEmitter2 全局总线承接「事务提交 -// → 实时广播」（ov-events.ts，单进程——多实例 P5）。cn 模块保持原样，两入口 -// 互不影响。 +// → 实时广播」（ov-events.ts，单进程——多实例 P5）；Task 8 起 interactions +// 增浏览上报（visits：窗口去重 + requestId 幂等）并加入 cleanup（互动流水 +// 超窗口行回收调度）。cn 模块保持原样，两入口互不影响。 import { type MiddlewareConsumer, Module, @@ -19,6 +20,7 @@ import { APP_FILTER, APP_PIPE } from '@nestjs/core'; import { EventEmitterModule } from '@nestjs/event-emitter'; import { LoggerModule } from 'nestjs-pino'; import { AllExceptionsFilter } from './all-exceptions.filter'; +import { CleanupModule } from './cleanup/cleanup.module'; import { EmergencyModule } from './emergency/emergency.module'; import { HealthController } from './health.controller'; import { InteractionsModule } from './interactions/interactions.module'; @@ -44,6 +46,9 @@ import { SyncInboxModule } from './sync-inbox/sync-inbox.module'; InteractionsModule, // ov 实时广播（P2 Task 7）：WS /v1/realtime（适配器见 main-ov）。 RealtimeModule, + // ov 互动流水清理调度（P2 Task 8）：like_batches/visit_requests/ + // visit_sessions 超窗口行回收（累计总数永不清理）。 + CleanupModule, ], controllers: [HealthController], providers: [ diff --git a/services/api/src/cleanup/cleanup.module.ts b/services/api/src/cleanup/cleanup.module.ts new file mode 100644 index 0000000..0cd9b73 --- /dev/null +++ b/services/api/src/cleanup/cleanup.module.ts @@ -0,0 +1,16 @@ +// services/api/src/cleanup/ —— ov 互动流水清理模块（P2 Task 8）。仅 ov 入口 +// 装配（app-ov.module 显式挑选；cn 入口不感知——cn 侧无互动流水表）。 +// CleanupService 同时导出：调度（setInterval，见 cleanup.service.ts 头注） +// 与 runCleanupOnce（e2e 直调断言）。语义边界：仅清 like_batches / +// visit_requests / visit_sessions 超窗口行，累计总数（album_stats、 +// photos_ov.likeTotal）永不触碰。 +import { Module } from '@nestjs/common'; +import { OvPrismaModule } from '../prisma-ov/ov-prisma.module'; +import { CleanupService } from './cleanup.service'; + +@Module({ + imports: [OvPrismaModule], + providers: [CleanupService], + exports: [CleanupService], +}) +export class CleanupModule {} diff --git a/services/api/src/cleanup/cleanup.service.ts b/services/api/src/cleanup/cleanup.service.ts new file mode 100644 index 0000000..f132d02 --- /dev/null +++ b/services/api/src/cleanup/cleanup.service.ts @@ -0,0 +1,142 @@ +// services/api/src/cleanup/cleanup.service.ts —— ov 互动流水清理调度 +// （P2 Task 8，评审缺口 2）。清理对象与边界： +// +// · like_batches：createdAt 早于 LIKE_BATCH_WINDOW_HOURS（24h）——点赞幂等 +// 账本。清理即「窗口外重放拒绝点」：行被清后重放同 batchId 会以新批次 +// 重新加计（无永久幂等账本与无界增长的取舍，likes.service.ts 头注已 +// 登记偏差；sessionId/IP 滑窗上限钳制重放影响）； +// · visit_requests：createdAt 早于 VISIT_REQUEST_TTL_MINUTES（35min ≥ +// 去重窗口 30min）——过期行先于清理即被重放拒绝（counted=false）， +// 清理只回收空间；TTL ≥ 窗口保证行被清后重放至多以「新 requestId」 +// 身份走窗口判定，不会绕过 visit_sessions 去重双计； +// · visit_sessions：lastCountedAt 早于 VISIT_WINDOW_MINUTES——去重锚点 +// 只需覆盖窗口内状态，窗外行等价「无行」（重访走新窗计入）； +// · **累计总数永不清理**：album_stats（viewCount）与 photos_ov.likeTotal +// 是权威计数，长期保留（Task 9 回传 cn 的数据源）。 +// +// 调度：setInterval 驱动（CLEANUP_INTERVAL_MINUTES，默认 60，≤0 停用）， +// 首轮延迟一个间隔（进程刚起无需立即清）。不引入 @nestjs/schedule—— +// realtime.gateway 心跳 / sync main 循环同款 setInterval 约定，env 可调 +// 间隔用原生定时器最直白；runCleanupOnce 导出为可注入方法，e2e 直接调用 +// 断言（不等待真实定时）。窗口阈值惰性读 env（同 visits/likes 服务约定， +// 缺省回落契约常量）。 +import { + Injectable, + Logger, + type OnModuleDestroy, + type OnModuleInit, +} from '@nestjs/common'; +import { + CLEANUP_INTERVAL_MINUTES, + LIKE_BATCH_WINDOW_HOURS, + VISIT_REQUEST_TTL_MINUTES, + VISIT_WINDOW_MINUTES, +} from '@photo/contracts'; +import { OvPrismaService } from '../prisma-ov/ov-prisma.module'; + +/** 清理调度间隔分钟（惰性读 env；≤0 = 停用调度，runCleanupOnce 仍可直调）。 */ +export function cleanupIntervalMinutes(): number { + const parsed = Number.parseInt( + process.env.CLEANUP_INTERVAL_MINUTES ?? '', + 10, + ); + return Number.isFinite(parsed) && parsed >= 0 + ? parsed + : CLEANUP_INTERVAL_MINUTES; +} + +/** 单轮清理观测口径（日志打点；e2e 断言删除行数）。 */ +export interface CleanupStats { + likeBatches: number; + visitRequests: number; + visitSessions: number; +} + +/** like_batches 保留窗毫秒（LIKE_BATCH_WINDOW_HOURS，惰性 env）。 */ +function likeBatchWindowMs(): number { + const parsed = Number.parseInt(process.env.LIKE_BATCH_WINDOW_HOURS ?? '', 10); + const hours = + Number.isFinite(parsed) && parsed > 0 ? parsed : LIKE_BATCH_WINDOW_HOURS; + return hours * 3_600_000; +} + +/** visit_requests 保留窗毫秒（VISIT_REQUEST_TTL_MINUTES，惰性 env）。 */ +function visitRequestTtlMs(): number { + const parsed = Number.parseInt( + process.env.VISIT_REQUEST_TTL_MINUTES ?? '', + 10, + ); + const minutes = + Number.isFinite(parsed) && parsed > 0 + ? parsed + : VISIT_REQUEST_TTL_MINUTES; + return minutes * 60_000; +} + +/** visit_sessions 保留窗毫秒（VISIT_WINDOW_MINUTES，惰性 env）。 */ +function visitWindowMs(): number { + const parsed = Number.parseInt(process.env.VISIT_WINDOW_MINUTES ?? '', 10); + const minutes = + Number.isFinite(parsed) && parsed > 0 ? parsed : VISIT_WINDOW_MINUTES; + return minutes * 60_000; +} + +@Injectable() +export class CleanupService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(CleanupService.name); + private timer: NodeJS.Timeout | null = null; + + constructor(private readonly prisma: OvPrismaService) {} + + onModuleInit(): void { + const minutes = cleanupIntervalMinutes(); + if (minutes <= 0) { + this.logger.log('清理调度停用（CLEANUP_INTERVAL_MINUTES≤0）'); + return; + } + this.timer = setInterval(() => { + this.runCleanupOnce().catch((err: unknown) => { + this.logger.error( + `清理轮失败: ${err instanceof Error ? err.message : String(err)}`, + ); + }); + }, minutes * 60_000); + // 不阻塞进程退出（应用关闭经 onModuleDestroy 显式清理定时器）。 + this.timer.unref?.(); + this.logger.log(`清理调度就绪：每 ${minutes} 分钟一轮`); + } + + onModuleDestroy(): void { + if (this.timer !== null) { + clearInterval(this.timer); + this.timer = null; + } + } + + /** 单轮清理：三表超窗口行删除，累计总数（album_stats/photos_ov）不触碰。 */ + async runCleanupOnce(): Promise { + const now = Date.now(); + const [likeBatches, visitRequests, visitSessions] = await Promise.all([ + this.prisma.like_batches.deleteMany({ + where: { createdAt: { lt: new Date(now - likeBatchWindowMs()) } }, + }), + this.prisma.visit_requests.deleteMany({ + where: { createdAt: { lt: new Date(now - visitRequestTtlMs()) } }, + }), + this.prisma.visit_sessions.deleteMany({ + where: { lastCountedAt: { lt: new Date(now - visitWindowMs()) } }, + }), + ]); + const stats: CleanupStats = { + likeBatches: likeBatches.count, + visitRequests: visitRequests.count, + visitSessions: visitSessions.count, + }; + this.logger.log( + `清理完成：like_batches=${stats.likeBatches} ` + + `visit_requests=${stats.visitRequests} ` + + `visit_sessions=${stats.visitSessions}`, + ); + return stats; + } +} diff --git a/services/api/src/interactions/dto.ts b/services/api/src/interactions/dto.ts index c148b0f..44b9982 100644 --- a/services/api/src/interactions/dto.ts +++ b/services/api/src/interactions/dto.ts @@ -1,8 +1,9 @@ -// services/api/src/interactions/dto.ts —— 点赞批次上报请求体（P2 Task 7）。 -// 校验边界与契约 LikeBatchRequest 逐字段对齐（sessionId/batchId 1..64、 -// delta 1..20 整数）；越界一律 400 VALIDATION_FAILED（全局 ValidationPipe -// whitelist 剥离未声明字段）。delta 必须是 JSON number 整数（管道未开 -// transform，字符串数字不静默转换——公开匿名端点从严）。 +// services/api/src/interactions/dto.ts —— 互动上报请求体（P2 Task 7 点赞批次 / +// Task 8 浏览上报）。校验边界与契约 LikeBatchRequest / VisitReportRequest +// 逐字段对齐（sessionId/batchId 1..64、delta 1..20 整数；visitorId/requestId +// 8..128）；越界一律 400 VALIDATION_FAILED（全局 ValidationPipe whitelist +// 剥离未声明字段）。delta 必须是 JSON number 整数（管道未开 transform， +// 字符串数字不静默转换——公开匿名端点从严）。 import { IsInt, IsString, Length, Max, Min } from 'class-validator'; export class LikeBatchDto { @@ -22,3 +23,16 @@ export class LikeBatchDto { @Max(20) delta!: number; } + +/** 浏览上报请求体（P2 Task 8；契约 VisitReportRequest）。 */ +export class VisitReportDto { + /** 访客标识（客户端持久化并在标签页间共享；去重窗口与限速的键）。 */ + @IsString() + @Length(8, 128) + visitorId!: string; + + /** 上报幂等键：counted 确认前跨刷新/重试持久复用（建议 UUID）。 */ + @IsString() + @Length(8, 128) + requestId!: string; +} diff --git a/services/api/src/interactions/interactions.module.ts b/services/api/src/interactions/interactions.module.ts index d02c9c8..140039b 100644 --- a/services/api/src/interactions/interactions.module.ts +++ b/services/api/src/interactions/interactions.module.ts @@ -1,17 +1,20 @@ // services/api/src/interactions/ —— ov 公开互动模块（P2 Task 7 起）。 // 仅 ov 入口装配（app-ov.module 显式挑选；cn 入口不感知）。Task 7 装配 -// 点赞批次端点；Task 8 起浏览上报（visits）同域加入。事件广播经 -// EventEmitter2（app 根模块 EventEmitterModule.forRoot() 全局注册）发往 -// realtime 网关，无模块间直接依赖。 +// 点赞批次端点；Task 8 装配浏览上报（visits.service/controller）。 +// 事件广播经 EventEmitter2（app 根模块 EventEmitterModule.forRoot() +// 全局注册）发往 realtime 网关（likes.burst / stats.updated），无模块间 +// 直接依赖。 import { Module } from '@nestjs/common'; import { OvPrismaModule } from '../prisma-ov/ov-prisma.module'; import { OvRedisModule } from '../redis-ov/ov-redis.module'; import { LikesController } from './likes.controller'; import { LikesService } from './likes.service'; +import { VisitsController } from './visits.controller'; +import { VisitsService } from './visits.service'; @Module({ imports: [OvPrismaModule, OvRedisModule], - controllers: [LikesController], - providers: [LikesService], + controllers: [LikesController, VisitsController], + providers: [LikesService, VisitsService], }) export class InteractionsModule {} diff --git a/services/api/src/interactions/visits.controller.ts b/services/api/src/interactions/visits.controller.ts new file mode 100644 index 0000000..1dce30c --- /dev/null +++ b/services/api/src/interactions/visits.controller.ts @@ -0,0 +1,34 @@ +// services/api/src/interactions/visits.controller.ts —— 浏览上报端点 +// （P2 Task 8，ov 域）。路由与契约 reportAlbumVisit 一致（tags [interactions]， +// 路径前缀不变、部署域名区分）；匿名（无 SessionGuard）。IP 取 req.ip 并归一 +// IPv6 映射前缀（::ffff:127.0.0.1 → 127.0.0.1），仅作限速键——不落库、不进 +// 日志（匿名观众端无 IP 留存需求；同 likes.controller 约定）。真实部署经反代 +// 时的 trust proxy 处理登记 P5。 +import { + Body, + Controller, + HttpCode, + HttpStatus, + Ip, + Param, + Post, +} from '@nestjs/common'; +import { VisitReportDto } from './dto'; +import { VisitsService, type VisitReportResponse } from './visits.service'; + +@Controller('public/albums') +export class VisitsController { + constructor(private readonly visits: VisitsService) {} + + /** 契约 200（窗口内重复/幂等重放/过期拒绝均 200，语义在 counted）。 */ + @HttpCode(HttpStatus.OK) + @Post(':token/visits') + report( + @Param('token') token: string, + @Body() dto: VisitReportDto, + @Ip() ip: string | undefined, + ): Promise { + const clientIp = ip === undefined ? undefined : ip.replace(/^::ffff:/, ''); + return this.visits.reportVisit(token, dto, clientIp); + } +} diff --git a/services/api/src/interactions/visits.service.ts b/services/api/src/interactions/visits.service.ts new file mode 100644 index 0000000..62689f7 --- /dev/null +++ b/services/api/src/interactions/visits.service.ts @@ -0,0 +1,319 @@ +// services/api/src/interactions/visits.service.ts —— 浏览统计（P2 Task 8， +// ov 域）。契约 POST /public/albums/{token}/visits（reportAlbumVisit）。 +// 全流程（A14 仿真级语义）： +// +// ① 校验：相册 publicToken → albums_ov + isAlbumVisibleOv（含紧急停用） +// → 统一 404 ALBUM_NOT_PUBLIC；DTO 边界（visitorId/requestId 8..128） +// 由全局管道 400。 +// ② 幂等/过期快速通道（不进事务、不消耗限速预算——重试已确认 requestId +// 是正常路径，同 likes ② 约定）：visit_requests 命中既有行 → +// 行龄 > VISIT_REQUEST_TTL_MINUTES → 拒绝重放 {counted:false}（无论 +// 原结局）；否则回显原结局 counted（确认响应丢失的重试仍见 true—— +// 客户端「requestId 持久到 counted 确认为止」的闭环依赖此语义）。 +// viewCount 恒为当前权威值（album_stats，单调不减）。 +// ③ 基础限速（Redis ov 固定窗，visitorId+IP 键，VISIT_WINDOW_REQUESTS/ +// 窗，窗长=去重窗口）：超限 429 RATE_LIMITED。去重本身是主缓解 +// （已登记），限速定位是「误客户端缓冲」而非防滥用（真实防护 P5）。 +// ④ 入账事务：visit_requests 插入（requestId 唯一）→ visit_sessions +// （albumId+visitorId 唯一）读窗口态：lastCountedAt 距今 < 窗口 → +// counted=false（窗口内重复，事务仅保留 counted=false 的幂等行）； +// 否则（无行 / 窗口外）→ session lastCountedAt=now + album_stats +// 原子 viewCount+1、statVersion+1 + outbox_stat_events {kind views, +// total, statVersion}（Task 9 回传队列）+ visit_requests.counted=true。 +// ⑤ 提交后：发布 OV_VISIT_COUNTED 进程内事件 → WS 网关向相册房间广播 +// stats.updated{albumId, viewCount, countVersion=statVersion}。 +// +// —— 并发语义（唯一约束 + 冲突后重读，不做盲目重试）—— +// · 并发同 requestId：事务内插入撞 visit_requests 主键（P2002）→ 冲突后 +// 重读行存在 → 走 ② 重放语义（胜者已提交，回显其结局）。 +// · 并发同 visitor 不同 requestId（首访竞态）：双方都读到「无 session」， +// 败者的 session create 撞 (albumId,visitorId) 唯一（P2002，整个事务 +// 回滚——含其 visit_requests 行）→ 冲突后重读：requestId 行不存在 ⇒ +// 是 session 竞态，且胜者刚把 lastCountedAt 置为 now（自己必在窗口内） +// → {counted:false, viewCount:权威}。重读而非重试：胜者已计数，重试 +// 只会得到同样的窗口判定，先答先正确。 +// · 并发不同 visitor 同册首访：album_stats 首行 create 竞态（albumId 主键 +// P2002）→ countView 内重走 updateMany 自增（InnoDB 当前读见已提交行）。 +// +// —— 上报排除规则（契约 description 落地）—— +// 浏览计数**仅**由本端点显式调用触发：公开读（album/列表/单图）、点赞、 +// WS 帧一概不写 visit_* / album_stats（e2e 断言行数不变）。 +import { HttpStatus, Injectable } from '@nestjs/common'; +import { EventEmitter2 } from '@nestjs/event-emitter'; +import { + VISIT_REQUEST_TTL_MINUTES, + VISIT_WINDOW_MINUTES, + VISIT_WINDOW_REQUESTS, + ErrorCodes, +} from '@photo/contracts'; +import type { components } from '@photo/contracts'; +import { ApiException } from '../api-exception'; +import { albumNotPublic } from '../public/support'; +import { OvPrismaService } from '../prisma-ov/ov-prisma.module'; +import { isAlbumVisibleOv } from '../shared/album-visibility'; +import { OvRedisService } from '../redis-ov/ov-redis.module'; +import { OV_VISIT_COUNTED } from '../realtime/ov-events'; +import type { VisitCountedEvent } from '../realtime/ov-events'; +import type { Prisma } from '@photo/db-ov'; +import type { VisitReportDto } from './dto'; + +export type VisitReportResponse = components['schemas']['VisitReportResponse']; + +/** 事务客户端类型别名（incrementViewCount 复用于并发首访自增路径）。 */ +type Tx = Prisma.TransactionClient; + +/** 去重窗口分钟（D07；惰性读 env 供测试调整）。 */ +export function visitWindowMinutes(): number { + const parsed = Number.parseInt(process.env.VISIT_WINDOW_MINUTES ?? '', 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : VISIT_WINDOW_MINUTES; +} + +/** requestId 保留分钟（过期拒绝重放；惰性读 env，≥ 去重窗口由初值保证）。 */ +export function visitRequestTtlMinutes(): number { + const parsed = Number.parseInt( + process.env.VISIT_REQUEST_TTL_MINUTES ?? '', + 10, + ); + return Number.isFinite(parsed) && parsed > 0 + ? parsed + : VISIT_REQUEST_TTL_MINUTES; +} + +/** visitorId+IP 固定窗内请求数上限（基础限速；去重为主缓解）。 */ +export function visitWindowRequests(): number { + const parsed = Number.parseInt(process.env.VISIT_WINDOW_REQUESTS ?? '', 10); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : VISIT_WINDOW_REQUESTS; +} + +function rateLimited(): ApiException { + return new ApiException( + ErrorCodes.RATE_LIMITED, + '上报过于频繁，请稍后再试', + HttpStatus.TOO_MANY_REQUESTS, + ); +} + +function isP2002(err: unknown): boolean { + return ( + typeof err === 'object' && + err !== null && + (err as { code?: unknown }).code === 'P2002' + ); +} + +@Injectable() +export class VisitsService { + constructor( + private readonly prisma: OvPrismaService, + private readonly redis: OvRedisService, + private readonly events: EventEmitter2, + ) {} + + async reportVisit( + token: string, + dto: VisitReportDto, + clientIp: string | undefined, + ): Promise { + // ① 相册公开校验（同 public-ov 读路径口径）。 + const album = await this.prisma.albums_ov.findUnique({ + where: { publicToken: token }, + }); + if (album === null || !isAlbumVisibleOv(album)) { + throw albumNotPublic(); + } + + // ② 幂等/过期快速通道（重放不消耗限速预算——同 likes ② 约定）。 + const existing = await this.prisma.visit_requests.findUnique({ + where: { requestId: dto.requestId }, + }); + if (existing !== null) { + return this.replay(album.albumId, existing.createdAt, existing.counted); + } + + // ③ 基础限速（通过后才进事务；拒绝也消耗预算——固定窗限速惯例）。 + await this.assertRateLimit(dto.visitorId, clientIp); + + // ④ 入账事务（两类唯一冲突的语义见文件头「并发语义」）。 + try { + const outcome = await this.prisma.$transaction(async (tx) => { + await tx.visit_requests.create({ + data: { + requestId: dto.requestId, + albumId: album.albumId, + visitorId: dto.visitorId, + }, + }); + const session = await tx.visit_sessions.findUnique({ + where: { + albumId_visitorId: { + albumId: album.albumId, + visitorId: dto.visitorId, + }, + }, + }); + const now = new Date(); + const windowMs = visitWindowMinutes() * 60_000; + const inWindow = + session !== null && + session.lastCountedAt !== null && + now.getTime() - session.lastCountedAt.getTime() < windowMs; + if (inWindow) { + // 窗口内重复：不计入——事务仅保留 counted=false 的幂等行。 + return { + viewCount: await this.currentViewCount(tx, album.albumId), + counted: false as const, + statVersion: null as number | null, + }; + } + // 计入：session 锚定本窗（新行直接带 lastCountedAt——null 行不存在）。 + if (session === null) { + await tx.visit_sessions.create({ + data: { + albumId: album.albumId, + visitorId: dto.visitorId, + lastCountedAt: now, + }, + }); + } else { + await tx.visit_sessions.update({ + where: { id: session.id }, + data: { lastCountedAt: now }, + }); + } + const { viewCount, statVersion } = await this.incrementViewCount( + tx, + album.albumId, + ); + // 回传队列记权威快照（Task 9：cn 侧仅更高 statVersion 替换）。 + await tx.outbox_stat_events.create({ + data: { + kind: 'views', + albumId: album.albumId, + total: viewCount, + statVersion, + }, + }); + await tx.visit_requests.update({ + where: { requestId: dto.requestId }, + data: { counted: true }, + }); + return { viewCount, counted: true as const, statVersion }; + }); + // ⑤ 提交后广播（仅计入路径；$transaction resolve 即已 COMMIT）。 + if (outcome.counted && outcome.statVersion !== null) { + const event: VisitCountedEvent = { + albumId: album.albumId, + viewCount: outcome.viewCount, + statVersion: outcome.statVersion, + }; + this.events.emit(OV_VISIT_COUNTED, event); + } + return { viewCount: outcome.viewCount, counted: outcome.counted }; + } catch (err) { + if (isP2002(err)) { + // 冲突后重读判型（不 inspect meta.target——判据更稳）： + // requestId 行已存在 → 并发同 requestId 重试 → ② 重放语义； + // 行不存在（整个事务已回滚）→ session 首访竞态，胜者刚计数 + // （lastCountedAt=now 已提交）→ 本请求必在窗口内 → counted=false。 + const winner = await this.prisma.visit_requests.findUnique({ + where: { requestId: dto.requestId }, + }); + if (winner !== null) { + return this.replay(album.albumId, winner.createdAt, winner.counted); + } + return { + viewCount: await this.currentViewCount(this.prisma, album.albumId), + counted: false, + }; + } + throw err; + } + } + + /** 重放应答：过期（行龄 > TTL）一律 counted=false；否则回显原结局。 */ + private async replay( + albumId: string, + createdAt: Date, + counted: boolean, + ): Promise { + const expired = + Date.now() - createdAt.getTime() > visitRequestTtlMinutes() * 60_000; + return { + viewCount: await this.currentViewCount(this.prisma, albumId), + counted: !expired && counted, + }; + } + + /** 当前权威浏览数（album_stats 无行默认 0；tx/根客户端皆可传入）。 */ + private async currentViewCount( + client: OvPrismaService | Tx, + albumId: string, + ): Promise { + const stats = await client.album_stats.findUnique({ + where: { albumId }, + select: { viewCount: true }, + }); + return stats?.viewCount ?? 0; + } + + /** + * viewCount+1 / statVersion+1（album_stats 权威自增）。无行先建（首访）， + * 并发不同 visitor 的首行竞态由 albumId 主键 P2002 兜底 → 重走 updateMany + * 自增（InnoDB 当前读可见对方已提交行），最多三轮。 + */ + private async incrementViewCount( + tx: Tx, + albumId: string, + ): Promise<{ viewCount: number; statVersion: number }> { + for (let attempt = 0; ; attempt++) { + const updated = await tx.album_stats.updateMany({ + where: { albumId }, + data: { + viewCount: { increment: 1 }, + statVersion: { increment: 1 }, + }, + }); + if (updated.count === 1) { + const row = await tx.album_stats.findUniqueOrThrow({ + where: { albumId }, + select: { viewCount: true, statVersion: true }, + }); + return { viewCount: row.viewCount, statVersion: row.statVersion }; + } + try { + await tx.album_stats.create({ + data: { albumId, viewCount: 1, statVersion: 1 }, + }); + return { viewCount: 1, statVersion: 1 }; + } catch (err) { + if (isP2002(err) && attempt < 2) { + continue; // 对方已建行：回到 updateMany 走自增。 + } + throw err; + } + } + } + + /** + * 固定窗限速：Redis ov INCR 计请求数，首次写键挂窗口 TTL（窗长=去重窗口， + * 键自本窗首次上报起存活一个窗口）。键 = visitorId+IP（同 NAT 出口的访客 + * 各占各的预算——去重窗口本身已钳制单 visitor 的计数影响，此处定位是 + * 误客户端缓冲，P5 速率层接真实防护）。 + */ + private async assertRateLimit( + visitorId: string, + clientIp: string | undefined, + ): Promise { + const key = `visit:window:${visitorId}|${clientIp ?? ''}`; + const counted = await this.redis.incr(key); + if (counted === 1) { + // 新键（本窗首次）：挂 TTL；EXPIRE 失败不影响本轮判定。 + await this.redis + .expire(key, visitWindowMinutes() * 60) + .catch(() => undefined); + } + if (counted > visitWindowRequests()) { + throw rateLimited(); + } + } +} diff --git a/services/api/src/realtime/ov-events.ts b/services/api/src/realtime/ov-events.ts index bf632d1..b858278 100644 --- a/services/api/src/realtime/ov-events.ts +++ b/services/api/src/realtime/ov-events.ts @@ -1,7 +1,8 @@ // services/api/src/realtime/ov-events.ts —— ov 侧「业务提交 → 实时广播」的 -// 进程内事件契约（P2 Task 7）。点赞端点（interactions）与收件箱应用 -// （sync-inbox）在**事务提交后**发布这些事件；WS 网关（realtime.gateway） -// 经 @OnEvent 监听并广播给相册房间。事件名/载荷只在本进程内有效。 +// 进程内事件契约（P2 Task 7；Task 8 增浏览计数）。点赞端点（interactions）、 +// 浏览计数（visits）与收件箱应用（sync-inbox）在**事务提交后**发布这些事件； +// WS 网关（realtime.gateway）经 @OnEvent 监听并广播给相册房间。事件名/载荷 +// 只在本进程内有效。 // // 单进程假设（登记偏差，P5 移交）：EventEmitter2 是进程内总线——ov 入口 // 当前单实例部署（compose 仿真），多实例横向扩展需换 Redis pub/sub 跨进程 @@ -24,6 +25,16 @@ export interface LikeBatchCommittedEvent { sessionId: string; } +/** 浏览已计入（visits 事务提交后发布；网关原样转 stats.updated）。 */ +export const OV_VISIT_COUNTED = 'ov.visit.counted'; + +export interface VisitCountedEvent { + albumId: string; + /** 提交事务后的权威累计浏览与统计版本（stats.updated 的 viewCount/countVersion）。 */ + viewCount: number; + statVersion: number; +} + /** 副本照片可见性已应用（sync-inbox 事务提交后发布；网关原样转 photo.*）。 */ export const OV_PHOTO_APPLIED = 'ov.photo.applied'; diff --git a/services/api/src/realtime/realtime.gateway.ts b/services/api/src/realtime/realtime.gateway.ts index ad79028..47650a4 100644 --- a/services/api/src/realtime/realtime.gateway.ts +++ b/services/api/src/realtime/realtime.gateway.ts @@ -44,8 +44,10 @@ import { isAlbumVisibleOv } from '../shared/album-visibility'; import { OV_LIKE_BATCH_COMMITTED, OV_PHOTO_APPLIED, + OV_VISIT_COUNTED, type LikeBatchCommittedEvent, type PhotoAppliedEvent, + type VisitCountedEvent, } from './ov-events'; /** likes.burst 合并窗口毫秒（惰性读 env，0=逐批即时广播；e2e 可调）。 */ @@ -227,6 +229,17 @@ export class RealtimeGateway implements OnModuleDestroy { pending.originBatchIds.push(event.batchId); } + /** 浏览已计入 → 原样转发契约 stats.updated 消息到相册房间（Task 8）。 */ + @OnEvent(OV_VISIT_COUNTED) + onVisitCounted(event: VisitCountedEvent): void { + this.sendToRoom(event.albumId, { + type: 'stats.updated', + albumId: event.albumId, + viewCount: event.viewCount, + countVersion: event.statVersion, + }); + } + /** 副本照片可见性应用 → 原样转发契约 photo.* 消息到相册房间。 */ @OnEvent(OV_PHOTO_APPLIED) onPhotoApplied(event: PhotoAppliedEvent): void { diff --git a/services/api/test/visits.e2e-spec.ts b/services/api/test/visits.e2e-spec.ts new file mode 100644 index 0000000..7f6145a --- /dev/null +++ b/services/api/test/visits.e2e-spec.ts @@ -0,0 +1,801 @@ +// services/api/test/visits.e2e-spec.ts —— P2 Task 8 浏览统计 + 去重数据清理 +// e2e。与 likes-realtime.e2e-spec 同模式：进程内起 cn + ov 两个真实入口， +// 业务动作全走 cn API（建相册/上传 complete/直驱发布事务/开关），OutboxSender +// （生产实现）指向 ov 入口真实 URL——浏览计数作用在与生产链路同源的数据上。 +// +// 覆盖（A14 仿真级 + Task 8 全部工作项）： +// ① A14 计数主干：首访 counted=true viewCount 1 → GET album 读到真实值 +// （Task 6 接线验证）；窗口内刷新（新 requestId）counted=false 不加计； +// 不同 visitorId counted=true 2；visit_requests/visit_sessions/ +// album_stats/outbox_stat_events（kind views）行级断言； +// ② requestId 幂等重试：重放回显原结局（counted=true 行仍 true / false 行 +// 仍 false），viewCount 当前权威，不加计； +// ③ 新窗口再计：OvPrisma 手工回退 visit_sessions.lastCountedAt 越窗 → +// 新 requestId counted=true 3； +// ④ 并发首访：Promise.all 同 visitorId 两个 requestId → 恰一个 +// counted=true（唯一约束 + 冲突后重读），总计数只 +1； +// ⑤ 过期 requestId 重放：回退 visit_requests.createdAt 越 TTL → +// counted=false（无论原结局），不加计； +// ⑥ 404/400：幽灵 token / publicEnabled=false 同步后的真实相册 → +// ALBUM_NOT_PUBLIC；visitorId 7 字符 / requestId 129 字符 / 缺字段 → +// VALIDATION_FAILED； +// ⑦ 基础限速：VISIT_WINDOW_REQUESTS 压小（惰性 env）→ 窗口超额 429 +// RATE_LIMITED；幂等快速通道不消耗限速预算（重放已确认 requestId 仍 200）； +// ⑧ 上报排除规则：GET album / 列表 / 单图 / 点赞 / WS 帧（subscribe+ping） +// 一律不触碰 visit_* / album_stats（行数与 viewCount 快照不变）； +// ⑨ WS stats.updated：计入 → 相册房间收 {albumId, viewCount, +// countVersion=statVersion}；窗口内重复（counted=false）不广播； +// ⑩ runCleanupOnce：fabricate 超窗旧行（like_batches 25h / visit_requests +// 36min / visit_sessions 31min）→ 全删；窗口内新行全留；累计总数 +// （album_stats / photos_ov.likeTotal）分毫不动； +// ⑪ 过期批次重放（Task 7 已登记偏差的文档化行为）：清理删除 like_batches +// 旧行后，重放同 batchId 以「新批次」身份重新加计（batchAccepted=true， +// within session cap）——无永久幂等账本的取舍边界，见 likes.service.ts +// 头注；此处固化该已登记行为（确定性：回退行 + 直调清理，无时间等待）。 +import { NestFactory } from '@nestjs/core'; +import { type NestExpressApplication } from '@nestjs/platform-express'; +import request from 'supertest'; +import sharp from 'sharp'; +import WebSocket from 'ws'; +import Redis from 'ioredis'; +import { PrismaClient } from '@photo/db-cn'; +import { PrismaClient as OvPrismaClient } from '@photo/db-ov'; +import { + type ClientMessage, + type ServerMessage, + ErrorCodes, + REALTIME_WS_PATH, +} from '@photo/contracts'; +import { + createStorageFromEnv, + type MinioStorageAdapter, +} from '@photo/storage'; +import { AppCnModule } from '../src/app-cn.module'; +import { AppOvModule } from '../src/app-ov.module'; +import { CleanupService } from '../src/cleanup/cleanup.service'; +import { seedAdmin } from '../src/scripts/seed-admin'; +import { ContractWsAdapter } from '../src/realtime/contract-ws.adapter'; +import { configureApp } from '../src/shared/configure-app'; +import { PublishService } from '../../worker/src/publish.service'; +import { resetDb } from './reset'; +import { resetOvDb } from './reset-ov'; +import { OutboxSender } from '../../sync/src/sender'; + +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'; + +/** subscribe 为异步校验（DB 查询）：发送后的就绪等待（本地查询 ~ms 级）。 */ +const SUBSCRIBE_SETTLE_MS = 300; +/** 「无新增广播」断言的观察窗（远大于任何提交路径时延）。 */ +const QUIET_MS = 1000; + +/** 测试内生成真实 JPEG（同 likes-realtime.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(); +} + +/** 浏览上报 POST（返回 supertest Response，断言交给调用方；body 宽松以覆盖 400 用例）。 */ +async function visit( + ovBase: string, + token: string, + body: { visitorId: string; requestId?: string }, +): Promise { + return request(ovBase) + .post(`/v1/public/albums/${token}/visits`) + .send(body); +} + +/** 契约帧 WS 测试客户端：收帧队列 + 条件等待（同 likes-realtime spec）。 */ +class WsClient { + private readonly ws: WebSocket; + private readonly frames: ServerMessage[] = []; + private readonly waiters: Array<{ + predicate: (frame: ServerMessage) => boolean; + resolve: (frame: ServerMessage) => void; + reject: (err: Error) => void; + timer: NodeJS.Timeout; + }> = []; + + private constructor(ws: WebSocket) { + this.ws = ws; + this.ws.on('message', (data) => { + let frame: ServerMessage; + try { + frame = JSON.parse(String(data)) as ServerMessage; + } catch { + return; + } + this.frames.push(frame); + const index = this.waiters.findIndex((w) => w.predicate(frame)); + if (index >= 0) { + const [waiter] = this.waiters.splice(index, 1); + clearTimeout(waiter.timer); + waiter.resolve(frame); + } + }); + } + + static async open(url: string): Promise { + const ws = new WebSocket(url); + await new Promise((resolve, reject) => { + ws.once('open', () => resolve()); + ws.once('error', (err: Error) => reject(err)); + }); + return new WsClient(ws); + } + + send(frame: ClientMessage): void { + this.ws.send(JSON.stringify(frame)); + } + + /** 等待首条满足谓词的帧（含历史帧；超时拒绝）。 */ + waitFor( + predicate: (frame: ServerMessage) => boolean, + timeoutMs: number, + label: string, + ): Promise { + const existing = this.frames.find(predicate); + if (existing !== undefined) { + return Promise.resolve(existing); + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + const index = this.waiters.findIndex((w) => w.reject === reject); + if (index >= 0) { + this.waiters.splice(index, 1); + } + reject( + new Error( + `waitFor(${label}) 超时 ${timeoutMs}ms；已收帧: ${JSON.stringify(this.frames)}`, + ), + ); + }, timeoutMs); + this.waiters.push({ predicate, resolve, reject, timer }); + }); + } + + /** 按类型统计（「不广播」断言用）。 */ + countOf(type: ServerMessage['type']): number { + return this.frames.filter((f) => f.type === type).length; + } + + close(): void { + this.ws.close(); + } +} + +describe('visits + cleanup (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 adminToken: string; + + let tokenA: string; + let tokenB: string; + let albumAId: string; + let photoA1: string; + + /** 排空发送器：循环 runOnce 直至无未投递事件。 */ + 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 仍有未投递事件'); + } + + /** cn 建相册 → 返回 {albumId, publicToken}。 */ + async function createAlbum(title: string): Promise<{ + albumId: string; + publicToken: string; + }> { + const album = await request(cnBase) + .post('/v1/albums') + .set('Authorization', `Bearer ${adminToken}`) + .send({ + translations: [{ locale: 'zh-CN', title }], + defaultLocale: 'zh-CN', + startAt: '2026-10-01T10:00:00.000Z', + }); + expect(album.status).toBe(201); + return { + albumId: album.body.id as string, + publicToken: album.body.publicToken as string, + }; + } + + /** 上传全链路：init → presign PUT 实写 MinIO → complete → 直驱发布事务。 */ + async function uploadAndPublish( + albumId: string, + clientTaskId: string, + jpeg: Buffer, + ): Promise { + const init = await request(cnBase) + .post('/v1/uploads') + .set('Authorization', `Bearer ${adminToken}`) + .send({ + albumId, + clientTaskId, + filename: 'visits.jpg', + bytes: jpeg.length, + contentType: 'image/jpeg', + }); + 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; + } + + /** ov 入口 WS 基址（同端口 HTTP+WS）。 */ + function wsUrl(): string { + return ovBase.replace(/^http/, 'ws') + REALTIME_WS_PATH; + } + + beforeAll(async () => { + expect(SYNC_KEY).not.toBe(''); + await resetDb(); + await resetOvDb(); + // 限速固定窗键跨 spec 运行残留（TTL=去重窗口 30min）：清场保证本 spec + // 从零计数（同 likes spec 的 like:window:* 清场约定）。 + const redis = new Redis(process.env.REDIS_URL_OV ?? 'redis://127.0.0.1:6381'); + let cursor = '0'; + do { + const [next, keys] = await redis.scan( + cursor, + 'MATCH', + 'visit:window:*', + 'COUNT', + 1000, + ); + cursor = next; + if (keys.length > 0) { + await redis.del(...keys); + } + } while (cursor !== '0'); + await redis.quit(); + + cnApp = await NestFactory.create(AppCnModule, { + rawBody: true, + }); + configureApp(cnApp); + await cnApp.listen(0); + cnBase = await cnApp.getUrl(); + + ovApp = await NestFactory.create(AppOvModule, { + rawBody: true, + }); + // main-ov 同形：契约帧适配器（WS 与 HTTP 同端口）。 + 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, + }); + + 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; + + // —— 种数据：相册 A（浏览主目标 + 1 张照片供排除规则/过期批次用例）、 + // 相册 B（publicEnabled=false → ov 同步后不可见，404 用例）—— + const albumA = await createAlbum('浏览统计 A'); + tokenA = albumA.publicToken; + albumAId = albumA.albumId; + const albumB = await createAlbum('浏览统计 B（未公开）'); + tokenB = albumB.publicToken; + photoA1 = await uploadAndPublish( + albumAId, + 'visits-a1', + await makeJpeg(900, 600, { r: 200, g: 100, b: 80 }), + ); + const off = await request(cnBase) + .patch(`/v1/albums/${albumB.albumId}`) + .set('Authorization', `Bearer ${adminToken}`) + .send({ publicEnabled: false }); + expect(off.status).toBe(200); + await drainSender(); + }); + + afterAll(async () => { + delete process.env.VISIT_WINDOW_REQUESTS; + await ovApp.close(); + await cnApp.close(); + await prismaCn.$disconnect(); + await prismaOv.$disconnect(); + }); + + it('① A14 计数主干：首访计入、窗口内去重、跨访客计入、GET album 真实 viewCount', async () => { + // 首访：counted=true，viewCount 1。 + const first = await visit(ovBase, tokenA, { + visitorId: 'visitor-a14-1', + requestId: 'req-a14-first', + }); + expect(first.status).toBe(200); + expect(first.body).toEqual({ viewCount: 1, counted: true }); + + // Task 6 接线：GET 公开相册读到 album_stats 真实值（非占位）。 + const album1 = await request(ovBase) + .get(`/v1/public/albums/${tokenA}`) + .query({ locale: 'zh-CN' }); + expect(album1.status).toBe(200); + expect(album1.body.viewCount).toBe(1); + + // 窗口内刷新（同 visitor 新 requestId）：counted=false，不加计。 + const refresh = await visit(ovBase, tokenA, { + visitorId: 'visitor-a14-1', + requestId: 'req-a14-refresh', + }); + expect(refresh.status).toBe(200); + expect(refresh.body).toEqual({ viewCount: 1, counted: false }); + + // 不同 visitorId：新访次计入。 + const second = await visit(ovBase, tokenA, { + visitorId: 'visitor-a14-2', + requestId: 'req-a14-second', + }); + expect(second.status).toBe(200); + expect(second.body).toEqual({ viewCount: 2, counted: true }); + + const album2 = await request(ovBase) + .get(`/v1/public/albums/${tokenA}`) + .query({ locale: 'zh-CN' }); + expect(album2.body.viewCount).toBe(2); + + // 行级断言：幂等行（counted 回映各自结局）、去重锚点、权威位、回传队列。 + const requests = await prismaOv.visit_requests.findMany({ + where: { albumId: albumAId }, + orderBy: { createdAt: 'asc' }, + }); + expect(requests.map((r) => [r.requestId, r.counted])).toEqual([ + ['req-a14-first', true], + ['req-a14-refresh', false], + ['req-a14-second', true], + ]); + expect(await prismaOv.visit_sessions.count()).toBe(2); + const stats = await prismaOv.album_stats.findUniqueOrThrow({ + where: { albumId: albumAId }, + }); + expect([stats.viewCount, stats.statVersion]).toEqual([2, 2]); + const outbox = await prismaOv.outbox_stat_events.findMany({ + where: { kind: 'views', albumId: albumAId }, + orderBy: { statVersion: 'asc' }, + }); + expect(outbox.map((r) => [r.total, r.statVersion])).toEqual([ + [1, 1], + [2, 2], + ]); + }); + + it('② requestId 幂等重试：回显原结局、viewCount 权威、不加计', async () => { + // 重放已计入的 requestId：counted 仍 true（确认丢失重试的闭环语义）。 + const retryCounted = await visit(ovBase, tokenA, { + visitorId: 'visitor-a14-1', + requestId: 'req-a14-first', + }); + expect(retryCounted.status).toBe(200); + expect(retryCounted.body).toEqual({ viewCount: 2, counted: true }); + + // 重放未计入（窗口内重复）的 requestId：counted 仍 false。 + const retrySkipped = await visit(ovBase, tokenA, { + visitorId: 'visitor-a14-1', + requestId: 'req-a14-refresh', + }); + expect(retrySkipped.body).toEqual({ viewCount: 2, counted: false }); + + // 不加计：幂等行不增、权威位不动。 + expect(await prismaOv.visit_requests.count()).toBe(3); + const stats = await prismaOv.album_stats.findUniqueOrThrow({ + where: { albumId: albumAId }, + }); + expect([stats.viewCount, stats.statVersion]).toEqual([2, 2]); + }); + + it('③ 新窗口再计：回退 lastCountedAt 越窗 → 同 visitor 再计入', async () => { + // 手工回退 visitor1 的去重锚点至窗口（30min）外。 + await prismaOv.visit_sessions.updateMany({ + where: { albumId: albumAId, visitorId: 'visitor-a14-1' }, + data: { lastCountedAt: new Date(Date.now() - 32 * 60_000) }, + }); + const again = await visit(ovBase, tokenA, { + visitorId: 'visitor-a14-1', + requestId: 'req-a14-newwindow', + }); + expect(again.status).toBe(200); + expect(again.body).toEqual({ viewCount: 3, counted: true }); + // 锚点被重新置为当前时刻（窗口重新起算）。 + const session = await prismaOv.visit_sessions.findUniqueOrThrow({ + where: { + albumId_visitorId: { albumId: albumAId, visitorId: 'visitor-a14-1' }, + }, + }); + expect( + Date.now() - session.lastCountedAt!.getTime(), + ).toBeLessThan(60_000); + }); + + it('④ 并发首访：同 visitorId 两 requestId → 恰一个 counted=true', async () => { + const [r1, r2] = await Promise.all([ + visit(ovBase, tokenA, { + visitorId: 'visitor-conc-1', + requestId: 'req-conc-1', + }), + visit(ovBase, tokenA, { + visitorId: 'visitor-conc-1', + requestId: 'req-conc-2', + }), + ]); + expect(r1.status).toBe(200); + expect(r2.status).toBe(200); + const outcomes = [r1.body, r2.body].sort( + (a: { counted: boolean }, b: { counted: boolean }) => + Number(b.counted) - Number(a.counted), + ); + // 恰一个计入（权威 viewCount=4），另一个窗口内重复（权威值同为 4）。 + expect(outcomes).toEqual([ + { viewCount: 4, counted: true }, + { viewCount: 4, counted: false }, + ]); + // 去重锚点唯一：并发只落一行 session；权威位只 +1。 + expect( + await prismaOv.visit_sessions.count({ + where: { visitorId: 'visitor-conc-1' }, + }), + ).toBe(1); + const stats = await prismaOv.album_stats.findUniqueOrThrow({ + where: { albumId: albumAId }, + }); + expect([stats.viewCount, stats.statVersion]).toEqual([4, 4]); + }); + + it('⑤ 过期 requestId 重放：越 TTL → counted=false（无论原结局），不加计', async () => { + // 回退首个已计入 requestId 的行龄至 TTL（35min）外。 + await prismaOv.visit_requests.update({ + where: { requestId: 'req-a14-first' }, + data: { createdAt: new Date(Date.now() - 36 * 60_000) }, + }); + const replay = await visit(ovBase, tokenA, { + visitorId: 'visitor-a14-1', + requestId: 'req-a14-first', + }); + expect(replay.status).toBe(200); + expect(replay.body).toEqual({ viewCount: 4, counted: false }); + const stats = await prismaOv.album_stats.findUniqueOrThrow({ + where: { albumId: albumAId }, + }); + expect(stats.viewCount).toBe(4); + }); + + it('⑥ 404/400：幽灵 token / 未公开相册 / 参数越界', async () => { + const ghost = await visit(ovBase, 'g' + '0'.repeat(23), { + visitorId: 'visitor-a14-1', + requestId: 'req-404-ghost', + }); + expect(ghost.status).toBe(404); + expect(ghost.body.code).toBe(ErrorCodes.ALBUM_NOT_PUBLIC); + + // 真实相册 publicEnabled=false（cn 已同步）：同码 404。 + const invisible = await visit(ovBase, tokenB, { + visitorId: 'visitor-a14-1', + requestId: 'req-404-invisible', + }); + expect(invisible.status).toBe(404); + expect(invisible.body.code).toBe(ErrorCodes.ALBUM_NOT_PUBLIC); + + for (const bad of [ + { visitorId: 'v'.repeat(7), requestId: 'req-bad-1' }, // visitorId < 8 + { visitorId: 'visitor-a14-1', requestId: 'r'.repeat(129) }, // requestId > 128 + { visitorId: 'visitor-a14-1' }, // 缺 requestId + ]) { + const res = await visit(ovBase, tokenA, bad); + expect(res.status).toBe(400); + expect(res.body.code).toBe(ErrorCodes.VALIDATION_FAILED); + } + }); + + it('⑦ 基础限速：visitorId+IP 固定窗超额 429；幂等快速通道不耗预算', async () => { + process.env.VISIT_WINDOW_REQUESTS = '3'; + try { + const visitor = 'visitor-rate-1'; + // 窗口预算 3：n1 计入（viewCount 5）、n2/n3 窗口内重复（仍消耗预算）。 + const n1 = await visit(ovBase, tokenA, { + visitorId: visitor, + requestId: 'req-rate-1', + }); + expect(n1.body).toEqual({ viewCount: 5, counted: true }); + const n2 = await visit(ovBase, tokenA, { + visitorId: visitor, + requestId: 'req-rate-2', + }); + expect(n2.body).toEqual({ viewCount: 5, counted: false }); + const n3 = await visit(ovBase, tokenA, { + visitorId: visitor, + requestId: 'req-rate-3', + }); + expect(n3.body).toEqual({ viewCount: 5, counted: false }); + + // 第 4 次：超额 → 429 RATE_LIMITED。 + const n4 = await visit(ovBase, tokenA, { + visitorId: visitor, + requestId: 'req-rate-4', + }); + expect(n4.status).toBe(429); + expect(n4.body.code).toBe(ErrorCodes.RATE_LIMITED); + + // 幂等快速通道：重放已确认 requestId 不进限速（客户端重试是正常路径）。 + const retry = await visit(ovBase, tokenA, { + visitorId: visitor, + requestId: 'req-rate-1', + }); + expect(retry.status).toBe(200); + expect(retry.body).toEqual({ viewCount: 5, counted: true }); + + // 被拒请求未入账：visit_requests 无 req-rate-4 行。 + expect( + await prismaOv.visit_requests.count({ + where: { requestId: 'req-rate-4' }, + }), + ).toBe(0); + } finally { + delete process.env.VISIT_WINDOW_REQUESTS; + } + }); + + it('⑧ 上报排除规则：读接口/点赞/WS 帧一律不触碰浏览数据', async () => { + const before = { + requests: await prismaOv.visit_requests.count(), + sessions: await prismaOv.visit_sessions.count(), + statsRows: await prismaOv.album_stats.count(), + viewCount: ( + await prismaOv.album_stats.findUniqueOrThrow({ + where: { albumId: albumAId }, + }) + ).viewCount, + }; + + // 公开读三端点。 + expect( + ( + await request(ovBase) + .get(`/v1/public/albums/${tokenA}`) + .query({ locale: 'zh-CN' }) + ).status, + ).toBe(200); + expect( + ( + await request(ovBase).get(`/v1/public/albums/${tokenA}/photos`) + ).status, + ).toBe(200); + expect( + ( + await request(ovBase) + .get(`/v1/public/albums/${tokenA}/photos/${photoA1}`) + ).status, + ).toBe(200); + + // 点赞（含照片计数写入——但与浏览表无关）。 + const like = await request(ovBase) + .post(`/v1/public/albums/${tokenA}/photos/${photoA1}/likes`) + .send({ sessionId: 'excl-session', batchId: 'excl-1', delta: 1 }); + expect(like.status).toBe(200); + + // WS 帧：subscribe（相册校验读）+ ping/pong。 + const client = await WsClient.open(wsUrl()); + try { + client.send({ type: 'subscribe', albumToken: tokenA }); + await new Promise((resolve) => setTimeout(resolve, SUBSCRIBE_SETTLE_MS)); + client.send({ type: 'ping', ts: Date.now() }); + await client.waitFor((f) => f.type === 'pong', 3000, 'pong'); + } finally { + client.close(); + } + + // 浏览域零变更：行数与权威 viewCount 均不变。 + expect(await prismaOv.visit_requests.count()).toBe(before.requests); + expect(await prismaOv.visit_sessions.count()).toBe(before.sessions); + expect(await prismaOv.album_stats.count()).toBe(before.statsRows); + expect( + ( + await prismaOv.album_stats.findUniqueOrThrow({ + where: { albumId: albumAId }, + }) + ).viewCount, + ).toBe(before.viewCount); + }); + + it('⑨ WS stats.updated：计入即广播（countVersion=statVersion）；重复不广播', async () => { + const client = await WsClient.open(wsUrl()); + try { + client.send({ type: 'subscribe', albumToken: tokenA }); + await new Promise((resolve) => setTimeout(resolve, SUBSCRIBE_SETTLE_MS)); + + // 新访客计入 → 房间收 stats.updated（viewCount 6 / countVersion 6）。 + const counted = await visit(ovBase, tokenA, { + visitorId: 'visitor-ws-1', + requestId: 'req-ws-1', + }); + expect(counted.body).toEqual({ viewCount: 6, counted: true }); + const frame = await client.waitFor( + (f) => f.type === 'stats.updated', + 5000, + 'stats.updated', + ); + if (frame.type !== 'stats.updated') { + throw new Error('unreachable'); + } + expect(frame.albumId).toBe(albumAId); + expect(frame.viewCount).toBe(6); + expect(frame.countVersion).toBe(6); + + // 窗口内重复（counted=false）：不广播。 + const skip = await visit(ovBase, tokenA, { + visitorId: 'visitor-ws-1', + requestId: 'req-ws-2', + }); + expect(skip.body).toEqual({ viewCount: 6, counted: false }); + await new Promise((resolve) => setTimeout(resolve, QUIET_MS)); + expect(client.countOf('stats.updated')).toBe(1); + } finally { + client.close(); + } + }); + + it('⑩ runCleanupOnce：超窗行全删、窗口内行全留、累计总数不动', async () => { + // fabricate 三类超窗旧行（直接落库，无时间等待）。 + await prismaOv.like_batches.create({ + data: { + photoId: photoA1, + sessionId: 'cleanup-old-session', + batchId: 'cleanup-old-batch', + delta: 5, + createdAt: new Date(Date.now() - 25 * 3_600_000), + }, + }); + await prismaOv.visit_requests.create({ + data: { + requestId: 'cleanup-old-req', + albumId: albumAId, + visitorId: 'cleanup-old-visitor', + createdAt: new Date(Date.now() - 36 * 60_000), + }, + }); + await prismaOv.visit_sessions.create({ + data: { + albumId: albumAId, + visitorId: 'cleanup-old-visitor', + lastCountedAt: new Date(Date.now() - 31 * 60_000), + }, + }); + + // 清理前基线（新鲜行数 + 两个权威总数）。 + const requestsBefore = await prismaOv.visit_requests.count(); + const sessionsBefore = await prismaOv.visit_sessions.count(); + const batchesBefore = await prismaOv.like_batches.count(); + const likeTotalBefore = ( + await prismaOv.photos_ov.findUniqueOrThrow({ where: { photoId: photoA1 } }) + ).likeTotal; + + const stats = await ovApp.get(CleanupService).runCleanupOnce(); + + // 删除量：like_batches 1 行旧批；visit_requests 2 行（fabricate 旧行 + + // ⑤ 回退的 req-a14-first，行龄 36min > TTL 35min）；visit_sessions 1 行。 + expect(stats.likeBatches).toBe(1); + expect(stats.visitRequests).toBe(2); + expect(stats.visitSessions).toBe(1); + + // 旧行确实消失。 + expect( + await prismaOv.like_batches.count({ + where: { batchId: 'cleanup-old-batch' }, + }), + ).toBe(0); + expect( + await prismaOv.visit_requests.count({ + where: { requestId: 'cleanup-old-req' }, + }), + ).toBe(0); + expect( + await prismaOv.visit_requests.count({ + where: { requestId: 'req-a14-first' }, + }), + ).toBe(0); + expect( + await prismaOv.visit_sessions.count({ + where: { visitorId: 'cleanup-old-visitor' }, + }), + ).toBe(0); + + // 窗口内新行全留（各减去被删的旧行数）。 + expect(await prismaOv.visit_requests.count()).toBe(requestsBefore - 2); + expect(await prismaOv.visit_sessions.count()).toBe(sessionsBefore - 1); + expect(await prismaOv.like_batches.count()).toBe(batchesBefore - 1); + + // 累计总数分毫不动：album_stats（viewCount/statVersion）与 photos_ov。 + const albumStats = await prismaOv.album_stats.findUniqueOrThrow({ + where: { albumId: albumAId }, + }); + expect([albumStats.viewCount, albumStats.statVersion]).toEqual([6, 6]); + expect( + ( + await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId: photoA1 }, + }) + ).likeTotal, + ).toBe(likeTotalBefore); + }); + + it('⑪ 过期批次重放（已登记偏差）：清理后重放旧 batchId 重新加计', async () => { + // 提交一批 → 回退行龄越窗 → 清理删除 → 重放同 batchId。 + const first = await request(ovBase) + .post(`/v1/public/albums/${tokenA}/photos/${photoA1}/likes`) + .send({ sessionId: 'dev-old-session', batchId: 'dev-old-batch', delta: 2 }); + expect(first.status).toBe(200); + const totalBefore = first.body.total as number; + + await prismaOv.like_batches.updateMany({ + where: { batchId: 'dev-old-batch' }, + data: { createdAt: new Date(Date.now() - 25 * 3_600_000) }, + }); + const cleanup = await ovApp.get(CleanupService).runCleanupOnce(); + expect(cleanup.likeBatches).toBe(1); + + // 已登记偏差（likes.service.ts 头注）：行被清后与「从未提交」不可区分， + // 重放以新批次身份重新加计（batchAccepted=true）——文档化行为固化； + // session 滑窗上限仍钳制其总量（delta 累计远低于默认 500 上限）。 + const replay = await request(ovBase) + .post(`/v1/public/albums/${tokenA}/photos/${photoA1}/likes`) + .send({ sessionId: 'dev-old-session', batchId: 'dev-old-batch', delta: 2 }); + expect(replay.status).toBe(200); + expect(replay.body.batchAccepted).toBe(true); + expect(replay.body.total).toBe(totalBefore + 2); + expect( + ( + await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId: photoA1 }, + }) + ).likeTotal, + ).toBe(totalBefore + 2); + }); +});