feat(ov): windowed dedup visit counting with retention cleanup
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 <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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 {}
|
||||
|
||||
@@ -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<VisitReportResponse> {
|
||||
const clientIp = ip === undefined ? undefined : ip.replace(/^::ffff:/, '');
|
||||
return this.visits.reportVisit(token, dto, clientIp);
|
||||
}
|
||||
}
|
||||
@@ -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<VisitReportResponse> {
|
||||
// ① 相册公开校验（同 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<VisitReportResponse> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user