Files
photo-live/services/sync/test/sync.e2e-spec.ts
T
浅念 85b2da3db5 feat(ov): idempotent batched likes and realtime broadcast
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>
2026-09-18 17:51:55 +08:00

792 lines
28 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// services/sync/test/sync.e2e-spec.ts —— P2 Task 3 cn→ov 同步链路 E2E。
// ONE spec 串完整同步流:进程内起**两个真实 HTTP 入口**(cn=AppCnModule、
// ov=AppOvModule,均 NestFactory rawBody:true + configureApp + listen(0)),
// 发送器为生产 OutboxSender 实例(指向 ov 入口真实 URL),业务动作全走 cn
// 公开 API(登录/建相册/分类/上传/complete),发布事务直驱 worker
// PublishService(同 api full-chain e2e 模式,免 BullMQ 异步等待)。
//
// 覆盖(任务验收 8 场景):
// ① 全链路复制:相册+分类+发布照片 → 排空发送器 → ov 副本齐备;
// ② 相册下线(publicEnabled=false)→ outbox high 优先 + 批内先于更早发生的
// normal 事件被受理;
// ③ 重复推送幂等:清 deliveredAt 模拟「推送成功未确认」重推 → 受理 applied=0;
// ④ 隐藏照片 → photos_ov.visibility=hidden;旧 published 以新 eventId 重放不
// 复活(A26 版本单调);同 eventId 重放受理不计应用;
// ⑤ bootstrap 空 ov → 快照事件入 outbox(eventCount/watermark)→ 排空 →
// 副本反映**当前**状态(下线/隐藏仍在);
// ⑥ 依赖暂存:ov 无相册行时 photo 事件 appliedAt=null 暂存,album 事件到达
// 即回扫补应用;
// ⑦ HMAC 验签:缺头/错签/签名覆盖体不符/时间戳超窗 → 401 SYNC_AUTH_FAILED
// (ov content-batches + cn bootstrap 同一守卫);
// ⑧ reconcile:按 entityType 返回当前状态清单(sinceVersion 过滤/升序);
// ⑨ 乱序:v7 hidden 先应用、迟到的 v6 published 被版本比较拒绝(A26)。
// 真实 MySQL×2 + MinIO(staging/originals/variants)+ HMAC 全链路字节一致。
import { randomUUID } from 'node:crypto';
import { NestFactory } from '@nestjs/core';
import { type NestExpressApplication } from '@nestjs/platform-express';
import request from 'supertest';
import sharp from 'sharp';
import { PrismaClient } from '@photo/db-cn';
import { PrismaClient as OvPrismaClient } from '@photo/db-ov';
import { ErrorCodes, signSyncBody } from '@photo/contracts';
import {
createStorageFromEnv,
type MinioStorageAdapter,
} from '@photo/storage';
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';
import { OutboxSender } from '../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';
const CONTENT_BATCHES_PATH = '/v1/internal/sync/content-batches';
const BOOTSTRAP_PATH = '/v1/internal/sync/bootstrap';
const RECONCILE_PATH = '/v1/internal/sync/reconcile';
/** 测试内生成真实 JPEG(同 api full-chain e2e;无二进制入库)。 */
async function makeJpeg(
width: number,
height: number,
rgb: { r: number; g: number; b: number },
): Promise<Buffer> {
return sharp({ create: { width, height, channels: 3, background: rgb } })
.jpeg({ quality: 85 })
.toBuffer();
}
/** 签名 POST(可注入坏签名/时间戳偏移/缺头,场景⑦专用)。 */
interface SignOverrides {
signature?: string;
timestampOffsetSeconds?: number;
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) + (overrides.timestampOffsetSeconds ?? 0),
);
}
if (!overrides.omitSignature) {
headers['x-sync-signature'] =
overrides.signature ?? signSyncBody(SYNC_KEY, raw);
}
return fetch(`${base}${path}`, { method: 'POST', headers, body: raw });
}
/** 线格式事件(openapi SyncEvent;测试内手工构造乱序/重放/暂存场景)。 */
interface WireEvent {
eventId: string;
schemaVersion: 1;
entityType: 'album' | 'category' | 'photo' | 'asset';
entityId: string;
albumId: string;
entityVersion: number;
tombstone: boolean;
priority: 'normal' | 'high';
payload: Record<string, unknown>;
}
/** reconcile 清单元素(断言用形状)。 */
interface ReconEvent {
eventId: string;
schemaVersion: number;
entityType: string;
entityId: string;
entityVersion: number;
payload: {
albumId?: string;
publicEnabled?: boolean;
visibility?: string;
entityVersion?: number;
};
}
describe('sync cn→ov (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 albumId: string;
let publicToken: string;
let categoryId: string;
let photo1: { photoId: string; objectKey: string };
let photo2: { photoId: string; objectKey: string };
/** 场景②的 high 事件行 id(场景③复用)。 */
let offlineEventId: string;
/** photo2 的 published 载荷(场景⑨乱序重放用)。 */
let photo2PublishedPayload: Record<string, unknown>;
/** 排空发送器:循环 runOnce 直至无未投递事件(每轮整批推送)。 */
async function drainSender(): Promise<void> {
for (let round = 0; round < 20; round++) {
const stats = await sender.runOnce();
if (stats.picked === 0) {
return;
}
}
throw new Error('drainSender 超时:20 轮后 outbox 仍有未投递事件');
}
/** 上传全链路:init → presign PUT 实写 MinIO → complete → 直驱发布事务。 */
async function uploadAndPublish(
clientTaskId: string,
jpeg: Buffer,
): Promise<{ photoId: string; objectKey: string }> {
const init = await request(cnBase)
.post('/v1/uploads')
.set('Authorization', `Bearer ${adminToken}`)
.send({
albumId,
clientTaskId,
filename: 'sync.jpg',
bytes: jpeg.length,
contentType: 'image/jpeg',
categoryId,
});
expect(init.status).toBe(201);
const { uploadId, photoId, credential } = init.body as {
uploadId: string;
photoId: string;
credential: { url: string; method: string; headers: Record<string, string> };
};
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',
);
// init 响应的 objectKey 是 staging 键;complete 受控复制后正式键在 DB 行上
//(同 api full-chain e2e 的取值口径)。
const finalKey = (
await prismaCn.photos.findUniqueOrThrow({
where: { id: photoId },
select: { objectKey: true },
})
).objectKey;
return { photoId, objectKey: finalKey };
}
/** 直推 ov content-batches(绕过发送器,构造乱序/重放/暂存场景)。 */
async function pushEvents(
events: WireEvent[],
): Promise<{ acceptedEventIds: string[]; applied: number }> {
const res = await signedPost(ovBase, CONTENT_BATCHES_PATH, { batch: events });
expect(res.status).toBe(200);
return (await res.json()) as { acceptedEventIds: string[]; applied: number };
}
beforeAll(async () => {
expect(SYNC_KEY).not.toBe('');
await resetDb();
await resetOvDb();
cnApp = await NestFactory.create<NestExpressApplication>(AppCnModule, {
rawBody: true,
});
configureApp(cnApp);
await cnApp.listen(0);
cnBase = await cnApp.getUrl();
ovApp = await NestFactory.create<NestExpressApplication>(AppOvModule, {
rawBody: true,
});
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;
});
afterAll(async () => {
await cnApp.close();
await ovApp.close();
await prismaCn.$disconnect();
await prismaOv.$disconnect();
});
it('① 全链路复制:建相册+分类+发布照片 → 发送器排空 → ov 副本齐备', async () => {
const album = await request(cnBase)
.post('/v1/albums')
.set('Authorization', `Bearer ${adminToken}`)
.send({
translations: [
{ locale: 'zh-CN', title: '同步晚会', description: '端到端验收' },
{ locale: 'en', title: 'Sync Gala' },
],
defaultLocale: 'zh-CN',
startAt: '2026-10-01T10:00:00.000Z',
});
expect(album.status).toBe(201);
albumId = album.body.id as string;
publicToken = album.body.publicToken as string;
const category = await request(cnBase)
.post(`/v1/albums/${albumId}/categories`)
.set('Authorization', `Bearer ${adminToken}`)
.send({
sortOrder: 0,
translations: [
{ locale: 'zh-CN', name: '主舞台' },
{ locale: 'en', name: 'Main Stage' },
],
});
expect(category.status).toBe(201);
categoryId = category.body.id as string;
photo1 = await uploadAndPublish(
'sync-a',
await makeJpeg(900, 600, { r: 180, g: 120, b: 60 }),
);
await drainSender();
// outbox:三类事件均已确认投递。
const albumEvent = await prismaCn.outbox_events.findFirstOrThrow({
where: { type: 'album.updated', albumId },
});
expect(albumEvent).toMatchObject({ entityVersion: 1, priority: 'normal' });
expect(albumEvent.deliveredAt).not.toBeNull();
expect(
await prismaCn.outbox_events.count({
where: { deliveredAt: null },
}),
).toBe(0);
// ov:相册副本(标量 + 翻译 locale 升序 + 分类 JSON 元素)。
const albumReplica = await prismaOv.albums_ov.findUniqueOrThrow({
where: { albumId },
});
expect(albumReplica).toMatchObject({
publicToken,
publicEnabled: true,
defaultLocale: 'zh-CN',
entityVersion: 1,
});
expect(albumReplica.translations).toEqual([
{ locale: 'en', title: 'Sync Gala', description: null, venue: null, address: null },
{
locale: 'zh-CN',
title: '同步晚会',
description: '端到端验收',
venue: null,
address: null,
},
]);
expect(albumReplica.categories).toEqual([
{
id: categoryId,
sortOrder: 0,
translations: [
{ locale: 'en', name: 'Main Stage' },
{ locale: 'zh-CN', name: '主舞台' },
],
},
]);
// ov:照片副本(发布载荷全量:三键 + 尺寸 + publishedAt)。
const photoReplica = await prismaOv.photos_ov.findUniqueOrThrow({
where: { photoId: photo1.photoId },
});
expect(photoReplica).toMatchObject({
albumId,
originalObjectKey: photo1.objectKey,
visibility: 'published',
width: 900,
height: 600,
entityVersion: 2,
});
expect(photoReplica.thumbObjectKey).toContain(`/variants/${photo1.photoId}/`);
expect(photoReplica.browseObjectKey).toContain(`/variants/${photo1.photoId}/`);
expect(photoReplica.publishedAt).not.toBeNull();
// ov:版本账本。
for (const [entityType, entityId, appliedVersion] of [
['album', albumId, 1],
['category', categoryId, 1],
['photo', photo1.photoId, 2],
] as const) {
expect(
await prismaOv.sync_state.findUniqueOrThrow({
where: { entityType_entityId: { entityType, entityId } },
}),
).toMatchObject({ appliedVersion, watermark: false });
}
});
it('② 相册下线 → high 优先事件先于更早发生的 normal 事件被受理', async () => {
// 先造一条更早发生的 normal 事件(photo2 发布),再下线相册(high)。
photo2 = await uploadAndPublish(
'sync-b',
await makeJpeg(800, 600, { r: 60, g: 130, b: 180 }),
);
const offline = await request(cnBase)
.patch(`/v1/albums/${albumId}`)
.set('Authorization', `Bearer ${adminToken}`)
.send({ publicEnabled: false });
expect(offline.status).toBe(200);
expect(offline.body).toMatchObject({ id: albumId, publicEnabled: false });
const highRow = await prismaCn.outbox_events.findFirstOrThrow({
where: { deliveredAt: null, priority: 'high' },
});
expect(highRow).toMatchObject({ type: 'album.updated', entityVersion: 2 });
offlineEventId = highRow.id;
const photo2Row = await prismaCn.outbox_events.findFirstOrThrow({
where: { deliveredAt: null, type: 'photo.published', photoId: photo2.photoId },
});
const stats = await sender.runOnce();
expect(stats.picked).toBe(2);
expect(stats.applied).toBe(2);
// 顺序断言:high(后发生)先于 normal(先发生)。
expect(stats.acceptedEventIds).toEqual([offlineEventId, photo2Row.id]);
expect(
await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId } }),
).toMatchObject({ publicEnabled: false, entityVersion: 2 });
expect(
await prismaOv.photos_ov.findUniqueOrThrow({ where: { photoId: photo2.photoId } }),
).toMatchObject({ visibility: 'published', entityVersion: 2 });
});
it('③ 重复推送幂等:重推已应用 eventId → 受理但 applied=0', async () => {
// 模拟「ov 已应用、cn 未收到确认」:清 deliveredAt 让发送器重推同一事件。
await prismaCn.outbox_events.update({
where: { id: offlineEventId },
data: { deliveredAt: null },
});
const stats = await sender.runOnce();
expect(stats.picked).toBe(1);
expect(stats.acceptedEventIds).toEqual([offlineEventId]);
expect(stats.applied).toBe(0);
const replica = await prismaOv.albums_ov.findUniqueOrThrow({
where: { albumId },
});
expect(replica).toMatchObject({ publicEnabled: false, entityVersion: 2 });
expect(
(await prismaCn.outbox_events.findUniqueOrThrow({
where: { id: offlineEventId },
})).deliveredAt,
).not.toBeNull();
});
it('④ 隐藏照片 → photos_ov.hidden;旧 published 重放不复活(A26)', async () => {
const hide = await request(cnBase)
.patch(`/v1/photos/${photo1.photoId}`)
.set('Authorization', `Bearer ${adminToken}`)
.send({ visibility: 'hidden' });
expect(hide.status).toBe(200);
const hiddenRow = await prismaCn.outbox_events.findFirstOrThrow({
where: { type: 'photo.hidden', photoId: photo1.photoId },
});
expect(hiddenRow).toMatchObject({ entityVersion: 3, priority: 'high' });
await drainSender();
expect(
await prismaOv.photos_ov.findUniqueOrThrow({
where: { photoId: photo1.photoId },
}),
).toMatchObject({ visibility: 'hidden', entityVersion: 3 });
// A26:旧 published(v2)换新 eventId 重放 → 版本比较拒绝,不复活。
const publishedRow = await prismaCn.outbox_events.findFirstOrThrow({
where: { type: 'photo.published', photoId: photo1.photoId },
});
const replayId = `e2e-${randomUUID()}`;
const replay = await pushEvents([
{
eventId: replayId,
schemaVersion: 1,
entityType: 'photo',
entityId: photo1.photoId,
albumId,
entityVersion: 2,
tombstone: false,
priority: 'normal',
payload: publishedRow.payload as Record<string, unknown>,
},
]);
expect(replay).toEqual({ acceptedEventIds: [replayId], applied: 0 });
expect(
await prismaOv.photos_ov.findUniqueOrThrow({
where: { photoId: photo1.photoId },
}),
).toMatchObject({ visibility: 'hidden', entityVersion: 3 });
// 同 eventId 重放:受理不计应用(eventId 幂等锚)。
const again = await pushEvents([
{
eventId: hiddenRow.id,
schemaVersion: 1,
entityType: 'photo',
entityId: photo1.photoId,
albumId,
entityVersion: 3,
tombstone: false,
priority: 'high',
payload: hiddenRow.payload as Record<string, unknown>,
},
]);
expect(again).toEqual({ acceptedEventIds: [hiddenRow.id], applied: 0 });
});
it('⑤ bootstrap 空 ov → 快照入 outbox(eventCount/watermark)→ 副本反映当前状态', async () => {
await resetOvDb();
const boot = await signedPost(cnBase, BOOTSTRAP_PATH, { albumId });
expect(boot.status).toBe(200);
// 当前状态:album v2(下线)+ category v1 + photo1 hidden v3 + photo2 published v2。
expect(await boot.json()).toEqual({ eventCount: 4, watermark: 3 });
expect(
await prismaCn.outbox_events.count({ where: { deliveredAt: null } }),
).toBe(4);
await drainSender();
expect(
await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId } }),
).toMatchObject({ publicEnabled: false, entityVersion: 2 });
// photo1 当前为 hidden 且空 ov 上从无 published 副本 → 无行(缺席即撤下
// 语义),版本账本照常推进(旧 published 重放会被版本比较拒绝)。
expect(
await prismaOv.photos_ov.count({ where: { photoId: photo1.photoId } }),
).toBe(0);
expect(
await prismaOv.sync_state.findUniqueOrThrow({
where: {
entityType_entityId: { entityType: 'photo', entityId: photo1.photoId },
},
}),
).toMatchObject({ appliedVersion: 3, watermark: false });
expect(
await prismaOv.photos_ov.findUniqueOrThrow({
where: { photoId: photo2.photoId },
}),
).toMatchObject({ visibility: 'published', entityVersion: 2 });
expect(await prismaOv.sync_state.count({ where: { watermark: false } })).toBe(
4,
);
});
it('⑥ 依赖暂存:相册缺失时 photo 暂存,album 到达后回扫补应用', async () => {
await resetOvDb();
const source = await prismaCn.outbox_events.findFirstOrThrow({
where: { type: 'photo.published', photoId: photo2.photoId },
});
photo2PublishedPayload = source.payload as Record<string, unknown>;
const stagedId = `e2e-${randomUUID()}`;
const stage = await pushEvents([
{
eventId: stagedId,
schemaVersion: 1,
entityType: 'photo',
entityId: photo2.photoId,
albumId,
entityVersion: 2,
tombstone: false,
priority: 'normal',
payload: photo2PublishedPayload,
},
]);
expect(stage).toEqual({ acceptedEventIds: [stagedId], applied: 0 });
// 暂存:inbox 行在、appliedAt 为空、无副本投影、无版本账本。
expect(
await prismaOv.sync_inbox.findUniqueOrThrow({ where: { eventId: stagedId } }),
).toMatchObject({ appliedAt: null, entityType: 'photo' });
expect(await prismaOv.photos_ov.count({ where: { albumId } })).toBe(0);
expect(
await prismaOv.sync_state.count({
where: { entityType: 'photo', entityId: photo2.photoId },
}),
).toBe(0);
// album 事件到达:应用 + 回扫暂存(photo 补应用)。
const albumSource = await prismaCn.outbox_events.findFirstOrThrow({
where: { type: 'album.updated', albumId },
orderBy: { occurredAt: 'desc' },
});
const albumEventId = `e2e-${randomUUID()}`;
const unlock = await pushEvents([
{
eventId: albumEventId,
schemaVersion: 1,
entityType: 'album',
entityId: albumId,
albumId,
entityVersion: albumSource.entityVersion,
tombstone: false,
priority: 'normal',
payload: albumSource.payload as Record<string, unknown>,
},
]);
expect(unlock).toEqual({
acceptedEventIds: [albumEventId],
applied: 2, // album 自身 + 回扫的 photo
});
expect(await prismaOv.albums_ov.count({ where: { albumId } })).toBe(1);
expect(
await prismaOv.photos_ov.findUniqueOrThrow({
where: { photoId: photo2.photoId },
}),
).toMatchObject({ visibility: 'published', entityVersion: 2 });
expect(
(
await prismaOv.sync_inbox.findUniqueOrThrow({
where: { eventId: stagedId },
})
).appliedAt,
).not.toBeNull();
});
it('⑦ HMAC 验签:缺头/错签/签名体不符/时间戳超窗 → 401 SYNC_AUTH_FAILED', async () => {
const body = {
batch: [
{
eventId: `e2e-${randomUUID()}`,
schemaVersion: 1,
entityType: 'album',
entityId: 'hmac-probe',
albumId: 'hmac-probe',
entityVersion: 1,
tombstone: false,
priority: 'normal',
payload: {},
},
],
};
const expectUnauthorized = async (res: Promise<Response>): Promise<void> => {
const settled = await res;
expect(settled.status).toBe(401);
expect(((await settled.json()) as { code: string }).code).toBe(
ErrorCodes.SYNC_AUTH_FAILED,
);
};
// 缺签名/时间戳头。
await expectUnauthorized(
signedPost(ovBase, CONTENT_BATCHES_PATH, body, {
omitSignature: true,
omitTimestamp: true,
}),
);
// 形状合法的错签。
await expectUnauthorized(
signedPost(ovBase, CONTENT_BATCHES_PATH, body, { signature: '0'.repeat(64) }),
);
// 签名覆盖体与实际请求体不符(验的是原始字节,非重序列化 body)。
await expectUnauthorized(
signedPost(ovBase, CONTENT_BATCHES_PATH, body, {
signature: signSyncBody(SYNC_KEY, '{"batch":[]}'),
}),
);
// 时间戳超窗(签名本身有效)。
await expectUnauthorized(
signedPost(ovBase, CONTENT_BATCHES_PATH, body, {
timestampOffsetSeconds: -400,
}),
);
// cn 侧端点同一守卫。
await expectUnauthorized(
signedPost(cnBase, BOOTSTRAP_PATH, {}, { omitSignature: true }),
);
});
it('⑧ reconcile:按 entityType 返回当前状态清单(sinceVersion 过滤、升序)', async () => {
const albumList = await signedPost(cnBase, RECONCILE_PATH, {
entityType: 'album',
sinceVersion: 0,
limit: 10,
});
expect(albumList.status).toBe(200);
const albumEvents = ((await albumList.json()) as { events: ReconEvent[] })
.events;
expect(albumEvents).toHaveLength(1);
expect(albumEvents[0]).toMatchObject({
eventId: `recon:album:${albumId}:2`,
schemaVersion: 1,
entityType: 'album',
entityId: albumId,
entityVersion: 2,
priority: 'normal',
});
expect(albumEvents[0].payload).toMatchObject({
albumId,
publicEnabled: false,
entityVersion: 2,
});
const photoList = await signedPost(cnBase, RECONCILE_PATH, {
entityType: 'photo',
albumId,
sinceVersion: 0,
limit: 10,
});
expect(photoList.status).toBe(200);
const photoEvents = ((await photoList.json()) as { events: ReconEvent[] })
.events;
// 升序:photo2 v2(published)→ photo1 v3(hidden)。
expect(
photoEvents.map((e) => [e.entityId, e.entityVersion]),
).toEqual([
[photo2.photoId, 2],
[photo1.photoId, 3],
]);
expect(photoEvents[1].payload.visibility).toBe('hidden');
// sinceVersion=当前版本 → 空(增量语义)。
const empty = await signedPost(cnBase, RECONCILE_PATH, {
entityType: 'album',
sinceVersion: 2,
limit: 10,
});
expect(((await empty.json()) as { events: unknown[] }).events).toEqual([]);
});
it('⑨ 乱序:v7 hidden 先应用、迟到的 v6 published 不复活(A26)', async () => {
const v7Id = `e2e-${randomUUID()}`;
const first = await pushEvents([
{
eventId: v7Id,
schemaVersion: 1,
entityType: 'photo',
entityId: photo2.photoId,
albumId,
entityVersion: 7,
tombstone: false,
priority: 'high',
payload: {
photoId: photo2.photoId,
albumId,
visibility: 'hidden',
entityVersion: 7,
},
},
]);
expect(first).toEqual({ acceptedEventIds: [v7Id], applied: 1 });
expect(
await prismaOv.photos_ov.findUniqueOrThrow({
where: { photoId: photo2.photoId },
}),
).toMatchObject({ visibility: 'hidden', entityVersion: 7 });
const v6Id = `e2e-${randomUUID()}`;
const second = await pushEvents([
{
eventId: v6Id,
schemaVersion: 1,
entityType: 'photo',
entityId: photo2.photoId,
albumId,
entityVersion: 6,
tombstone: false,
priority: 'normal',
payload: { ...photo2PublishedPayload, entityVersion: 6 },
},
]);
expect(second).toEqual({ acceptedEventIds: [v6Id], applied: 0 });
expect(
await prismaOv.photos_ov.findUniqueOrThrow({
where: { photoId: photo2.photoId },
}),
).toMatchObject({ visibility: 'hidden', entityVersion: 7 });
});
it('⑩ 分类更新/删除:category.updated 投影 upsert、墓碑移除元素', async () => {
const rename = await request(cnBase)
.patch(`/v1/categories/${categoryId}`)
.set('Authorization', `Bearer ${adminToken}`)
.send({
sortOrder: 5,
translations: [
{ locale: 'zh-CN', name: '主舞台' },
{ locale: 'en', name: 'Main Stage Renamed' },
],
});
expect(rename.status).toBe(200);
const updatedRow = await prismaCn.outbox_events.findFirstOrThrow({
where: { type: 'category.updated', albumId },
orderBy: { occurredAt: 'desc' },
});
expect(updatedRow).toMatchObject({ entityVersion: 2, priority: 'normal' });
await drainSender();
let replica = await prismaOv.albums_ov.findUniqueOrThrow({
where: { albumId },
});
expect(replica.categories).toEqual([
{
id: categoryId,
sortOrder: 5,
translations: [
{ locale: 'en', name: 'Main Stage Renamed' },
{ locale: 'zh-CN', name: '主舞台' },
],
},
]);
const remove = await request(cnBase)
.delete(`/v1/categories/${categoryId}`)
.set('Authorization', `Bearer ${adminToken}`);
expect(remove.status).toBe(204);
const tombstoneRow = await prismaCn.outbox_events.findFirstOrThrow({
where: { type: 'category.deleted', albumId },
});
expect(tombstoneRow.entityVersion).toBe(3);
await drainSender();
replica = await prismaOv.albums_ov.findUniqueOrThrow({
where: { albumId },
});
expect(replica.categories).toEqual([]);
});
});