浅念
1f2a0c658f
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>
336 lines
11 KiB
TypeScript
336 lines
11 KiB
TypeScript
// services/api/src/realtime/realtime.gateway.ts —— ov 实时网关（P2 Task 7）。
|
||
// 与 HTTP 同端口（ov 8081）挂载 /v1/realtime（ContractWsAdapter 按契约帧
|
||
// {type,...} 分发）。模型：**公开订阅、按相册房间广播**——
|
||
// · 连接无鉴权（观众匿名域）：subscribe 时逐相册校验 isAlbumVisibleOv
|
||
// （含紧急停用），失败回 {type:'error', code:ALBUM_NOT_PUBLIC}，连接保持；
|
||
// · 房间 = album:{albumId}：likes.burst（500ms 合并窗，originBatchIds 供
|
||
// 客户端过滤自身批次）、photo.*（sync-inbox 应用后转发）按房间扇出，
|
||
// 广播不排除发送者（无服务端「发送者」概念——originBatchIds 是过滤依据）；
|
||
// · 心跳：应用层 ping→pong（回显 ts）+ 传输层 HEARTBEAT_MS 周期 ping、
|
||
// 无 pong 的死连接 terminate（ws 客户端/浏览器按 RFC 自动回 pong）。
|
||
// 进程内事件源（ov-events.ts）：EventEmitter2 单进程扇出——多实例部署需
|
||
// Redis pub/sub（P5，登记偏差）。
|
||
import { randomUUID } from 'node:crypto';
|
||
import {
|
||
type OnModuleDestroy,
|
||
Injectable,
|
||
Logger,
|
||
} from '@nestjs/common';
|
||
import { OnEvent } from '@nestjs/event-emitter';
|
||
import {
|
||
ErrorCodes,
|
||
HEARTBEAT_MS,
|
||
LIKE_BURST_MERGE_MS,
|
||
REALTIME_WS_PATH,
|
||
} from '@photo/contracts';
|
||
// 契约帧类型与 Nest 装饰器重名（SubscribeMessage/WebSocketServer），以别名
|
||
// 区分：帧类型加 Frame 后缀，ws 原生服务器类型用 WsServer。
|
||
import type {
|
||
PongMessage,
|
||
RealtimeErrorMessage,
|
||
ServerMessage,
|
||
SubscribeMessage as SubscribeFrame,
|
||
UnsubscribeMessage as UnsubscribeFrame,
|
||
} from '@photo/contracts';
|
||
import { WebSocket } from 'ws';
|
||
import type { WebSocketServer as WsServer } from 'ws';
|
||
import {
|
||
SubscribeMessage,
|
||
WebSocketGateway,
|
||
WebSocketServer,
|
||
} from '@nestjs/websockets';
|
||
import { OvPrismaService } from '../prisma-ov/ov-prisma.module';
|
||
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 可调）。 */
|
||
export function likeBurstMergeMs(): number {
|
||
const parsed = Number.parseInt(process.env.LIKE_BURST_MERGE_MS ?? '', 10);
|
||
return Number.isFinite(parsed) && parsed >= 0 ? parsed : LIKE_BURST_MERGE_MS;
|
||
}
|
||
|
||
/** 房间键：album:{albumId}。 */
|
||
function roomKey(albumId: string): string {
|
||
return `album:${albumId}`;
|
||
}
|
||
|
||
/** 合并窗内的待发 burst（同照片聚合；flush 时一次广播）。 */
|
||
interface PendingBurst {
|
||
albumId: string;
|
||
delta: number;
|
||
total: number;
|
||
countVersion: number;
|
||
originBatchIds: string[];
|
||
timer: NodeJS.Timeout;
|
||
}
|
||
|
||
@Injectable()
|
||
@WebSocketGateway({ path: REALTIME_WS_PATH })
|
||
export class RealtimeGateway implements OnModuleDestroy {
|
||
private readonly logger = new Logger(RealtimeGateway.name);
|
||
|
||
@WebSocketServer()
|
||
private readonly server!: WsServer;
|
||
|
||
/** album:{albumId} → 订阅连接集（房间）。 */
|
||
private readonly rooms = new Map<string, Set<WebSocket>>();
|
||
/** 连接 → 已订 {publicToken → albumId}（退订/断开清理用）。 */
|
||
private readonly subscriptions = new Map<WebSocket, Map<string, string>>();
|
||
/** 传输层心跳存活标记（pong 置回 true；轮询时仍 false → terminate）。 */
|
||
private readonly alive = new Map<WebSocket, boolean>();
|
||
private heartbeatTimer: NodeJS.Timeout | null = null;
|
||
/** photoId → 合并窗待发 burst。 */
|
||
private readonly bursts = new Map<string, PendingBurst>();
|
||
|
||
constructor(private readonly prisma: OvPrismaService) {}
|
||
|
||
/** 传输层心跳：周期 ping，两个周期无 pong 的死连接踢除。 */
|
||
afterInit(): void {
|
||
this.heartbeatTimer = setInterval(() => {
|
||
for (const client of this.server.clients) {
|
||
const isAlive = this.alive.get(client) === true;
|
||
if (!isAlive) {
|
||
client.terminate();
|
||
continue;
|
||
}
|
||
this.alive.set(client, false);
|
||
client.ping();
|
||
}
|
||
}, HEARTBEAT_MS);
|
||
}
|
||
|
||
/** 应用生命周期结束：停心跳与合并窗定时器（房间随连接关闭清空）。 */
|
||
onModuleDestroy(): void {
|
||
if (this.heartbeatTimer !== null) {
|
||
clearInterval(this.heartbeatTimer);
|
||
this.heartbeatTimer = null;
|
||
}
|
||
for (const pending of this.bursts.values()) {
|
||
clearTimeout(pending.timer);
|
||
}
|
||
this.bursts.clear();
|
||
}
|
||
|
||
handleConnection(client: WebSocket): void {
|
||
this.alive.set(client, true);
|
||
this.subscriptions.set(client, new Map());
|
||
// ws 客户端/浏览器按 RFC 自动回传输层 pong——这里只记录存活。
|
||
client.on('pong', () => this.alive.set(client, true));
|
||
}
|
||
|
||
handleDisconnect(client: WebSocket): void {
|
||
const subs = this.subscriptions.get(client);
|
||
if (subs !== undefined) {
|
||
for (const albumId of subs.values()) {
|
||
this.leaveRoom(client, albumId);
|
||
}
|
||
this.subscriptions.delete(client);
|
||
}
|
||
this.alive.delete(client);
|
||
}
|
||
|
||
/** 订阅相册房间（逐相册校验公开性；失败回 error 帧，连接保持）。 */
|
||
@SubscribeMessage('subscribe')
|
||
async onSubscribe(
|
||
client: WebSocket,
|
||
payload: SubscribeFrame,
|
||
): Promise<RealtimeErrorMessage | void> {
|
||
const token = (payload as SubscribeFrame | undefined)?.albumToken;
|
||
if (typeof token !== 'string' || token === '') {
|
||
return {
|
||
type: 'error',
|
||
code: ErrorCodes.VALIDATION_FAILED,
|
||
message: 'albumToken 缺失',
|
||
};
|
||
}
|
||
try {
|
||
const album = await this.prisma.albums_ov.findUnique({
|
||
where: { publicToken: token },
|
||
});
|
||
if (album === null || !isAlbumVisibleOv(album)) {
|
||
return { type: 'error', code: ErrorCodes.ALBUM_NOT_PUBLIC };
|
||
}
|
||
this.joinRoom(client, album.albumId, token);
|
||
} catch (err) {
|
||
this.logger.error(
|
||
`subscribe 校验失败 token=${token}: ${
|
||
err instanceof Error ? err.message : String(err)
|
||
}`,
|
||
);
|
||
return { type: 'error', code: ErrorCodes.INTERNAL_ERROR };
|
||
}
|
||
}
|
||
|
||
/** 退订（幂等：未订阅的 token 静默忽略）。 */
|
||
@SubscribeMessage('unsubscribe')
|
||
onUnsubscribe(
|
||
client: WebSocket,
|
||
payload: UnsubscribeFrame,
|
||
): void {
|
||
const token = (payload as UnsubscribeFrame | undefined)?.albumToken;
|
||
if (typeof token !== 'string') {
|
||
return;
|
||
}
|
||
const albumId = this.subscriptions.get(client)?.get(token);
|
||
if (albumId !== undefined) {
|
||
this.subscriptions.get(client)?.delete(token);
|
||
this.leaveRoom(client, albumId);
|
||
}
|
||
}
|
||
|
||
/** 应用层心跳：回显 ts。 */
|
||
@SubscribeMessage('ping')
|
||
onPing(
|
||
client: WebSocket,
|
||
payload: { ts?: unknown },
|
||
): PongMessage {
|
||
void client;
|
||
return { type: 'pong', ts: typeof payload?.ts === 'number' ? payload.ts : Date.now() };
|
||
}
|
||
|
||
/** 点赞批次提交 → 合并窗聚合（LIKE_BURST_MERGE_MS，惰性 env）。 */
|
||
@OnEvent(OV_LIKE_BATCH_COMMITTED)
|
||
onLikeBatchCommitted(event: LikeBatchCommittedEvent): void {
|
||
const mergeMs = likeBurstMergeMs();
|
||
if (mergeMs <= 0) {
|
||
this.broadcastBurst(
|
||
event.albumId,
|
||
event.photoId,
|
||
event.delta,
|
||
event.total,
|
||
event.countVersion,
|
||
[event.batchId],
|
||
);
|
||
return;
|
||
}
|
||
let pending = this.bursts.get(event.photoId);
|
||
if (pending === undefined) {
|
||
pending = {
|
||
albumId: event.albumId,
|
||
delta: 0,
|
||
total: event.total,
|
||
countVersion: event.countVersion,
|
||
originBatchIds: [],
|
||
timer: setTimeout(() => this.flushBurst(event.photoId), mergeMs),
|
||
};
|
||
this.bursts.set(event.photoId, pending);
|
||
}
|
||
pending.delta += event.delta;
|
||
// 权威值取合并窗内最新提交（后到覆盖先到——countVersion 单调）。
|
||
pending.total = event.total;
|
||
pending.countVersion = event.countVersion;
|
||
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 {
|
||
const base = { photoId: event.photoId, albumId: event.albumId };
|
||
let message: ServerMessage;
|
||
if (event.kind === 'published') {
|
||
message = {
|
||
type: 'photo.published',
|
||
...base,
|
||
publishedAt: event.publishedAt ?? new Date(0).toISOString(),
|
||
entityVersion: event.entityVersion,
|
||
};
|
||
} else if (event.kind === 'hidden') {
|
||
message = { type: 'photo.hidden', ...base, entityVersion: event.entityVersion };
|
||
} else {
|
||
message = { type: 'photo.deleted', ...base, entityVersion: event.entityVersion };
|
||
}
|
||
this.sendToRoom(event.albumId, message);
|
||
}
|
||
|
||
private flushBurst(photoId: string): void {
|
||
const pending = this.bursts.get(photoId);
|
||
if (pending === undefined) {
|
||
return;
|
||
}
|
||
this.bursts.delete(photoId);
|
||
this.broadcastBurst(
|
||
pending.albumId,
|
||
photoId,
|
||
pending.delta,
|
||
pending.total,
|
||
pending.countVersion,
|
||
pending.originBatchIds,
|
||
);
|
||
}
|
||
|
||
private broadcastBurst(
|
||
albumId: string,
|
||
photoId: string,
|
||
delta: number,
|
||
total: number,
|
||
countVersion: number,
|
||
originBatchIds: string[],
|
||
): void {
|
||
this.sendToRoom(albumId, {
|
||
type: 'likes.burst',
|
||
photoId,
|
||
albumId,
|
||
delta,
|
||
total,
|
||
countVersion,
|
||
burstId: randomUUID(),
|
||
originBatchIds,
|
||
});
|
||
}
|
||
|
||
/** 房间扇出（跳过中途断开的连接；无房间 = 无人订阅，自然 no-op）。 */
|
||
private sendToRoom(albumId: string, message: ServerMessage): void {
|
||
const room = this.rooms.get(roomKey(albumId));
|
||
if (room === undefined) {
|
||
return;
|
||
}
|
||
const raw = JSON.stringify(message);
|
||
for (const client of room) {
|
||
if (client.readyState === WebSocket.OPEN) {
|
||
client.send(raw);
|
||
}
|
||
}
|
||
}
|
||
|
||
private joinRoom(client: WebSocket, albumId: string, token: string): void {
|
||
const key = roomKey(albumId);
|
||
let room = this.rooms.get(key);
|
||
if (room === undefined) {
|
||
room = new Set();
|
||
this.rooms.set(key, room);
|
||
}
|
||
room.add(client);
|
||
this.subscriptions.get(client)?.set(token, albumId);
|
||
}
|
||
|
||
private leaveRoom(client: WebSocket, albumId: string): void {
|
||
const key = roomKey(albumId);
|
||
const room = this.rooms.get(key);
|
||
if (room === undefined) {
|
||
return;
|
||
}
|
||
room.delete(client);
|
||
if (room.size === 0) {
|
||
this.rooms.delete(key);
|
||
}
|
||
}
|
||
}
|