diff --git a/.env.example b/.env.example index 57363bd..745b986 100644 --- a/.env.example +++ b/.env.example @@ -48,4 +48,10 @@ HOT_SORT_TTL_MS=30000 STAT_RETURN_INTERVAL_MS=60000 VISIT_WINDOW_MINUTES=30 LIKE_BATCH_WINDOW_HOURS=24 +# P2 Task 7：点赞基础限速（Redis ov 滑窗，窗口=LIKE_BATCH_WINDOW_HOURS）与 +# likes.burst 实时合并窗口——sessionId 窗口累计 delta 上限 / 来源 IP 上限 / +# 同照片批次合并毫秒数（0=不合并）。默认值见 packages/contracts/src/constants.ts。 +LIKE_SESSION_WINDOW_DELTA=500 +LIKE_IP_WINDOW_DELTA=2000 +LIKE_BURST_MERGE_MS=500 VISIT_REQUEST_TTL_MINUTES=35 diff --git a/packages/contracts/src/__tests__/realtime.test.ts b/packages/contracts/src/__tests__/realtime.test.ts index 764aa11..8e8be4a 100644 --- a/packages/contracts/src/__tests__/realtime.test.ts +++ b/packages/contracts/src/__tests__/realtime.test.ts @@ -63,6 +63,16 @@ describe('ServerMessage', () => { expect(pong.type).toBe('pong'); }); + it('error 帧携带域码（subscribe 校验失败等场景）', () => { + const error: ServerMessage = { type: 'error', code: 'ALBUM_NOT_PUBLIC' }; + if (error.type !== 'error') throw new Error('unreachable'); + expect(error.code).toBe('ALBUM_NOT_PUBLIC'); + // message 可选：缺省合法（客户端只应消费 code）。 + // @ts-expect-error 非 error 分支没有 code + const absent = (burst as ServerMessage).code; + expect(absent).toBeUndefined(); + }); + it('非法 type 在编译期被拒绝', () => { // @ts-expect-error type 必须严格属于联合成员 const invalid: ServerMessage = { ...burst, type: 'likes.update' }; @@ -81,7 +91,7 @@ describe('ServerMessage', () => { expect(wrongTypes.type).toBe('likes.burst'); }); - it('type 联合与 6 个消息成员 1:1（穷尽锁定）', () => { + it('type 联合与 7 个消息成员 1:1（穷尽锁定）', () => { const exhaustive: Equal< ServerMessage['type'], | 'photo.published' @@ -90,6 +100,7 @@ describe('ServerMessage', () => { | 'likes.burst' | 'stats.updated' | 'pong' + | 'error' > = true; expect(exhaustive).toBe(true); }); @@ -102,6 +113,7 @@ describe('ServerMessage', () => { { type: 'photo.deleted', photoId: 'p', albumId: 'a', entityVersion: 3 }, { type: 'stats.updated', albumId: 'a', viewCount: 99, countVersion: 9 }, { type: 'pong', ts: 1758187200000 }, + { type: 'error', code: 'ALBUM_NOT_PUBLIC', message: '相册不存在或未公开' }, ]; const kinds = messages.map((m) => { switch (m.type) { @@ -115,6 +127,8 @@ describe('ServerMessage', () => { return `stats:${m.viewCount}`; case 'pong': return `pong:${m.ts}`; + case 'error': + return `error:${m.code}`; } }); expect(kinds).toEqual([ @@ -124,6 +138,7 @@ describe('ServerMessage', () => { 'photo:3', 'stats:99', 'pong:1758187200000', + 'error:ALBUM_NOT_PUBLIC', ]); }); }); diff --git a/packages/contracts/src/constants.ts b/packages/contracts/src/constants.ts index 445adb0..43af7cd 100644 --- a/packages/contracts/src/constants.ts +++ b/packages/contracts/src/constants.ts @@ -20,5 +20,11 @@ export const STAT_RETURN_INTERVAL_MS = 60000; export const VISIT_WINDOW_MINUTES = 30; /** 点赞批次幂等窗口小时：窗口外 batchId 拒绝重放（返回权威 total，不加计）。 */ export const LIKE_BATCH_WINDOW_HOURS = 24; +/** 点赞基础限速：单 sessionId 滑窗（=幂等窗口）累计 delta 上限，超限 429。 */ +export const LIKE_SESSION_WINDOW_DELTA = 500; +/** 点赞基础限速：来源 IP 滑窗累计 delta 上限（NAT 出口共享，宽于单会话）。 */ +export const LIKE_IP_WINDOW_DELTA = 2000; +/** likes.burst 合并窗口毫秒（技术方案 §7 短窗合并；0=不合并逐批即时广播）。 */ +export const LIKE_BURST_MERGE_MS = 500; /** 浏览上报 requestId 保留分钟：过期 requestId 拒绝重放（≥ 去重窗口）。 */ export const VISIT_REQUEST_TTL_MINUTES = 35; diff --git a/packages/contracts/src/realtime.ts b/packages/contracts/src/realtime.ts index 9a71131..f6dc232 100644 --- a/packages/contracts/src/realtime.ts +++ b/packages/contracts/src/realtime.ts @@ -6,6 +6,12 @@ /** 心跳间隔（毫秒）：客户端周期发送 ping、服务端回 pong；超时无心跳的连接由服务端踢除。 */ export const HEARTBEAT_MS = 20000; +/** + * ov 入口 WS 网关挂载路径：与 HTTP 同端口（ov 8081），升级请求按本路径路由 + * （services/api realtime 网关装饰器与 e2e/viewer 客户端共用，单一来源）。 + */ +export const REALTIME_WS_PATH = '/v1/realtime'; + // ── server → client ── /** 照片发布：副本应用后向相册房间广播（ov 侧，下同）。 */ @@ -59,13 +65,26 @@ export interface PongMessage { ts: number; } +/** + * 错误帧：客户端消息无法处理时的服务端应答（如 subscribe 的相册不存在/ + * 不公开）。连接保持打开——客户端可修正后重发（如改订其它相册或等待恢复）。 + */ +export interface RealtimeErrorMessage { + type: 'error'; + /** 契约 ErrorCodes 域码（subscribe 校验失败 → ALBUM_NOT_PUBLIC 等）。 */ + code: string; + /** 可读说明（可选，不承诺稳定性，客户端只应消费 code）。 */ + message?: string; +} + export type ServerMessage = | PhotoPublishedMessage | PhotoHiddenMessage | PhotoDeletedMessage | LikesBurstMessage | StatsUpdatedMessage - | PongMessage; + | PongMessage + | RealtimeErrorMessage; // ── client → server ── diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7f5172e..ddd1f37 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -312,10 +312,19 @@ importers: version: 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) '@nestjs/core': specifier: ^10.4.6 - version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(@nestjs/websockets@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/event-emitter': + specifier: ^3.1.0 + version: 3.1.0(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22) '@nestjs/platform-express': specifier: ^10.4.6 version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22) + '@nestjs/platform-ws': + specifier: ^10.4.6 + version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@10.4.22)(rxjs@7.8.2) + '@nestjs/websockets': + specifier: ^10.4.6 + version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) '@node-rs/argon2': specifier: ^2.2.1 version: 2.2.1 @@ -358,6 +367,9 @@ importers: rxjs: specifier: ^7.8.1 version: 7.8.2 + ws: + specifier: ^8.18.0 + version: 8.21.3 devDependencies: '@aws-sdk/client-s3': specifier: ^3.700.0 @@ -374,6 +386,9 @@ importers: '@types/supertest': specifier: ^6.0.2 version: 6.0.3 + '@types/ws': + specifier: ^8.5.12 + version: 8.18.1 dotenv: specifier: ^16.4.5 version: 16.6.1 @@ -404,6 +419,9 @@ importers: '@photo/db-cn': specifier: workspace:* version: link:../../packages/db-cn + '@photo/db-ov': + specifier: workspace:* + version: link:../../packages/db-ov pino: specifier: ^10.0.0 version: 10.3.1 @@ -413,13 +431,10 @@ importers: version: 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) '@nestjs/core': specifier: ^10.4.6 - version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(@nestjs/websockets@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) '@nestjs/platform-express': specifier: ^10.4.6 version: 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22) - '@photo/db-ov': - specifier: workspace:* - version: link:../../packages/db-ov '@photo/storage': specifier: workspace:^ version: link:../../packages/storage @@ -437,7 +452,7 @@ importers: version: 16.6.1 jest: specifier: ^29.7.0 - version: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) + version: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)) sharp: specifier: ^0.35.0 version: 0.35.4(@types/node@22.20.3) @@ -446,7 +461,7 @@ importers: version: 7.2.2 ts-jest: specifier: ^29.2.5 - version: 29.4.12(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)))(typescript@5.6.3) + version: 29.4.12(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.20.3))(typescript@5.6.3) typescript: specifier: ^5.6.0 version: 5.6.3 @@ -1568,12 +1583,25 @@ packages: '@nestjs/websockets': optional: true + '@nestjs/event-emitter@3.1.0': + resolution: {integrity: sha512-DOY/4XBGyIjYyOJKkO6jl1kzFE0ZfX0wV+M2HR5NWymPT9Z0zdCEcZGxTXXkoMRwPtglnvCGJALSjOpXPIcM3g==} + peerDependencies: + '@nestjs/common': ^10.0.0 || ^11.0.0 + '@nestjs/core': ^10.0.0 || ^11.0.0 + '@nestjs/platform-express@10.4.22': resolution: {integrity: sha512-ySSq7Py/DFozzZdNDH67m/vHoeVdphDniWBnl6q5QVoXldDdrZIHLXLRMPayTDh5A95nt7jjJzmD4qpTbNQ6tA==} peerDependencies: '@nestjs/common': ^10.0.0 '@nestjs/core': ^10.0.0 + '@nestjs/platform-ws@10.4.22': + resolution: {integrity: sha512-ZBL66p8axCyvQw6lP6R5uMAamVGfDb0/LtbdxDjMjbWb5/wi070P0MWrjzTudEA3ThsDMNOsfawZlsFUkSfCzg==} + peerDependencies: + '@nestjs/common': ^10.0.0 + '@nestjs/websockets': ^10.0.0 + rxjs: ^7.1.0 + '@nestjs/testing@10.4.22': resolution: {integrity: sha512-HO9aPus3bAedAC+jKVAA8jTdaj4fs5M9fing4giHrcYV2txe9CvC1l1WAjwQ9RDhEHdugjY4y+FZA/U/YqPZrA==} peerDependencies: @@ -1587,6 +1615,18 @@ packages: '@nestjs/platform-express': optional: true + '@nestjs/websockets@10.4.22': + resolution: {integrity: sha512-OLd4i0Faq7vgdtB5vVUrJ54hWEtcXy9poJ6n7kbbh/5ms+KffUl+wwGsbe7uSXLrkoyI8xXU6fZPkFArI+XiRg==} + peerDependencies: + '@nestjs/common': ^10.0.0 + '@nestjs/core': ^10.0.0 + '@nestjs/platform-socket.io': ^10.0.0 + reflect-metadata: ^0.1.12 || ^0.2.0 + rxjs: ^7.1.0 + peerDependenciesMeta: + '@nestjs/platform-socket.io': + optional: true + '@noble/hashes@1.8.0': resolution: {integrity: sha512-jCs9ldd7NwzpgXDIf6P3+NrHh9/sD6CQdxHyjQI+h/6rDNo88ypBxxz45UDuZHz9r3tNz7N/VInSVoVdtXEI4A==} engines: {node: ^14.21.3 || >=16} @@ -2306,6 +2346,9 @@ packages: '@types/web-bluetooth@0.0.21': resolution: {integrity: sha512-oIQLCGWtcFZy2JW77j9k8nHzAOpqMHLQejDA48XXMWH6tjCQHz5RCFz1bzsmROyL6PUm+LLnUiI4BCn221inxA==} + '@types/ws@8.18.1': + resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==} + '@types/yargs-parser@21.0.3': resolution: {integrity: sha512-I4q9QU9MQv4oEOz4tAHJtNz1cwuLxn2F3xcc2iV5WdqLPpUnj30aUuxt1mAxYTG+oe8CZMV/+6rU4S4gRDzqtQ==} @@ -3386,6 +3429,9 @@ packages: resolution: {integrity: sha512-i/2XbnSz/uxRCU6+NdVJgKWDTM427+MqYbkQzD321DuCQJUqOuJKIA0IM2+W2xtYHdKOmZ4dR6fExsd4SXL+WQ==} engines: {node: '>=6'} + eventemitter2@6.4.9: + resolution: {integrity: sha512-JEPTiaOt9f04oa6NOkc4aH+nVp5I3wEjpHbIPqfgCdD5v5bUzy7xQqwcVO2aDQgOWhI28da57HksMrzK9HlRxg==} + eventemitter3@5.0.4: resolution: {integrity: sha512-mlsTRyGaPBjPedk6Bvw+aqbsXDtoAyAzm5MO7JgU+yVRyMQ5O8bD4Kcci7BS85f93veegeCPkL8R4GLClnjLFw==} @@ -4443,6 +4489,10 @@ packages: resolution: {integrity: sha512-rJgTQnkUnH1sFw8yT6VSU3zD3sWmu6sZhIseY8VX+GRu3P6F7Fu+JNDoXfklElbLJSnc3FUQHVe4cU5hj+BcUg==} engines: {node: '>=0.10.0'} + object-hash@3.0.0: + resolution: {integrity: sha512-RSn9F68PjH9HqtltsSnqYC1XXoWe9Bju5+213R98cNGttag9q9yAOTzdbsqvIa7aNm5WffBZFpWYr2aWrklWAw==} + engines: {node: '>= 6'} + object-inspect@1.13.4: resolution: {integrity: sha512-W67iLl4J2EXEGTbfeHCffrjDfitvLANg0UlX3wFUUSTx92KXRFegMHUVgSqE+wvhAbi4WqjGg9czysTV2Epbew==} engines: {node: '>= 0.4'} @@ -5787,6 +5837,18 @@ packages: utf-8-validate: optional: true + ws@8.18.0: + resolution: {integrity: sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw==} + engines: {node: '>=10.0.0'} + peerDependencies: + bufferutil: ^4.0.1 + utf-8-validate: '>=5.0.2' + peerDependenciesMeta: + bufferutil: + optional: true + utf-8-validate: + optional: true + ws@8.21.3: resolution: {integrity: sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==} engines: {node: '>=10.0.0'} @@ -6721,41 +6783,6 @@ snapshots: jest-util: 29.7.0 slash: 3.0.0 - '@jest/core@29.7.0(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3))': - dependencies: - '@jest/console': 29.7.0 - '@jest/reporters': 29.7.0 - '@jest/test-result': 29.7.0 - '@jest/transform': 29.7.0 - '@jest/types': 29.6.3 - '@types/node': 24.13.5 - ansi-escapes: 4.3.2 - chalk: 4.1.2 - ci-info: 3.9.0 - exit: 0.1.2 - graceful-fs: 4.2.11 - jest-changed-files: 29.7.0 - jest-config: 29.7.0(@types/node@24.13.5)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - jest-haste-map: 29.7.0 - jest-message-util: 29.7.0 - jest-regex-util: 29.6.3 - jest-resolve: 29.7.0 - jest-resolve-dependencies: 29.7.0 - jest-runner: 29.7.0 - jest-runtime: 29.7.0 - jest-snapshot: 29.7.0 - jest-util: 29.7.0 - jest-validate: 29.7.0 - jest-watcher: 29.7.0 - micromatch: 4.0.8 - pretty-format: 29.7.0 - slash: 3.0.0 - strip-ansi: 6.0.1 - transitivePeerDependencies: - - babel-plugin-macros - - supports-color - - ts-node - '@jest/core@29.7.0(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3))': dependencies: '@jest/console': 29.7.0 @@ -6983,7 +7010,7 @@ snapshots: transitivePeerDependencies: - supports-color - '@nestjs/core@10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2)': + '@nestjs/core@10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(@nestjs/websockets@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2)': dependencies: '@nestjs/common': 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) '@nuxtjs/opencollective': 0.3.2 @@ -6996,13 +7023,20 @@ snapshots: uid: 2.0.2 optionalDependencies: '@nestjs/platform-express': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22) + '@nestjs/websockets': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) transitivePeerDependencies: - encoding + '@nestjs/event-emitter@3.1.0(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22)': + dependencies: + '@nestjs/common': 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/core': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(@nestjs/websockets@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + eventemitter2: 6.4.9 + '@nestjs/platform-express@10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22)': dependencies: '@nestjs/common': 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) - '@nestjs/core': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/core': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(@nestjs/websockets@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) body-parser: 1.20.4 cors: 2.8.5 express: 4.22.1 @@ -7011,14 +7045,35 @@ snapshots: transitivePeerDependencies: - supports-color + '@nestjs/platform-ws@10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@10.4.22)(rxjs@7.8.2)': + dependencies: + '@nestjs/common': 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/websockets': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + rxjs: 7.8.2 + tslib: 2.8.1 + ws: 8.18.0 + transitivePeerDependencies: + - bufferutil + - utf-8-validate + '@nestjs/testing@10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22)(@nestjs/platform-express@10.4.22)': dependencies: '@nestjs/common': 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) - '@nestjs/core': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/core': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(@nestjs/websockets@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) tslib: 2.8.1 optionalDependencies: '@nestjs/platform-express': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22) + '@nestjs/websockets@10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2)': + dependencies: + '@nestjs/common': 10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/core': 10.4.22(@nestjs/common@10.4.22(class-transformer@0.5.1)(class-validator@0.15.1)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@10.4.22)(@nestjs/websockets@10.4.22)(reflect-metadata@0.2.2)(rxjs@7.8.2) + iterare: 1.2.1 + object-hash: 3.0.0 + reflect-metadata: 0.2.2 + rxjs: 7.8.2 + tslib: 2.8.1 + '@noble/hashes@1.8.0': {} '@nodable/entities@3.0.0': {} @@ -7688,6 +7743,10 @@ snapshots: '@types/web-bluetooth@0.0.21': {} + '@types/ws@8.18.1': + dependencies: + '@types/node': 24.13.5 + '@types/yargs-parser@21.0.3': {} '@types/yargs@17.0.35': @@ -8672,21 +8731,6 @@ snapshots: object-assign: 4.1.1 vary: 1.1.2 - create-jest@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)): - dependencies: - '@jest/types': 29.6.3 - chalk: 4.1.2 - exit: 0.1.2 - graceful-fs: 4.2.11 - jest-config: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - jest-util: 29.7.0 - prompts: 2.4.2 - transitivePeerDependencies: - - '@types/node' - - babel-plugin-macros - - supports-color - - ts-node - create-jest@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)): dependencies: '@jest/types': 29.6.3 @@ -9072,6 +9116,8 @@ snapshots: event-target-shim@5.0.1: {} + eventemitter2@6.4.9: {} + eventemitter3@5.0.4: {} execa@5.1.1: @@ -9578,25 +9624,6 @@ snapshots: - babel-plugin-macros - supports-color - jest-cli@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)): - dependencies: - '@jest/core': 29.7.0(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - '@jest/test-result': 29.7.0 - '@jest/types': 29.6.3 - chalk: 4.1.2 - create-jest: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - exit: 0.1.2 - import-local: 3.2.0 - jest-config: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - jest-util: 29.7.0 - jest-validate: 29.7.0 - yargs: 17.7.3 - transitivePeerDependencies: - - '@types/node' - - babel-plugin-macros - - supports-color - - ts-node - jest-cli@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)): dependencies: '@jest/core': 29.7.0(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)) @@ -9616,37 +9643,6 @@ snapshots: - supports-color - ts-node - jest-config@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)): - dependencies: - '@babel/core': 7.29.7 - '@jest/test-sequencer': 29.7.0 - '@jest/types': 29.6.3 - babel-jest: 29.7.0(@babel/core@7.29.7) - chalk: 4.1.2 - ci-info: 3.9.0 - deepmerge: 4.3.1 - glob: 7.2.3 - graceful-fs: 4.2.11 - jest-circus: 29.7.0 - jest-environment-node: 29.7.0 - jest-get-type: 29.6.3 - jest-regex-util: 29.6.3 - jest-resolve: 29.7.0 - jest-runner: 29.7.0 - jest-util: 29.7.0 - jest-validate: 29.7.0 - micromatch: 4.0.8 - parse-json: 5.2.0 - pretty-format: 29.7.0 - slash: 3.0.0 - strip-json-comments: 3.1.1 - optionalDependencies: - '@types/node': 22.20.3 - ts-node: 10.9.2(@types/node@22.20.3)(typescript@5.6.3) - transitivePeerDependencies: - - babel-plugin-macros - - supports-color - jest-config@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)): dependencies: '@babel/core': 7.29.7 @@ -9678,37 +9674,6 @@ snapshots: - babel-plugin-macros - supports-color - jest-config@29.7.0(@types/node@24.13.5)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)): - dependencies: - '@babel/core': 7.29.7 - '@jest/test-sequencer': 29.7.0 - '@jest/types': 29.6.3 - babel-jest: 29.7.0(@babel/core@7.29.7) - chalk: 4.1.2 - ci-info: 3.9.0 - deepmerge: 4.3.1 - glob: 7.2.3 - graceful-fs: 4.2.11 - jest-circus: 29.7.0 - jest-environment-node: 29.7.0 - jest-get-type: 29.6.3 - jest-regex-util: 29.6.3 - jest-resolve: 29.7.0 - jest-runner: 29.7.0 - jest-util: 29.7.0 - jest-validate: 29.7.0 - micromatch: 4.0.8 - parse-json: 5.2.0 - pretty-format: 29.7.0 - slash: 3.0.0 - strip-json-comments: 3.1.1 - optionalDependencies: - '@types/node': 24.13.5 - ts-node: 10.9.2(@types/node@22.20.3)(typescript@5.6.3) - transitivePeerDependencies: - - babel-plugin-macros - - supports-color - jest-config@29.7.0(@types/node@24.13.5)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)): dependencies: '@babel/core': 7.29.7 @@ -9955,18 +9920,6 @@ snapshots: merge-stream: 2.0.0 supports-color: 8.1.1 - jest@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)): - dependencies: - '@jest/core': 29.7.0(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - '@jest/types': 29.6.3 - import-local: 3.2.0 - jest-cli: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - transitivePeerDependencies: - - '@types/node' - - babel-plugin-macros - - supports-color - - ts-node - jest@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)): dependencies: '@jest/core': 29.7.0(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)) @@ -10415,6 +10368,8 @@ snapshots: object-assign@4.1.1: {} + object-hash@3.0.0: {} + object-inspect@1.13.4: {} obug@2.2.1: {} @@ -11269,26 +11224,6 @@ snapshots: dependencies: typescript: 5.9.3 - ts-jest@29.4.12(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)))(typescript@5.6.3): - dependencies: - bs-logger: 0.2.6 - fast-json-stable-stringify: 2.1.0 - handlebars: 4.7.9 - jest: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3)) - json5: 2.2.3 - lodash.memoize: 4.1.2 - make-error: 1.3.6 - semver: 7.8.5 - type-fest: 4.41.0 - typescript: 5.6.3 - yargs-parser: 21.1.1 - optionalDependencies: - '@babel/core': 7.29.7 - '@jest/transform': 29.7.0 - '@jest/types': 29.6.3 - babel-jest: 29.7.0(@babel/core@7.29.7) - jest-util: 29.7.0 - ts-jest@29.4.12(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)))(typescript@5.9.3): dependencies: bs-logger: 0.2.6 @@ -11309,24 +11244,25 @@ snapshots: babel-jest: 29.7.0(@babel/core@7.29.7) jest-util: 29.7.0 - ts-node@10.9.2(@types/node@22.20.3)(typescript@5.6.3): + ts-jest@29.4.12(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.20.3))(typescript@5.6.3): dependencies: - '@cspotcode/source-map-support': 0.8.1 - '@tsconfig/node10': 1.0.13 - '@tsconfig/node12': 1.0.11 - '@tsconfig/node14': 1.0.3 - '@tsconfig/node16': 1.0.4 - '@types/node': 22.20.3 - acorn: 8.18.0 - acorn-walk: 8.3.5 - arg: 4.1.3 - create-require: 1.1.1 - diff: 4.0.4 + bs-logger: 0.2.6 + fast-json-stable-stringify: 2.1.0 + handlebars: 4.7.9 + jest: 29.7.0(@types/node@22.20.3)(ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3)) + json5: 2.2.3 + lodash.memoize: 4.1.2 make-error: 1.3.6 + semver: 7.8.5 + type-fest: 4.41.0 typescript: 5.6.3 - v8-compile-cache-lib: 3.0.1 - yn: 3.1.1 - optional: true + yargs-parser: 21.1.1 + optionalDependencies: + '@babel/core': 7.29.7 + '@jest/transform': 29.7.0 + '@jest/types': 29.6.3 + babel-jest: 29.7.0(@babel/core@7.29.7) + jest-util: 29.7.0 ts-node@10.9.2(@types/node@22.20.3)(typescript@5.9.3): dependencies: @@ -11838,6 +11774,8 @@ snapshots: ws@7.5.13: {} + ws@8.18.0: {} + ws@8.21.3: {} xml-name-validator@4.0.0: {} diff --git a/services/api/package.json b/services/api/package.json index 810257a..37e4075 100644 --- a/services/api/package.json +++ b/services/api/package.json @@ -17,7 +17,10 @@ "dependencies": { "@nestjs/common": "^10.4.6", "@nestjs/core": "^10.4.6", + "@nestjs/event-emitter": "^3.1.0", "@nestjs/platform-express": "^10.4.6", + "@nestjs/platform-ws": "^10.4.6", + "@nestjs/websockets": "^10.4.6", "@node-rs/argon2": "^2.2.1", "@photo/contracts": "workspace:*", "@photo/db-cn": "workspace:*", @@ -31,7 +34,8 @@ "pino": "^10.0.0", "pino-http": "^11.0.0", "reflect-metadata": "^0.2.2", - "rxjs": "^7.8.1" + "rxjs": "^7.8.1", + "ws": "^8.18.0" }, "devDependencies": { "@aws-sdk/client-s3": "^3.700.0", @@ -39,6 +43,7 @@ "@types/jest": "^29.5.13", "@types/node": "^22.10.0", "@types/supertest": "^6.0.2", + "@types/ws": "^8.5.12", "dotenv": "^16.4.5", "jest": "^29.7.0", "sharp": "^0.35.0", diff --git a/services/api/src/app-ov.module.ts b/services/api/src/app-ov.module.ts index bb44830..cf229e9 100644 --- a/services/api/src/app-ov.module.ts +++ b/services/api/src/app-ov.module.ts @@ -4,8 +4,11 @@ // ov 角色模块：cn 专属域（auth/albums/uploads/photos/admin——写路径与 MinIO // 上传）不装配；P2 Task 3 加入 sync-inbox（内容事件接收 + 副本投影）；Task 5 // 加入 emergency（海外紧急停用通道，独立于同步）；公开读自 Task 6 起装配 -// （public-ov：读 photos_ov 副本的同形契约端点 + hot 排序）；likes/visits、 -// WS 网关自 Task 7+ 起逐个加入。cn 模块保持原样，两入口互不影响。 +// （public-ov：读 photos_ov 副本的同形契约端点 + hot 排序）；Task 7 起加入 +// interactions（点赞批次幂等 + Redis ov 限速）与 realtime（WS 网关：契约帧 +// 分发见 main-ov 的 ContractWsAdapter），EventEmitter2 全局总线承接「事务提交 +// → 实时广播」（ov-events.ts，单进程——多实例 P5）。cn 模块保持原样，两入口 +// 互不影响。 import { type MiddlewareConsumer, Module, @@ -13,11 +16,14 @@ import { ValidationPipe, } from '@nestjs/common'; 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 { EmergencyModule } from './emergency/emergency.module'; import { HealthController } from './health.controller'; +import { InteractionsModule } from './interactions/interactions.module'; import { PublicOvModule } from './public-ov/public-ov.module'; +import { RealtimeModule } from './realtime/realtime.module'; import { RequestIdMiddleware } from './request-id.middleware'; import { createLoggerOptions } from './shared/logger-options'; import { SyncInboxModule } from './sync-inbox/sync-inbox.module'; @@ -25,12 +31,19 @@ import { SyncInboxModule } from './sync-inbox/sync-inbox.module'; @Module({ imports: [ LoggerModule.forRoot(createLoggerOptions()), + // 进程内事件总线（Task 7）：interactions/sync-inbox 发布 → realtime 网关 + // 消费。forRoot 缺省 global——子模块直接注入 EventEmitter2。 + EventEmitterModule.forRoot(), // ov 内容事件收件箱（P2 Task 3）：POST /v1/internal/sync/content-batches。 SyncInboxModule, // 海外紧急停用通道（P2 Task 5）：POST /v1/internal/emergency/disable|enable。 EmergencyModule, // ov 公开读（P2 Task 6）：/public/albums/* 读 ov 副本（含 sort=hot 快照）。 PublicOvModule, + // ov 公开互动（P2 Task 7）：POST /public/albums/*/photos/*/likes。 + InteractionsModule, + // ov 实时广播（P2 Task 7）：WS /v1/realtime（适配器见 main-ov）。 + RealtimeModule, ], controllers: [HealthController], providers: [ diff --git a/services/api/src/interactions/dto.ts b/services/api/src/interactions/dto.ts new file mode 100644 index 0000000..c148b0f --- /dev/null +++ b/services/api/src/interactions/dto.ts @@ -0,0 +1,24 @@ +// 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，字符串数字不静默转换——公开匿名端点从严）。 +import { IsInt, IsString, Length, Max, Min } from 'class-validator'; + +export class LikeBatchDto { + /** 会话标识（客户端随机生成、会话期间复用；限速滑窗的键之一）。 */ + @IsString() + @Length(1, 64) + sessionId!: string; + + /** 批次幂等键：同批失败重试必须复用，跨批次不得复用。 */ + @IsString() + @Length(1, 64) + batchId!: string; + + /** 本批新增点赞数（客户端短窗合并后的累计点击数）。 */ + @IsInt() + @Min(1) + @Max(20) + delta!: number; +} diff --git a/services/api/src/interactions/interactions.module.ts b/services/api/src/interactions/interactions.module.ts new file mode 100644 index 0000000..d02c9c8 --- /dev/null +++ b/services/api/src/interactions/interactions.module.ts @@ -0,0 +1,17 @@ +// services/api/src/interactions/ —— ov 公开互动模块（P2 Task 7 起）。 +// 仅 ov 入口装配（app-ov.module 显式挑选；cn 入口不感知）。Task 7 装配 +// 点赞批次端点；Task 8 起浏览上报（visits）同域加入。事件广播经 +// EventEmitter2（app 根模块 EventEmitterModule.forRoot() 全局注册）发往 +// realtime 网关，无模块间直接依赖。 +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'; + +@Module({ + imports: [OvPrismaModule, OvRedisModule], + controllers: [LikesController], + providers: [LikesService], +}) +export class InteractionsModule {} diff --git a/services/api/src/interactions/likes.controller.ts b/services/api/src/interactions/likes.controller.ts new file mode 100644 index 0000000..66331d0 --- /dev/null +++ b/services/api/src/interactions/likes.controller.ts @@ -0,0 +1,35 @@ +// services/api/src/interactions/likes.controller.ts —— 点赞批次端点 +// （P2 Task 7，ov 域）。路由与契约 likePhotoBatch 一致（tags [interactions]， +// 路径前缀不变、部署域名区分）；匿名（无 SessionGuard）。IP 取 req.ip 并 +// 归一 IPv6 映射前缀（::ffff:127.0.0.1 → 127.0.0.1），仅作限速键——不落库、 +// 不进日志（匿名观众端无 IP 留存需求）。真实部署经反代时应配置 trust proxy +// 系按部署环境处理，登记 P5。 +import { + Body, + Controller, + HttpCode, + HttpStatus, + Ip, + Param, + Post, +} from '@nestjs/common'; +import { LikeBatchDto } from './dto'; +import { LikesService, type LikeBatchResponse } from './likes.service'; + +@Controller('public/albums') +export class LikesController { + constructor(private readonly likes: LikesService) {} + + /** 契约 200（含幂等重试与过期拒绝——拒绝也 200，语义在 batchAccepted）。 */ + @HttpCode(HttpStatus.OK) + @Post(':token/photos/:photoId/likes') + likeBatch( + @Param('token') token: string, + @Param('photoId') photoId: string, + @Body() dto: LikeBatchDto, + @Ip() ip: string | undefined, + ): Promise { + const clientIp = ip === undefined ? undefined : ip.replace(/^::ffff:/, ''); + return this.likes.likeBatch(token, photoId, dto, clientIp); + } +} diff --git a/services/api/src/interactions/likes.service.ts b/services/api/src/interactions/likes.service.ts new file mode 100644 index 0000000..e998470 --- /dev/null +++ b/services/api/src/interactions/likes.service.ts @@ -0,0 +1,248 @@ +// services/api/src/interactions/likes.service.ts —— 匿名点赞批次（P2 Task 7， +// ov 域）。契约 POST /public/albums/{token}/photos/{photoId}/likes。全流程： +// +// ① 校验：相册 publicToken → albums_ov + isAlbumVisibleOv（含紧急停用） +// → 统一 404 ALBUM_NOT_PUBLIC；照片不存在/不属于该册/非 published → +// 404 PHOTO_NOT_PUBLIC；DTO 边界（1..64/1..20）由全局管道 400。 +// ② 幂等快速通道：like_batches 唯一 [photoId,sessionId,batchId] 命中既有行 +// → 返回当前权威 {total, countVersion, batchAccepted:true}，不进事务、 +// 不消耗限速预算（客户端重试同 batchId 是正常路径，不因限速误伤）。 +// ③ 基础限速（Redis ov 滑窗，见 assertRateLimit）：超限 429 RATE_LIMITED。 +// ④ 入账事务：like_batches 插入（唯一约束兜底并发重试 → P2002 走 ② 语义） +// → photos_ov.updateMany 原子 likeTotal+=delta、countVersion+=1（行锁 +// 串行化并发批次，随后同事务读回新值）→ outbox_stat_events 记权威快照 +// {total, statVersion}（Task 9 回传 cn 的「仅更高版本替换」队列）。 +// ⑤ 提交后：发布 OV_LIKE_BATCH_COMMITTED 进程内事件 → WS 网关合并广播 +// likes.burst（realtime.gateway）。 +// +// —— 过期批次重放的窗口边界（登记偏差，Task 8 清理落地后生效）—— +// like_batches 行被清理调度按 LIKE_BATCH_WINDOW_HOURS（24h）删除后，同一 +// (photoId,sessionId,batchId) 的重放会以「新批次」身份重新插入并再次加计。 +// 真正的窗口外拒绝需要**永久保留幂等账本**——与清理调度存在的理由（无界 +// 增长）直接冲突，故本实现接受该 24h 边界，缓解措施： +// · sessionId 滑窗上限 LIKE_SESSION_WINDOW_DELTA（默认 500/窗口）：重放 +// 消耗同一预算，单会话每窗口的重复加计被钳制； +// · 匿名端点本就无强身份，伪造新 sessionId 的增量滥用与重放滥用同级， +// 基础限速的定位是「误客户端缓冲」而非防滥用（真实防护归 P5 速率层）。 +// 契约中「窗口外拒绝重放」语义由 Task 8 的清理端补全：清理即拒绝点——清理 +// 后行不存在与「从未提交」在数据上不可区分（batchId 无客户端时间戳可信）， +// 这是无永久账本下的理论边界，登记偏差供验收评审裁决。 +import { HttpStatus, Injectable } from '@nestjs/common'; +import { EventEmitter2 } from '@nestjs/event-emitter'; +import { + LIKE_BATCH_WINDOW_HOURS, + LIKE_IP_WINDOW_DELTA, + LIKE_SESSION_WINDOW_DELTA, + ErrorCodes, +} from '@photo/contracts'; +import type { components } from '@photo/contracts'; +import { ApiException } from '../api-exception'; +import { albumNotPublic, photoNotPublic } 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_LIKE_BATCH_COMMITTED, + type LikeBatchCommittedEvent, +} from '../realtime/ov-events'; +import type { LikeBatchDto } from './dto'; + +export type LikeBatchResponse = components['schemas']['LikeBatchResponse']; + +/** 限速滑窗秒数（= 幂等窗口，惰性读 env 供测试调整）。 */ +export function likeWindowSeconds(): 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 * 3600; +} + +/** 单 sessionId 窗口累计 delta 上限（惰性读 env，同 hotSortTtlMs 模式）。 */ +export function likeSessionWindowDelta(): number { + const parsed = Number.parseInt( + process.env.LIKE_SESSION_WINDOW_DELTA ?? '', + 10, + ); + return Number.isFinite(parsed) && parsed >= 0 + ? parsed + : LIKE_SESSION_WINDOW_DELTA; +} + +/** 来源 IP 窗口累计 delta 上限（NAT 出口共享，宽于单会话）。 */ +export function likeIpWindowDelta(): number { + const parsed = Number.parseInt(process.env.LIKE_IP_WINDOW_DELTA ?? '', 10); + return Number.isFinite(parsed) && parsed >= 0 + ? parsed + : LIKE_IP_WINDOW_DELTA; +} + +function rateLimited(): ApiException { + return new ApiException( + ErrorCodes.RATE_LIMITED, + '点赞过于频繁，请稍后再试', + HttpStatus.TOO_MANY_REQUESTS, + ); +} + +@Injectable() +export class LikesService { + constructor( + private readonly prisma: OvPrismaService, + private readonly redis: OvRedisService, + private readonly events: EventEmitter2, + ) {} + + async likeBatch( + token: string, + photoId: string, + dto: LikeBatchDto, + clientIp: string | undefined, + ): Promise { + // ① 相册/照片公开校验（同 public-ov 读路径口径）。 + const album = await this.prisma.albums_ov.findUnique({ + where: { publicToken: token }, + }); + if (album === null || !isAlbumVisibleOv(album)) { + throw albumNotPublic(); + } + const photo = await this.prisma.photos_ov.findUnique({ + where: { photoId }, + }); + if ( + photo === null || + photo.albumId !== album.albumId || + photo.visibility !== 'published' + ) { + throw photoNotPublic(); + } + + // ② 幂等快速通道：既有批次 = 客户端重试已确认批次 → 权威值 + true。 + const existing = await this.prisma.like_batches.findUnique({ + where: { + photoId_sessionId_batchId: { + photoId, + sessionId: dto.sessionId, + batchId: dto.batchId, + }, + }, + select: { id: true }, + }); + if (existing !== null) { + return this.authoritative(photoId, true); + } + + // ③ 基础限速（通过后才进事务；拒绝也消耗预算——固定窗限速惯例）。 + await this.assertRateLimit(dto.sessionId, clientIp, dto.delta); + + // ④ 入账事务（并发同批由唯一约束兜底 → ② 语义）。 + try { + const committed = await this.prisma.$transaction(async (tx) => { + await tx.like_batches.create({ + data: { + photoId, + sessionId: dto.sessionId, + batchId: dto.batchId, + delta: dto.delta, + }, + }); + // 原子自增（行锁串行并发批次）+ 同事务读回新值（own-write 可见）。 + await tx.photos_ov.updateMany({ + where: { photoId }, + data: { + likeTotal: { increment: dto.delta }, + countVersion: { increment: 1 }, + }, + }); + const row = await tx.photos_ov.findUniqueOrThrow({ + where: { photoId }, + select: { likeTotal: true, countVersion: true }, + }); + // 回传队列记权威快照（Task 9：cn 侧仅更高 statVersion 替换）。 + await tx.outbox_stat_events.create({ + data: { + kind: 'likes', + albumId: album.albumId, + photoId, + total: row.likeTotal, + statVersion: row.countVersion, + }, + }); + return { total: row.likeTotal, countVersion: row.countVersion }; + }); + // ⑤ 提交后广播事件（$transaction resolve 即已 COMMIT；失败时上面已抛）。 + const event: LikeBatchCommittedEvent = { + photoId, + albumId: album.albumId, + delta: dto.delta, + total: committed.total, + countVersion: committed.countVersion, + batchId: dto.batchId, + sessionId: dto.sessionId, + }; + this.events.emit(OV_LIKE_BATCH_COMMITTED, event); + return { ...committed, batchAccepted: true }; + } catch (err) { + if ( + typeof err === 'object' && + err !== null && + (err as { code?: unknown }).code === 'P2002' + ) { + // 并发窗口的同批重试：胜者已提交（唯一冲突在对方 COMMIT 后才可见）， + // 返回权威值 + batchAccepted:true（幂等语义，不加计、不重发事件）。 + return this.authoritative(photoId, true); + } + throw err; + } + } + + /** 当前权威计数（幂等/冲突路径共用的只读应答）。 */ + private async authoritative( + photoId: string, + batchAccepted: boolean, + ): Promise { + const row = await this.prisma.photos_ov.findUniqueOrThrow({ + where: { photoId }, + select: { likeTotal: true, countVersion: true }, + }); + return { + total: row.likeTotal, + countVersion: row.countVersion, + batchAccepted, + }; + } + + /** + * 滑窗限速：Redis ov INCRBY 累计 delta，首次写键挂窗口 TTL（固定窗近似 + * 滑动窗——键自首次点赞起存活一个窗口）。sessionId 键先查（拒绝不消耗 + * IP 预算）；IP 键兜底伪造新 sessionId 的批量滥用（NAT 共享场景上限放宽）。 + */ + private async assertRateLimit( + sessionId: string, + clientIp: string | undefined, + delta: number, + ): Promise { + const windowSeconds = likeWindowSeconds(); + const windows: Array<{ key: string; cap: number }> = [ + { key: `like:window:sess:${sessionId}`, cap: likeSessionWindowDelta() }, + ]; + if (clientIp !== undefined && clientIp !== '') { + windows.push({ + key: `like:window:ip:${clientIp}`, + cap: likeIpWindowDelta(), + }); + } + for (const { key, cap } of windows) { + const counted = await this.redis.incrby(key, delta); + if (counted === delta) { + // 新键（本窗口首次）：挂 TTL；EXPIRE 失败不影响本轮判定。 + await this.redis.expire(key, windowSeconds).catch(() => undefined); + } + if (counted > cap) { + throw rateLimited(); + } + } + } +} diff --git a/services/api/src/main-ov.ts b/services/api/src/main-ov.ts index 41bf46a..769448a 100644 --- a/services/api/src/main-ov.ts +++ b/services/api/src/main-ov.ts @@ -4,11 +4,16 @@ // 不变，部署域名区分」，ov 观众端（viewer :5175，P2 Task 10 切换 VITE_API_BASE // 指向本入口）与 cn viewer 共用 CORS_ORIGINS env，无需 ov 专属配置。env 一律 // 来自进程环境（根 dotenv-cli 注入）。 +// Task 7 起：WS 网关与 HTTP **同端口**（契约执行注意：WS 与 HTTP 同端口 +// 8081）——ContractWsAdapter 按 realtime.ts 契约帧 {type,...} 分发，网关挂载 +// /v1/realtime（REALTIME_WS_PATH），upgrade 请求经适配器路由到 noServer 的 +// ws Server，普通 HTTP 路由不受影响。 import 'reflect-metadata'; import { NestFactory } from '@nestjs/core'; import { type NestExpressApplication } from '@nestjs/platform-express'; import { Logger } from 'nestjs-pino'; import { AppOvModule } from './app-ov.module'; +import { ContractWsAdapter } from './realtime/contract-ws.adapter'; import { configureApp } from './shared/configure-app'; async function bootstrap(): Promise { @@ -20,6 +25,8 @@ async function bootstrap(): Promise { rawBody: true, }); app.useLogger(app.get(Logger)); + // WS 与 HTTP 同端口：网关（@WebSocketGateway 无 port）挂到本应用 HTTP 服务。 + app.useWebSocketAdapter(new ContractWsAdapter(app)); configureApp(app); const port = Number.parseInt(process.env.PORT_OV ?? '', 10) || 8081; await app.listen(port); diff --git a/services/api/src/realtime/contract-ws.adapter.ts b/services/api/src/realtime/contract-ws.adapter.ts new file mode 100644 index 0000000..94a8c0d --- /dev/null +++ b/services/api/src/realtime/contract-ws.adapter.ts @@ -0,0 +1,40 @@ +// 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, + transform: (data: unknown) => Observable, + ): Observable { + 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)); + } +} diff --git a/services/api/src/realtime/ov-events.ts b/services/api/src/realtime/ov-events.ts new file mode 100644 index 0000000..bf632d1 --- /dev/null +++ b/services/api/src/realtime/ov-events.ts @@ -0,0 +1,38 @@ +// services/api/src/realtime/ov-events.ts —— ov 侧「业务提交 → 实时广播」的 +// 进程内事件契约（P2 Task 7）。点赞端点（interactions）与收件箱应用 +// （sync-inbox）在**事务提交后**发布这些事件；WS 网关（realtime.gateway） +// 经 @OnEvent 监听并广播给相册房间。事件名/载荷只在本进程内有效。 +// +// 单进程假设（登记偏差，P5 移交）：EventEmitter2 是进程内总线——ov 入口 +// 当前单实例部署（compose 仿真），多实例横向扩展需换 Redis pub/sub 跨进程 +// 扇出（REDIS_URL_OV 已就位，P5 部署项）；届时本文件的事件名/载荷可原样 +// 迁移为频道名/消息体。 + +/** 点赞批次已提交入账（事务提交后发布；网关合并为 likes.burst）。 */ +export const OV_LIKE_BATCH_COMMITTED = 'ov.like.batch.committed'; + +export interface LikeBatchCommittedEvent { + photoId: string; + albumId: string; + /** 本批 delta（burst 合并前的单批值）。 */ + delta: number; + /** 提交事务后的权威累计与计数版本（burst 取合并窗内最新值）。 */ + total: number; + countVersion: number; + /** 幂等批次键（burst 的 originBatchIds 元素，客户端据此过滤自身批次）。 */ + batchId: string; + sessionId: string; +} + +/** 副本照片可见性已应用（sync-inbox 事务提交后发布；网关原样转 photo.*）。 */ +export const OV_PHOTO_APPLIED = 'ov.photo.applied'; + +export interface PhotoAppliedEvent { + kind: 'published' | 'hidden' | 'deleted'; + photoId: string; + albumId: string; + /** cn 权威版本（photo.* 消息的 entityVersion）。 */ + entityVersion: number; + /** kind=published 时必填（photo.published 消息字段）。 */ + publishedAt?: string; +} diff --git a/services/api/src/realtime/realtime.gateway.ts b/services/api/src/realtime/realtime.gateway.ts new file mode 100644 index 0000000..ad79028 --- /dev/null +++ b/services/api/src/realtime/realtime.gateway.ts @@ -0,0 +1,322 @@ +// 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, + type LikeBatchCommittedEvent, + type PhotoAppliedEvent, +} 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>(); + /** 连接 → 已订 {publicToken → albumId}（退订/断开清理用）。 */ + private readonly subscriptions = new Map>(); + /** 传输层心跳存活标记（pong 置回 true；轮询时仍 false → terminate）。 */ + private readonly alive = new Map(); + private heartbeatTimer: NodeJS.Timeout | null = null; + /** photoId → 合并窗待发 burst。 */ + private readonly bursts = new Map(); + + 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 { + 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); + } + + /** 副本照片可见性应用 → 原样转发契约 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); + } + } +} diff --git a/services/api/src/realtime/realtime.module.ts b/services/api/src/realtime/realtime.module.ts new file mode 100644 index 0000000..29dbbe0 --- /dev/null +++ b/services/api/src/realtime/realtime.module.ts @@ -0,0 +1,15 @@ +// services/api/src/realtime/ —— ov 实时广播模块（P2 Task 7）。仅 ov 入口装配： +// 网关挂载由 main-ov 的 ContractWsAdapter 承担（同端口 /v1/realtime），本模块 +// 注册网关 provider（订阅校验读 ov 副本 → OvPrismaModule）。广播事件源为 +// 进程内 EventEmitter2（app 根 EventEmitterModule.forRoot() 全局注册； +// interactions/sync-inbox 发布，网关 @OnEvent 消费——无模块间直接依赖）。 +// stats.updated（Task 8 浏览计数）将经同一模式接入。 +import { Module } from '@nestjs/common'; +import { OvPrismaModule } from '../prisma-ov/ov-prisma.module'; +import { RealtimeGateway } from './realtime.gateway'; + +@Module({ + imports: [OvPrismaModule], + providers: [RealtimeGateway], +}) +export class RealtimeModule {} diff --git a/services/api/src/redis-ov/ov-redis.module.ts b/services/api/src/redis-ov/ov-redis.module.ts new file mode 100644 index 0000000..853ad90 --- /dev/null +++ b/services/api/src/redis-ov/ov-redis.module.ts @@ -0,0 +1,42 @@ +// services/api/src/redis-ov/ —— OV 侧 Redis（redis-ov 6381）的可注入封装 +// （P2 Task 7）。与 OvPrismaService 同款约定：非 @Global、独立类名，仅 ov +// 角色模块（interactions 等）显式 import，防止 cn 角色模块误注入。首个消费方 +// 是点赞基础限速（sessionId/IP 滑窗 INCRBY+EXPIRE）；P5 多实例部署时本连接 +// 亦是跨进程实时扇出（pub/sub）的就位基础（见 realtime/ov-events.ts 头注）。 +// +// 生命周期：构造即连接（ioredis 缺省急连 + 自动重连），onModuleDestroy 断开 +// ——e2e 的 app.close() 不残留句柄。error 事件挂 no-op 监听（同 queue.provider +// 约定：未监听的 error 会崩进程，重连风暴交给命令失败语义）。Redis 不可用时 +// 依赖它的端点 5xx（限速是 ov 互动域的组成环节，不做 fail-open——开放匿名 +// 写入端点在基础设施故障时的正确姿势是拒绝而非放行）。 +import { + Injectable, + Module, + type OnModuleDestroy, +} from '@nestjs/common'; +import IORedis from 'ioredis'; + +@Injectable() +export class OvRedisService + extends IORedis + implements OnModuleDestroy +{ + constructor() { + super(process.env.REDIS_URL_OV ?? 'redis://127.0.0.1:6381', { + // 滑窗限速命令的失败要快速上抛（5xx），不做长时间排队重试。 + maxRetriesPerRequest: 2, + }); + // no-op 监听：吞掉重连期 error 事件（命令层失败已向上传播）。 + this.on('error', () => undefined); + } + + async onModuleDestroy(): Promise { + await this.quit().catch(() => undefined); + } +} + +@Module({ + providers: [OvRedisService], + exports: [OvRedisService], +}) +export class OvRedisModule {} diff --git a/services/api/src/sync-inbox/sync-inbox.service.ts b/services/api/src/sync-inbox/sync-inbox.service.ts index f6c9955..840450c 100644 --- a/services/api/src/sync-inbox/sync-inbox.service.ts +++ b/services/api/src/sync-inbox/sync-inbox.service.ts @@ -23,10 +23,15 @@ // 即撤下语义）；asset → banner 投影 albums_ov.bannerAssetKey（cover/logo // 仅推进版本，ov schema 无位）； // ⑤ sync_state upsert appliedVersion（watermark=false 实体行；水位行预留）。 +// ⑥ 实时转发（Task 7）：photo 事件**应用事务提交后**发布 OV_PHOTO_APPLIED +// 进程内事件 → realtime 网关向相册房间广播契约 photo.* 消息（stale/ +// duplicate/staged 不转发；从未复制即撤下的行也不转发——观众端无此照片 +// 可撤，广播无意义）。 // // 墓碑优先由 ② 保证：tombstone 事件版本必高于其前的任意旧版本，旧 // photo.published/category.updated 重放被版本比较直接拒绝。 import { Injectable } from '@nestjs/common'; +import { EventEmitter2 } from '@nestjs/event-emitter'; import type { components } from '@photo/contracts'; import type { AlbumUpdatedPayload, @@ -40,6 +45,7 @@ import type { Prisma } from '@photo/db-ov'; // 值导入（非 import type）：构造器注入的 design:paramtypes 依赖类引用， // type-only 导入会让 Nest 侧元数据退化为 Function 而 DI 失败（repo 约定）。 import { OvPrismaService } from '../prisma-ov/ov-prisma.module'; +import { OV_PHOTO_APPLIED, type PhotoAppliedEvent } from '../realtime/ov-events'; export type ContentBatchResponse = components['schemas']['ContentBatchResponse']; @@ -57,7 +63,10 @@ interface CategoryReplicaEntry { @Injectable() export class SyncInboxService { - constructor(private readonly prisma: OvPrismaService) {} + constructor( + private readonly prisma: OvPrismaService, + private readonly events: EventEmitter2, + ) {} /** * 批量接收：批内按序逐事件事务处理；任一事件失败抛错（整批 500）——发送器 @@ -82,10 +91,11 @@ export class SyncInboxService { return { acceptedEventIds, applied }; } - /** 单事件事务（见文件头 ①-⑤）。P2002 = 并发同 eventId → duplicate。 */ + /** 单事件事务（见文件头 ①-⑥）。P2002 = 并发同 eventId → duplicate。 */ async applyOne(event: SyncEventWire): Promise { + let realtime: PhotoAppliedEvent | undefined; try { - return await this.prisma.$transaction(async (tx) => { + const outcome = await this.prisma.$transaction(async (tx) => { // ① eventId 幂等锚：先查后插 + 唯一约束兜底（并发窗口）。 const seen = await tx.sync_inbox.findUnique({ where: { eventId: event.eventId }, @@ -106,8 +116,15 @@ export class SyncInboxService { payload: event.payload as unknown as Prisma.InputJsonValue, }, }); - return this.decideAndApply(tx, event); + const applied = await this.decideAndApply(tx, event); + realtime = applied.realtime; + return applied.outcome; }); + // ⑥ 提交后转发（$transaction resolve 即已 COMMIT；回滚路径上面已抛）。 + if (realtime !== undefined) { + this.events.emit(OV_PHOTO_APPLIED, realtime); + } + return outcome; } catch (err) { // P2002 鸭子类型判别（同 albums/support.ts isUniqueViolationOn 的约定： // Prisma 命名空间经 @photo/db-ov 仅 type 出口，不做值导入）——本端点 @@ -126,12 +143,13 @@ export class SyncInboxService { /** * ②-⑤ 裁决与应用（inbox 行已存在的共用核心）：新事件（applyOne 落库后） * 与暂存回扫（sweepStaged 对既有行重跑）同一出口。回扫**不能**走 applyOne - * 整程——其 eventId 幂等检查会把暂存行判成 duplicate 而永不应用。 + * 整程——其 eventId 幂等检查会把暂存行判成 duplicate 而永不应用。返回 + * ⑥ 的实时转发事件（photo 事件才有；stale/staged 无）。 */ private async decideAndApply( tx: Prisma.TransactionClient, event: SyncEventWire, - ): Promise { + ): Promise<{ outcome: Outcome; realtime?: PhotoAppliedEvent }> { // ② 版本单调：**更旧**版本裁决即终（appliedAt 置位，不入积压）；**等于** // 已应用版本 → 重断言应用（同版本=同状态，幂等重投影；digest→bootstrap // 补漏修复缺失副本行的机制，见文件头②）。 @@ -148,7 +166,7 @@ export class SyncInboxService { where: { eventId: event.eventId }, data: { appliedAt: new Date() }, }); - return 'stale' as const; + return { outcome: 'stale' }; } // ③ 依赖检查：非 album 事件须所属相册副本已存在，否则暂存。 if (event.entityType !== 'album') { @@ -157,11 +175,11 @@ export class SyncInboxService { select: { albumId: true }, }); if (album === null) { - return 'staged' as const; + return { outcome: 'staged' }; } } - // ④ 投影。 - await this.project(tx, event); + // ④ 投影（photo 事件带回实时转发载荷）。 + const realtime = await this.project(tx, event); // ⑤ 版本账本。 await tx.sync_state.upsert({ where: { @@ -185,14 +203,14 @@ export class SyncInboxService { where: { eventId: event.eventId }, data: { appliedAt: new Date() }, }); - return 'applied' as const; + return { outcome: 'applied', realtime }; } /** ④ 按实体类型投影到副本表（版本已由 ② 保证更高）。 */ private async project( tx: Prisma.TransactionClient, event: SyncEventWire, - ): Promise { + ): Promise { if (event.entityType === 'album') { const p = event.payload as unknown as AlbumUpdatedPayload; // categories 全量清单（可选，仅 bootstrap/reconcile 快照事件携带）：整体 @@ -237,7 +255,7 @@ export class SyncInboxService { entityVersion: event.entityVersion, }, }); - return; + return undefined; } if (event.entityType === 'category') { const album = await tx.albums_ov.findUniqueOrThrow({ @@ -258,7 +276,7 @@ export class SyncInboxService { ) as unknown as Prisma.InputJsonValue, }, }); - return; + return undefined; } const p = event.payload as unknown as CategoryUpdatedPayload; const entry: CategoryReplicaEntry = { @@ -276,7 +294,7 @@ export class SyncInboxService { where: { albumId: event.albumId }, data: { categories: entries as unknown as Prisma.InputJsonValue }, }); - return; + return undefined; } if (event.entityType === 'photo') { const payload = event.payload as unknown as Partial< @@ -293,7 +311,7 @@ export class SyncInboxService { if (visibility !== 'published') { // 墓碑/隐藏先于任何 published 到达且副本行不存在：缺席即撤下语义， // 无行可改——版本账本照常推进（旧 published 之后到会被版本比较拒绝）。 - return; + return undefined; } const p = payload as PhotoPublishedPayload; await tx.photos_ov.create({ @@ -311,7 +329,14 @@ export class SyncInboxService { entityVersion: event.entityVersion, }, }); - return; + // ⑥ 实时转发载荷（photo.published）。 + return { + kind: 'published', + photoId: event.entityId, + albumId: event.albumId, + entityVersion: event.entityVersion, + publishedAt: p.publishedAt, + }; } // update 分支保留 likeTotal/countVersion（ov 权威计数，Task 7/9）。 const data: Prisma.photos_ovUpdateInput = { @@ -332,7 +357,25 @@ export class SyncInboxService { where: { photoId: event.entityId }, data, }); - return; + // ⑥ 实时转发载荷：published 直映；hidden/deleted 按契约消息；其余 + // visibility 字符串（unpublished 等防御路径）按 hidden 语义转发—— + // 观众端的正确反应都是「即刻移出列表」。 + if (visibility === 'published') { + const p = payload as PhotoPublishedPayload; + return { + kind: 'published', + photoId: event.entityId, + albumId: event.albumId, + entityVersion: event.entityVersion, + publishedAt: p.publishedAt, + }; + } + return { + kind: visibility === 'deleted' ? 'deleted' : 'hidden', + photoId: event.entityId, + albumId: event.albumId, + entityVersion: event.entityVersion, + }; } if (event.entityType === 'asset') { const p = event.payload as unknown as AssetUpdatedPayload; @@ -343,9 +386,10 @@ export class SyncInboxService { }); } // cover/logo：ov schema 无投影位（YAGNI，schema 注释），仅推进版本账本。 - return; + return undefined; } // slot-state 预留：受理即 applied（无投影），版本账本照常推进。 + return undefined; } /** album 应用后回扫该相册暂存事件（receivedAt 升序），返回补应用数。 */ @@ -367,13 +411,17 @@ export class SyncInboxService { priority: row.priority as SyncEventWire['priority'], payload: row.payload as unknown as SyncEventWire['payload'], }; - // 行已存在（eventId 即本行主键），走 ②-⑤ 核心而非 applyOne 整程 + // 行已存在（eventId 即本行主键），走 ②-⑥ 核心而非 applyOne 整程 // （幂等检查会把回扫误判 duplicate，见 decideAndApply 注释）。 - const outcome = await this.prisma.$transaction((tx) => + const result = await this.prisma.$transaction((tx) => this.decideAndApply(tx, event), ); - if (outcome === 'applied') { + if (result.outcome === 'applied') { applied += 1; + // ⑥ 暂存补应用同样提交后转发（依赖相册到达的迟到照片事件）。 + if (result.realtime !== undefined) { + this.events.emit(OV_PHOTO_APPLIED, result.realtime); + } } } return applied; diff --git a/services/api/test/emergency.e2e-spec.ts b/services/api/test/emergency.e2e-spec.ts index 23c6613..da47d46 100644 --- a/services/api/test/emergency.e2e-spec.ts +++ b/services/api/test/emergency.e2e-spec.ts @@ -17,6 +17,9 @@ 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'; @@ -80,6 +83,7 @@ describe('ov emergency disable channel (e2e)', () => { rawBody: true, logger: false, }); + ovApp.useWebSocketAdapter(new ContractWsAdapter(ovApp)); configureApp(ovApp); await ovApp.listen(0); ovBase = await ovApp.getUrl(); diff --git a/services/api/test/likes-realtime.e2e-spec.ts b/services/api/test/likes-realtime.e2e-spec.ts new file mode 100644 index 0000000..764f467 --- /dev/null +++ b/services/api/test/likes-realtime.e2e-spec.ts @@ -0,0 +1,653 @@ +// services/api/test/likes-realtime.e2e-spec.ts —— P2 Task 7 点赞与实时 e2e。 +// 与 public-ov.e2e-spec 同模式：进程内起 cn + ov 两个真实入口，业务动作全走 +// cn API（建相册/上传 complete/直驱发布事务），OutboxSender（生产实现）指向 +// ov 入口真实 URL——点赞与 WS 广播作用在与生产链路同源的数据上。ov 入口按 +// main-ov 同形装配 ContractWsAdapter（WS 与 HTTP 同端口 /v1/realtime）。 +// +// 覆盖（任务验收 A09/A10/A21 部分）： +// ① A09 点赞批次：连点 3 批累计 1+2+3（total/countVersion 单调）、同 batchId +// 重试幂等不加计（batchAccepted=true、like_batches 行数不增）、outbox_ +// stat_events 逐批落权威快照；delta 0/21、sessionId/batchId 65 字符 → +// 400；幽灵 token → 404 ALBUM_NOT_PUBLIC；幽灵照片 → 404 PHOTO_NOT_PUBLIC； +// ② 基础限速：LIKE_SESSION_WINDOW_DELTA 压小（惰性 env）→ 窗口超额 429 +// RATE_LIMITED（代码语义验证；窗口本身 24h，不做时间旅行）； +// ③ WS 协议：ping→pong 回显 ts；幽灵 token subscribe → error 帧 +// （ALBUM_NOT_PUBLIC）且连接保持（后续 ping 仍应答）； +// ④ A10 房间隔离：两相册两客户端各自订阅（先以一次本册点赞证明双方订阅 +// 均生效——非空转通过），跨册点赞只达本册房间，另一房间零消息； +// ⑤ A21 合并广播：合并窗内两批同照片 → 恰一条 likes.burst（delta 求和、 +// originBatchIds 双元素、total 权威、burstId 存在），窗外无第二条； +// ⑥ photo.hidden 实时：cn 隐藏 → 高优先级同步 → ov 应用后相册房间收到 +// photo.hidden（entityVersion 与副本行一致）；隐藏后点赞 → 404。 +// +// —— 登记偏差（过期批次重放，不可 e2e）—— +// 24h 窗口外重放已清理 batchId 的「再次加计」边界在 likes.service.ts 头注 +// 分析（无永久幂等账本下的理论边界，由 sessionId/IP 滑窗上限钳制）；窗口 +// 时间不可旅行，e2e 无法覆盖，随 Task 8 清理调度落地时以单元语义评审。 +// +// —— 不在本 spec 的语义（对应实现位）—— +// · 传输层心跳踢除（HEARTBEAT_MS=20s 周期 ping/terminate）：真实 20s 等待 +// 不值 e2e 时长，realtime.gateway.afterInit 直读契约常量（类型锁定见 +// contracts __tests__/realtime.test.ts）；P5 压测域验证。 +// · stats.updated：Task 8 浏览计数接线后由该任务测试。 +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 { 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'; + +/** 合并窗加大到 1s：两批 HTTP 提交的完成时间远小于窗（防慢机抖动）。 */ +const BURST_MERGE_TEST_MS = 1000; +/** subscribe 为异步校验（DB 查询）：发送后的就绪等待（本地查询 ~ms 级）。 */ +const SUBSCRIBE_SETTLE_MS = 300; + +/** 测试内生成真实 JPEG（同 public-ov.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，断言交给调用方）。 */ +async function like( + ovBase: string, + token: string, + photoId: string, + body: { sessionId: string; batchId: string; delta: number }, +): Promise { + return request(ovBase) + .post(`/v1/public/albums/${token}/photos/${photoId}/likes`) + .send(body); +} + +/** 契约帧 WS 测试客户端：收帧队列 + 条件等待。 */ +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 }); + }); + } + + /** 已收帧快照（隔离断言用：跨房间不应新增）。 */ + received(): readonly ServerMessage[] { + return this.frames; + } + + /** 按类型统计（合并广播「恰一条」断言用）。 */ + countOf(type: ServerMessage['type']): number { + return this.frames.filter((f) => f.type === type).length; + } + + close(): void { + this.ws.close(); + } +} + +describe('likes + realtime (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; + let photoA2: string; + let photoB1: 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: 'likes-rt.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(); + // 合并窗与限速阈值均为惰性 env（同 HOT_SORT_TTL_MS 模式）：测试内压/放宽。 + process.env.LIKE_BURST_MERGE_MS = String(BURST_MERGE_TEST_MS); + // 限速滑窗键跨 spec 运行残留（TTL 24h）：清场保证本 spec 从零计数。 + 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', + 'like: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（两张照片：A1 点赞主目标 / A2 隐藏目标）+ 相册 B —— + const albumA = await createAlbum('点赞实时 A'); + tokenA = albumA.publicToken; + albumAId = albumA.albumId; + const albumB = await createAlbum('点赞实时 B'); + tokenB = albumB.publicToken; + + photoA1 = await uploadAndPublish( + albumAId, + 'likes-rt-a1', + await makeJpeg(900, 600, { r: 180, g: 120, b: 60 }), + ); + photoA2 = await uploadAndPublish( + albumAId, + 'likes-rt-a2', + await makeJpeg(800, 500, { r: 60, g: 130, b: 180 }), + ); + photoB1 = await uploadAndPublish( + albumB.albumId, + 'likes-rt-b1', + await makeJpeg(700, 450, { r: 120, g: 180, b: 90 }), + ); + await drainSender(); + }); + + afterAll(async () => { + delete process.env.LIKE_BURST_MERGE_MS; + delete process.env.LIKE_SESSION_WINDOW_DELTA; + await ovApp.close(); + await cnApp.close(); + await prismaCn.$disconnect(); + await prismaOv.$disconnect(); + }); + + it('① A09 点赞批次：连点累计、幂等重试不加计、outbox 快照、参数/404 校验', async () => { + const session = 'e2e-session-a09'; + // 连点 3 批：total 1 → 3 → 6，countVersion 逐批 +1。 + const b1 = await like(ovBase, tokenA, photoA1, { + sessionId: session, + batchId: 'batch-1', + delta: 1, + }); + expect(b1.status).toBe(200); + expect(b1.body).toEqual({ total: 1, countVersion: 1, batchAccepted: true }); + const b2 = await like(ovBase, tokenA, photoA1, { + sessionId: session, + batchId: 'batch-2', + delta: 2, + }); + expect(b2.body).toEqual({ total: 3, countVersion: 2, batchAccepted: true }); + const b3 = await like(ovBase, tokenA, photoA1, { + sessionId: session, + batchId: 'batch-3', + delta: 3, + }); + expect(b3.body).toEqual({ total: 6, countVersion: 3, batchAccepted: true }); + + // 幂等重试：同 batchId 再提交 → 权威 total 不变、batchAccepted 仍 true。 + const retry = await like(ovBase, tokenA, photoA1, { + sessionId: session, + batchId: 'batch-2', + delta: 2, + }); + expect(retry.status).toBe(200); + expect(retry.body).toEqual({ total: 6, countVersion: 3, batchAccepted: true }); + // like_batches 不因重试增行（3 批 3 行）。 + expect(await prismaOv.like_batches.count({ where: { photoId: photoA1 } })).toBe( + 3, + ); + + // outbox_stat_events 逐批落权威快照（total 单调 1/3/6，statVersion 对齐）。 + const outbox = await prismaOv.outbox_stat_events.findMany({ + where: { photoId: photoA1 }, + orderBy: { statVersion: 'asc' }, + }); + expect(outbox.map((r) => [r.total, r.statVersion, r.kind])).toEqual([ + [1, 1, 'likes'], + [3, 2, 'likes'], + [6, 3, 'likes'], + ]); + + // 参数校验：delta 越界 / 字段长度 65 → 400 VALIDATION_FAILED。 + for (const bad of [ + { sessionId: session, batchId: 'bad-0', delta: 0 }, + { sessionId: session, batchId: 'bad-21', delta: 21 }, + { sessionId: 'x'.repeat(65), batchId: 'bad-len', delta: 1 }, + { sessionId: session, batchId: 'y'.repeat(65), delta: 1 }, + ]) { + const res = await like(ovBase, tokenA, photoA1, bad); + expect(res.status).toBe(400); + expect(res.body.code).toBe(ErrorCodes.VALIDATION_FAILED); + } + + // 404：幽灵相册 token / 幽灵照片（合法相册下）。 + const ghostToken = await like(ovBase, 'g' + '0'.repeat(23), photoA1, { + sessionId: session, + batchId: 'bad-token', + delta: 1, + }); + expect(ghostToken.status).toBe(404); + expect(ghostToken.body.code).toBe(ErrorCodes.ALBUM_NOT_PUBLIC); + const ghostPhoto = await like(ovBase, tokenA, 'p-ghost', { + sessionId: session, + batchId: 'bad-photo', + delta: 1, + }); + expect(ghostPhoto.status).toBe(404); + expect(ghostPhoto.body.code).toBe(ErrorCodes.PHOTO_NOT_PUBLIC); + }); + + it('② 基础限速：sessionId 滑窗超额 → 429 RATE_LIMITED', async () => { + process.env.LIKE_SESSION_WINDOW_DELTA = '5'; + try { + const session = 'e2e-session-rate'; + const ok = await like(ovBase, tokenA, photoA1, { + sessionId: session, + batchId: 'rl-1', + delta: 3, + }); + expect(ok.status).toBe(200); + // 累计 3+3=6 > 5 → 拒绝（且本批不加计）。 + const limited = await like(ovBase, tokenA, photoA1, { + sessionId: session, + batchId: 'rl-2', + delta: 3, + }); + expect(limited.status).toBe(429); + expect(limited.body.code).toBe(ErrorCodes.RATE_LIMITED); + // 未入账：批次表无 rl-2 行，权威 total = ① 的 6 + 本测试已受理的 rl-1=3 + // （若被拒批次加计则为 12）。 + expect( + await prismaOv.like_batches.count({ + where: { sessionId: session }, + }), + ).toBe(1); + const row = await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId: photoA1 }, + select: { likeTotal: true }, + }); + expect(row.likeTotal).toBe(9); + } finally { + delete process.env.LIKE_SESSION_WINDOW_DELTA; + } + }); + + it('③ WS 协议：ping→pong 回显、幽灵 token error 帧、错误后连接保持', async () => { + const client = await WsClient.open(wsUrl()); + try { + const ts = 1758187200123; + client.send({ type: 'ping', ts }); + const pong = await client.waitFor( + (f) => f.type === 'pong', + 3000, + 'pong', + ); + expect(pong).toEqual({ type: 'pong', ts }); + + // 幽灵 token：error 帧（ALBUM_NOT_PUBLIC）。 + client.send({ + type: 'subscribe', + albumToken: 'g' + '0'.repeat(23), + }); + const error = await client.waitFor( + (f) => f.type === 'error', + 3000, + 'error', + ); + expect(error).toEqual({ + type: 'error', + code: ErrorCodes.ALBUM_NOT_PUBLIC, + }); + + // 连接保持：错误后仍可 ping→pong（可修正后重订其它相册）。 + client.send({ type: 'ping', ts: 1758187200456 }); + const pong2 = await client.waitFor( + (f) => f.type === 'pong' && f.ts === 1758187200456, + 3000, + 'pong2', + ); + if (pong2.type !== 'pong') { + throw new Error('unreachable'); + } + expect(pong2.ts).toBe(1758187200456); + } finally { + client.close(); + } + }); + + it('④ A10 房间隔离：跨册点赞只达本册房间，另一房间零消息', async () => { + const clientA = await WsClient.open(wsUrl()); + const clientB = await WsClient.open(wsUrl()); + try { + clientA.send({ type: 'subscribe', albumToken: tokenA }); + clientB.send({ type: 'subscribe', albumToken: tokenB }); + await new Promise((resolve) => setTimeout(resolve, SUBSCRIBE_SETTLE_MS)); + + // 非空转证明：先各以一次本册点赞证明两侧订阅均已生效。 + await like(ovBase, tokenA, photoA1, { + sessionId: 'e2e-session-iso-a', + batchId: 'iso-a1', + delta: 1, + }); + await clientA.waitFor((f) => f.type === 'likes.burst', 5000, 'A 首个 burst'); + await like(ovBase, tokenB, photoB1, { + sessionId: 'e2e-session-iso-b', + batchId: 'iso-b1', + delta: 1, + }); + await clientB.waitFor((f) => f.type === 'likes.burst', 5000, 'B 首个 burst'); + + // 隔离断言：再赞 B，A 房间在「B 收到 burst + 合并窗外一拍」后仍零新增。 + const aBefore = clientA.received().length; + await like(ovBase, tokenB, photoB1, { + sessionId: 'e2e-session-iso-b', + batchId: 'iso-b2', + delta: 2, + }); + await clientB.waitFor( + (f) => f.type === 'likes.burst' && f.originBatchIds.includes('iso-b2'), + 5000, + 'B 第二个 burst', + ); + await new Promise((resolve) => + setTimeout(resolve, BURST_MERGE_TEST_MS + 300), + ); + expect(clientA.received().length).toBe(aBefore); // A 房间零新增 + } finally { + clientA.close(); + clientB.close(); + } + }); + + it('⑤ A21 合并广播：合并窗内两批 → 恰一条 likes.burst（delta 求和 + 双 originBatchIds）', async () => { + const client = await WsClient.open(wsUrl()); + try { + client.send({ type: 'subscribe', albumToken: tokenA }); + await new Promise((resolve) => setTimeout(resolve, SUBSCRIBE_SETTLE_MS)); + + const before = await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId: photoA1 }, + select: { likeTotal: true, countVersion: true }, + }); + + // 两批快速提交（远小于合并窗）：同照片不同 batchId。 + const first = await like(ovBase, tokenA, photoA1, { + sessionId: 'e2e-session-merge', + batchId: 'merge-1', + delta: 1, + }); + const second = await like(ovBase, tokenA, photoA1, { + sessionId: 'e2e-session-merge', + batchId: 'merge-2', + delta: 2, + }); + expect(first.body.total).toBe(before.likeTotal + 1); + expect(second.body.total).toBe(before.likeTotal + 3); + + // 恰一条 burst：delta=1+2、originBatchIds 双元素、total 权威、版本对齐。 + const burst = await client.waitFor( + (f) => f.type === 'likes.burst' && f.originBatchIds.includes('merge-2'), + 5000, + '合并 burst', + ); + if (burst.type !== 'likes.burst') { + throw new Error('unreachable'); + } + expect(burst.photoId).toBe(photoA1); + expect(burst.albumId).toBe(albumAId); + expect(burst.delta).toBe(3); + expect([...burst.originBatchIds].sort()).toEqual(['merge-1', 'merge-2']); + expect(burst.total).toBe(before.likeTotal + 3); + expect(burst.countVersion).toBe(before.countVersion + 2); + expect(typeof burst.burstId).toBe('string'); + + // 窗外不再补发第二条（等待一个完整合并窗 + 余量）。 + await new Promise((resolve) => + setTimeout(resolve, BURST_MERGE_TEST_MS + 300), + ); + expect(client.countOf('likes.burst')).toBe(1); + } finally { + client.close(); + } + }); + + it('⑥ photo.hidden 实时：cn 隐藏 → 同步应用 → 房间收 photo.hidden；隐藏后点赞 404', async () => { + const client = await WsClient.open(wsUrl()); + try { + client.send({ type: 'subscribe', albumToken: tokenA }); + await new Promise((resolve) => setTimeout(resolve, SUBSCRIBE_SETTLE_MS)); + + const hide = await request(cnBase) + .patch(`/v1/photos/${photoA2}`) + .set('Authorization', `Bearer ${adminToken}`) + .send({ visibility: 'hidden' }); + expect(hide.status).toBe(200); + await drainSender(); + + const hidden = await client.waitFor( + (f) => f.type === 'photo.hidden' && f.photoId === photoA2, + 5000, + 'photo.hidden', + ); + if (hidden.type !== 'photo.hidden') { + throw new Error('unreachable'); + } + expect(hidden.albumId).toBe(albumAId); + const replica = await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId: photoA2 }, + select: { entityVersion: true, visibility: true }, + }); + expect(hidden.entityVersion).toBe(replica.entityVersion); + expect(replica.visibility).toBe('hidden'); + + // 隐藏后点赞：404 PHOTO_NOT_PUBLIC（与公开读同口径）。 + const denied = await like(ovBase, tokenA, photoA2, { + sessionId: 'e2e-session-hidden', + batchId: 'hidden-1', + delta: 1, + }); + expect(denied.status).toBe(404); + expect(denied.body.code).toBe(ErrorCodes.PHOTO_NOT_PUBLIC); + } finally { + client.close(); + } + }); +}); diff --git a/services/api/test/public-ov.e2e-spec.ts b/services/api/test/public-ov.e2e-spec.ts index 0bae16d..f51a848 100644 --- a/services/api/test/public-ov.e2e-spec.ts +++ b/services/api/test/public-ov.e2e-spec.ts @@ -38,6 +38,9 @@ import { AppCnModule } from '../src/app-cn.module'; import { AppOvModule } from '../src/app-ov.module'; import { seedAdmin } from '../src/scripts/seed-admin'; 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 { PublishService } from '../../worker/src/publish.service'; import { resetDb } from './reset'; import { resetOvDb } from './reset-ov'; @@ -194,6 +197,7 @@ describe('public-ov (e2e)', () => { ovApp = await NestFactory.create(AppOvModule, { rawBody: true, }); + ovApp.useWebSocketAdapter(new ContractWsAdapter(ovApp)); configureApp(ovApp); await ovApp.listen(0); ovBase = await ovApp.getUrl(); diff --git a/services/sync/test/reconcile.e2e-spec.ts b/services/sync/test/reconcile.e2e-spec.ts index e5bb6c3..8965ea5 100644 --- a/services/sync/test/reconcile.e2e-spec.ts +++ b/services/sync/test/reconcile.e2e-spec.ts @@ -27,6 +27,9 @@ import { AppCnModule } from '../../api/src/app-cn.module'; import { AppOvModule } from '../../api/src/app-ov.module'; import { seedAdmin } from '../../api/src/scripts/seed-admin'; import { configureApp } from '../../api/src/shared/configure-app'; +// Task 7 起 AppOvModule 含 realtime 网关：进程内 ov 入口须与生产 main-ov +// 一致安装契约 WS 适配器（否则 Nest 落到默认 socket.io 驱动 → listen 即报错）。 +import { ContractWsAdapter } from '../../api/src/realtime/contract-ws.adapter'; import { PublishService } from '../../worker/src/publish.service'; import { resetDb } from '../../api/test/reset'; import { resetOvDb } from '../../api/test/reset-ov'; @@ -144,6 +147,7 @@ describe('sync reconcile + digest healing (e2e)', () => { ovApp = await NestFactory.create(AppOvModule, { rawBody: true, }); + ovApp.useWebSocketAdapter(new ContractWsAdapter(ovApp)); configureApp(ovApp); await ovApp.listen(0); ovBase = await ovApp.getUrl(); diff --git a/services/sync/test/sync.e2e-spec.ts b/services/sync/test/sync.e2e-spec.ts index 4ef019e..dd6c8b5 100644 --- a/services/sync/test/sync.e2e-spec.ts +++ b/services/sync/test/sync.e2e-spec.ts @@ -37,6 +37,9 @@ import { AppCnModule } from '../../api/src/app-cn.module'; import { AppOvModule } from '../../api/src/app-ov.module'; import { seedAdmin } from '../../api/src/scripts/seed-admin'; import { configureApp } from '../../api/src/shared/configure-app'; +// Task 7 起 AppOvModule 含 realtime 网关：进程内 ov 入口须与生产 main-ov +// 一致安装契约 WS 适配器（否则 Nest 落到默认 socket.io 驱动 → listen 即报错）。 +import { ContractWsAdapter } from '../../api/src/realtime/contract-ws.adapter'; import { PublishService } from '../../worker/src/publish.service'; import { resetDb } from '../../api/test/reset'; import { resetOvDb } from '../../api/test/reset-ov'; @@ -221,6 +224,7 @@ describe('sync cn→ov (e2e)', () => { ovApp = await NestFactory.create(AppOvModule, { rawBody: true, }); + ovApp.useWebSocketAdapter(new ContractWsAdapter(ovApp)); configureApp(ovApp); await ovApp.listen(0); ovBase = await ovApp.getUrl();