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>
263 lines
8.9 KiB
TypeScript
263 lines
8.9 KiB
TypeScript
// services/api/test/emergency.e2e-spec.ts —— P2 Task 5 ov 紧急停用通道 E2E。
|
||
// 进程内起 ov 入口(AppOvModule + rawBody),审计 logger 覆写为内存目的地
|
||
// 的 pino 实例(断言 audit=emergency 字段结构);相册副本直接落 ov 库
|
||
// (不依赖 cn 链路——本通道纯 ov 本地)。覆盖:
|
||
// · disable by albumToken → 200 + albums_ov.emergencyDisabled=true + 审计行;
|
||
// · disable by albumId 幂等(重复返回同结果,审计 alreadyDisabled=true);
|
||
// · enable by albumId 恢复(显式唯一通道);
|
||
// · HMAC 缺头/错签 → 401 SYNC_AUTH_FAILED(与同步端点同一守卫);
|
||
// · albumToken/albumId 均缺省 → 400 VALIDATION_FAILED;未知相册 → 404
|
||
// ALBUM_NOT_FOUND;
|
||
// · 同步事件不得解除停用:album.updated 直推 content-batches(更高版本)
|
||
// → 副本其余字段更新、emergencyDisabled 保持 true。
|
||
import { Test } from '@nestjs/testing';
|
||
import { type NestExpressApplication } from '@nestjs/platform-express';
|
||
import { pino } from 'pino';
|
||
import { PrismaClient as OvPrismaClient } from '@photo/db-ov';
|
||
import { ErrorCodes, signSyncBody } from '@photo/contracts';
|
||
import { AppOvModule } from '../src/app-ov.module';
|
||
import { configureApp } from '../src/shared/configure-app';
|
||
// Task 7 起 AppOvModule 含 realtime 网关:进程内 ov 入口须与生产 main-ov
|
||
// 一致安装契约 WS 适配器(否则 Nest 落到默认 socket.io 驱动 → listen 即报错)。
|
||
import { ContractWsAdapter } from '../src/realtime/contract-ws.adapter';
|
||
import { EMERGENCY_AUDIT_LOGGER } from '../src/emergency/emergency.provider';
|
||
import { resetOvDb } from './reset-ov';
|
||
|
||
jest.setTimeout(60_000);
|
||
|
||
const SYNC_KEY = process.env.SYNC_INTERNAL_KEY ?? 'dev-sync-secret-change-me';
|
||
const ALBUM_ID = 'e2e-emergency-album';
|
||
const PUBLIC_TOKEN = 'emergency-token-01';
|
||
|
||
interface SignOverrides {
|
||
signature?: string;
|
||
omitSignature?: boolean;
|
||
omitTimestamp?: boolean;
|
||
}
|
||
|
||
async function signedPost(
|
||
base: string,
|
||
path: string,
|
||
body: unknown,
|
||
overrides: SignOverrides = {},
|
||
): Promise<Response> {
|
||
const raw = JSON.stringify(body);
|
||
const headers: Record<string, string> = { 'content-type': 'application/json' };
|
||
if (!overrides.omitTimestamp) {
|
||
headers['x-sync-timestamp'] = String(Math.floor(Date.now() / 1000));
|
||
}
|
||
if (!overrides.omitSignature) {
|
||
headers['x-sync-signature'] =
|
||
overrides.signature ?? signSyncBody(SYNC_KEY, raw);
|
||
}
|
||
return fetch(`${base}${path}`, { method: 'POST', headers, body: raw });
|
||
}
|
||
|
||
describe('ov emergency disable channel (e2e)', () => {
|
||
let ovApp: NestExpressApplication;
|
||
let ovBase: string;
|
||
let prismaOv: OvPrismaClient;
|
||
let auditLines: string[];
|
||
|
||
beforeAll(async () => {
|
||
expect(SYNC_KEY).not.toBe('');
|
||
await resetOvDb();
|
||
|
||
auditLines = [];
|
||
// 与默认 provider 同构(child 携带 audit=emergency 常驻字段),仅输出
|
||
// 目的地换成内存收集。
|
||
const auditLogger = pino(
|
||
{ level: 'info' },
|
||
{ write: (line: string) => void auditLines.push(line) },
|
||
).child({ audit: 'emergency' });
|
||
// overrideProvider 在 TestingModule 上(NestFactory 产物没有)——审计
|
||
// logger 以内存目的地实例覆写,断言 audit=emergency 字段。
|
||
const moduleRef = await Test.createTestingModule({
|
||
imports: [AppOvModule],
|
||
})
|
||
.overrideProvider(EMERGENCY_AUDIT_LOGGER)
|
||
.useValue(auditLogger)
|
||
.compile();
|
||
ovApp =
|
||
moduleRef.createNestApplication<NestExpressApplication>({
|
||
rawBody: true,
|
||
logger: false,
|
||
});
|
||
ovApp.useWebSocketAdapter(new ContractWsAdapter(ovApp));
|
||
configureApp(ovApp);
|
||
await ovApp.listen(0);
|
||
ovBase = await ovApp.getUrl();
|
||
|
||
prismaOv = new OvPrismaClient();
|
||
await prismaOv.albums_ov.create({
|
||
data: {
|
||
albumId: ALBUM_ID,
|
||
publicToken: PUBLIC_TOKEN,
|
||
status: 'active',
|
||
publicEnabled: true,
|
||
downloadEnabled: true,
|
||
originalDownloadEnabled: false,
|
||
emergencyDisabled: false,
|
||
translations: [],
|
||
categories: [],
|
||
defaultLocale: 'zh-CN',
|
||
timezone: 'Asia/Shanghai',
|
||
startAt: new Date('2026-10-01T10:00:00.000Z'),
|
||
endAt: null,
|
||
entityVersion: 1,
|
||
},
|
||
});
|
||
});
|
||
|
||
afterAll(async () => {
|
||
await ovApp.close();
|
||
await prismaOv.$disconnect();
|
||
});
|
||
|
||
it('未签名请求 → 401 SYNC_AUTH_FAILED(与同步端点同一 HMAC 守卫)', async () => {
|
||
const res = await signedPost(
|
||
ovBase,
|
||
'/v1/internal/emergency/disable',
|
||
{ albumToken: PUBLIC_TOKEN, reason: 'x' },
|
||
{ omitSignature: true, omitTimestamp: true },
|
||
);
|
||
expect(res.status).toBe(401);
|
||
expect(((await res.json()) as { code: string }).code).toBe(
|
||
ErrorCodes.SYNC_AUTH_FAILED,
|
||
);
|
||
expect(
|
||
(
|
||
await prismaOv.albums_ov.findUniqueOrThrow({
|
||
where: { albumId: ALBUM_ID },
|
||
})
|
||
).emergencyDisabled,
|
||
).toBe(false);
|
||
});
|
||
|
||
it('disable by albumToken → 200 + 置位 + 审计行(audit=emergency)', async () => {
|
||
const res = await signedPost(ovBase, '/v1/internal/emergency/disable', {
|
||
albumToken: PUBLIC_TOKEN,
|
||
reason: 'content review',
|
||
});
|
||
expect(res.status).toBe(200);
|
||
expect(await res.json()).toEqual({ disabled: true, albumId: ALBUM_ID });
|
||
expect(
|
||
(
|
||
await prismaOv.albums_ov.findUniqueOrThrow({
|
||
where: { albumId: ALBUM_ID },
|
||
})
|
||
).emergencyDisabled,
|
||
).toBe(true);
|
||
|
||
expect(auditLines.length).toBe(1);
|
||
const audit = JSON.parse(auditLines[0]) as Record<string, unknown>;
|
||
expect(audit).toMatchObject({
|
||
audit: 'emergency',
|
||
action: 'disable',
|
||
albumId: ALBUM_ID,
|
||
reason: 'content review',
|
||
alreadyDisabled: false,
|
||
});
|
||
});
|
||
|
||
it('disable by albumId 幂等:重复停用返回同结果,审计 alreadyDisabled=true', async () => {
|
||
const res = await signedPost(ovBase, '/v1/internal/emergency/disable', {
|
||
albumId: ALBUM_ID,
|
||
reason: 'repeat',
|
||
});
|
||
expect(res.status).toBe(200);
|
||
expect(await res.json()).toEqual({ disabled: true, albumId: ALBUM_ID });
|
||
|
||
expect(auditLines.length).toBe(2);
|
||
expect(JSON.parse(auditLines[1])).toMatchObject({
|
||
audit: 'emergency',
|
||
action: 'disable',
|
||
alreadyDisabled: true,
|
||
});
|
||
});
|
||
|
||
it('同步事件不得解除停用:更高版本 album.updated 直推 → 标志保持 true', async () => {
|
||
const res = await signedPost(ovBase, '/v1/internal/sync/content-batches', {
|
||
batch: [
|
||
{
|
||
eventId: 'e2e-emergency-sync-1',
|
||
schemaVersion: 1,
|
||
entityType: 'album',
|
||
entityId: ALBUM_ID,
|
||
albumId: ALBUM_ID,
|
||
entityVersion: 2,
|
||
tombstone: false,
|
||
priority: 'normal',
|
||
payload: {
|
||
albumId: ALBUM_ID,
|
||
publicToken: PUBLIC_TOKEN,
|
||
status: 'active',
|
||
uploadEnabled: true,
|
||
publicEnabled: true,
|
||
downloadEnabled: true,
|
||
originalDownloadEnabled: false,
|
||
defaultLocale: 'zh-CN',
|
||
timezone: 'Asia/Shanghai',
|
||
startAt: '2026-10-01T10:00:00.000Z',
|
||
endAt: null,
|
||
translations: [],
|
||
entityVersion: 2,
|
||
},
|
||
},
|
||
],
|
||
});
|
||
expect(res.status).toBe(200);
|
||
// 副本推进到 v2(其余字段照常同步),紧急停用位不受任何同步事件影响。
|
||
expect(
|
||
await prismaOv.albums_ov.findUniqueOrThrow({
|
||
where: { albumId: ALBUM_ID },
|
||
}),
|
||
).toMatchObject({ entityVersion: 2, emergencyDisabled: true });
|
||
});
|
||
|
||
it('enable by albumId → 显式恢复(唯一解除通道)', async () => {
|
||
const res = await signedPost(ovBase, '/v1/internal/emergency/enable', {
|
||
albumId: ALBUM_ID,
|
||
});
|
||
expect(res.status).toBe(200);
|
||
expect(await res.json()).toEqual({ disabled: false, albumId: ALBUM_ID });
|
||
expect(
|
||
(
|
||
await prismaOv.albums_ov.findUniqueOrThrow({
|
||
where: { albumId: ALBUM_ID },
|
||
})
|
||
).emergencyDisabled,
|
||
).toBe(false);
|
||
|
||
const enableAudit = auditLines
|
||
.map((line) => JSON.parse(line) as Record<string, unknown>)
|
||
.find((line) => line.action === 'enable');
|
||
expect(enableAudit).toMatchObject({
|
||
audit: 'emergency',
|
||
albumId: ALBUM_ID,
|
||
wasDisabled: true,
|
||
});
|
||
});
|
||
|
||
it('albumToken 与 albumId 均缺省 → 400 VALIDATION_FAILED', async () => {
|
||
const res = await signedPost(ovBase, '/v1/internal/emergency/disable', {
|
||
reason: 'no target',
|
||
});
|
||
expect(res.status).toBe(400);
|
||
expect(((await res.json()) as { code: string }).code).toBe(
|
||
ErrorCodes.VALIDATION_FAILED,
|
||
);
|
||
});
|
||
|
||
it('未知相册 → 404 ALBUM_NOT_FOUND', async () => {
|
||
const res = await signedPost(ovBase, '/v1/internal/emergency/disable', {
|
||
albumToken: 'never-synced-token',
|
||
reason: 'x',
|
||
});
|
||
expect(res.status).toBe(404);
|
||
expect(((await res.json()) as { code: string }).code).toBe(
|
||
ErrorCodes.ALBUM_NOT_FOUND,
|
||
);
|
||
});
|
||
});
|