85b2da3db5
P2 Task 7(点赞批次端点 + ov 实时网关):
- interactions 模块(仅 ov 装配):POST /public/albums/{token}/photos/
{photoId}/likes——相册不可见/照片未发布 404、delta 1..20 与
sessionId/batchId 1..64 校验 400;单事务 like_batches 唯一插入
(P2002=幂等重试,batchAccepted=true 不重加)→ photos_ov.likeTotal/
countVersion 原子自增回读 → outbox_stat_events(kind=likes) 兜底;
Redis ov(REDIS_URL_OV)INCRBY+TTL 会话/IP 双窗口限流(默认
500/2000,超限 429 RATE_LIMITED),幂等重试先于限流不耗额度
- 过期批次重放边界(24h 清理后的 batchId 重插会再次计入):接受该
边界——无永久账本可查(保留全量批次行违背清理策略),会话/IP 窗口
上限即重放天然封顶(同一 sessionId 重放挤占自身额度),偏差已在
likes.service 头注释登记,Task 8 统计侧按清理后语义处理
- realtime:ContractWsAdapter(契约 {type,...} 帧分发,ws 与 HTTP 同
端口挂 /v1/realtime)+ RealtimeGateway 相册房间模型——subscribe 逐
相册校验 isAlbumVisibleOv,失败回 error 帧且连接保持;应用层 ping→
pong 回显 ts;传输层 HEARTBEAT_MS 周期 ping 踢死连接;点赞提交经
EventEmitter2(单进程,多实例 Redis pub/sub 登记 P5)触发
likes.burst 500ms 合并窗(delta 聚合、originBatchIds 供客户端过滤
自身批次);sync-inbox 应用事务提交后发 OV_PHOTO_APPLIED →
photo.published/hidden/deleted 按房间广播(stale/duplicate/暂存/
缺席即撤下不转发)
- 契约:REALTIME_WS_PATH、ServerMessage 增 error 帧(穷尽锁 7 成员
+ 类型测试)、LIKE_SESSION_WINDOW_DELTA/LIKE_IP_WINDOW_DELTA/
LIKE_BURST_MERGE_MS 常量与 .env.example
- e2e(likes-realtime.e2e-spec.ts,6 用例):A09 三批累加/幂等重试/
outbox 行、小额度 429、ping/pong+幽灵 token error 帧连接保持、
A10 双房间隔离(先证非空调订)、合并窗单 burst(delta 求和+双
originBatchIds)、cn 隐藏→房间收 photo.hidden+隐藏后点赞 404;
凡进程内起 AppOvModule 的 spec(sync×2/public-ov/emergency)补装
ContractWsAdapter 与生产 main-ov 一致——api e2e 168(162+6)
Co-Authored-By: Claude Code <noreply@anthropic.com>
41 lines
1.8 KiB
TypeScript
41 lines
1.8 KiB
TypeScript
// services/api/src/realtime/contract-ws.adapter.ts —— 契约帧协议的 WS 适配器
|
||
// (P2 Task 7)。@nestjs/platform-ws 的 WsAdapter 只分发 {event,data} 信封
|
||
// 帧,而 P2 契约(@photo/contracts realtime.ts)的客户端消息是 {type,...}
|
||
// 判别联合——本子类仅覆写 bindMessageHandler:按 frame.type 查处理映射,
|
||
// 整帧作为载荷传入网关方法(@SubscribeMessage(type)),其余(同端口 HTTP
|
||
// 挂载、upgrade 路由、关闭语义)全部继承 WsAdapter。非法 JSON / 非对象帧 /
|
||
// 未知 type 静默丢弃(EMPTY)——协议层 error 帧只对「可理解但被拒绝」的
|
||
// 消息(如 subscribe 校验失败)由网关显式返回。
|
||
import { WsAdapter } from '@nestjs/platform-ws';
|
||
import type { MessageMappingProperties } from '@nestjs/websockets/gateway-metadata-explorer';
|
||
import { EMPTY, type Observable } from 'rxjs';
|
||
|
||
export class ContractWsAdapter extends WsAdapter {
|
||
override bindMessageHandler(
|
||
buffer: { data?: unknown },
|
||
handlersMap: Map<string, MessageMappingProperties>,
|
||
transform: (data: unknown) => Observable<unknown>,
|
||
): Observable<unknown> {
|
||
let frame: unknown;
|
||
try {
|
||
frame = JSON.parse(String(buffer.data));
|
||
} catch {
|
||
return EMPTY;
|
||
}
|
||
if (typeof frame !== 'object' || frame === null) {
|
||
return EMPTY;
|
||
}
|
||
const { type } = frame as { type?: unknown };
|
||
if (typeof type !== 'string') {
|
||
return EMPTY;
|
||
}
|
||
const handler = handlersMap.get(type);
|
||
if (handler === undefined) {
|
||
return EMPTY;
|
||
}
|
||
// 整帧作为载荷(网关方法按契约类型解构);返回值经 transform 送回客户端
|
||
// (返回 undefined/void 时上层 filter 掉,不发送——无应答语义的消息用此)。
|
||
return transform(handler.callback(frame, type));
|
||
}
|
||
}
|