Files
photo-live/services/api/test/reset.ts
T
浅念 3681999667 feat(api): scheduled user sync with mock source (T30 semantics)
P4a Task 7 —— cn 侧用户定期同步(mock 源先行,G4 真实接口到位后仅替换
UserSourcePort provider,引擎/契约/表不变)。

数据模型(db-cn 迁移 20260920114149,utf8mb4_0900_bin 逐表校对):
- user_sync_runs:批次账本(trigger cron|manual、state running|success|
  partial|failed、cursor 水位、counts JSON、起止时间与错误摘要);
- user_sync_errors:单批次坏行明细(runId 索引);
- external_user_mappings:本地↔外部 ID 映射(@@unique([sourceSystem,
  externalUserId]),sourceVersion 逐行高水位);
- user_sync_state:单行状态(lastRunAt/lastSuccessAt/cursor/runCount)。

同步引擎(UserSyncEngine):
- Redis SETNX 分布式互斥(TTL 10min,锁值=runId,Lua CAS 释放);
  定时 tick 与手动触发共用,运行中批次复用 runId(202);
- 游标增量 + 周期全量对账(RECONCILE_EVERY_N_RUNS 默认 10);
- 逐页单事务提交(run counts+cursor / state cursor),分页中断→failed
  且水位不前进、会话不动;
- 红线遵守:sync 不写 password_hash/role/album_members/localLocked;
  空/部分页不误伤停用;上游恢复不解除本地锁定;
- 新用户按 pending 建(随机不可用哨兵哈希,激活+重置后方可登录);
- 停用即撤销该用户全部会话;本地预建同名账号走关联而非重建。

Admin API(/admin/user-sync):GET status|runs|errors?runId=、POST runs
(202 {runId});runs 为 startedAt+ID 倒序游标分页。

auth.service:登录对哨兵/损坏哈希加固(等代价 dummy verify,
401 而非 500,时序平坦)。

测试:user-sync.e2e-spec 12 用例(首轮全量/改名幂等/停用撤会话/
恢复不解锁/新增/空页/分页中断/部分无效/手动互斥/增量+对账/API 形状
与权限/pending 登录门);api e2e 22 套件 239 用例全绿。

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-20 20:01:17 +08:00

142 lines
5.5 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/api/test/reset.ts
// e2e 共用隔离工具:每个 e2e spec 的 beforeAll 调用 resetDb(),防止 spec 间
// 唯一约束冲突与脏数据。
// 实现:SET FOREIGN_KEY_CHECKS=0 包裹逐表 TRUNCATE(users 被 sessions/audit_logs
// 引用,即便子表已清空,存在 FK 约束本身即令 TRUNCATE 失败,必须关检查);
// 表序仍按 FK 安全顺序排列(子表在前),关检查失败时也不至于中途卡死。
// 顺序 await 在 Prisma 连接池上复用同一空闲连接,会话级 FOREIGN_KEY_CHECKS
// 对后续 TRUNCATE 生效;finally 恢复检查并断开,失败也不泄漏连接。
// Task 18 起 e2e 会向 MinIO 写入真实对象(staging/originals):同一函数内追加
// `photo-live/dev` 前缀对象清理(reset.ts 文件头 TODO 的落实——只清本前缀,
// packages/storage 的 test/<ts>-<uuid>/ 前缀对象不受影响,两包测试可并行)。
import {
DeleteObjectsCommand,
ListObjectsV2Command,
S3Client,
} from '@aws-sdk/client-s3';
import {
PrismaClient,
USAGE_THRESHOLDS_SINGLETON_ID,
USER_SYNC_STATE_SINGLETON_ID,
} from '@photo/db-cn';
export const RESET_TABLES: readonly string[] = [
// 照片与上传域 + outbox(Task 10):均为扁平字符串外键(无 FK 约束),
// 顺序按逻辑依赖排列(事件/作业/变体/会话先于 photos)。
'outbox_events',
'photo_jobs',
'photo_variants',
'upload_sessions',
'photos',
// 用户同步域(P4a Task 7):坏行明细先于批次账本;单行状态重种在
// resetDb 尾部;映射表为扁平字符串外键(无 FK),先于 users 仅为逻辑可读。
// 注意:Redis 互斥锁(user-sync:lock)不在本函数清理范围——e2e spec 自行
// 清(仅用户同步 spec 触碰触发路径)。
'user_sync_errors',
'user_sync_runs',
'user_sync_state',
'external_user_mappings',
// 用量域(P4a Task 5):告警账本先于阈值单行(重种在 resetDb 尾部)。
'usage_alerts',
'usage_thresholds',
// 水印域(P4a Task 3):任务/配置账本先于 albums(watermarkVersion 等位列
// 随 albums 清零)。
'watermark_jobs',
'watermark_configs',
// 相册域(Task 9):子表(翻译/资产/成员)在前,albums 在后;相册域为扁平
// 字符串外键(无 FK 约束),此处顺序仅为逻辑可读,不依赖 FK 检查开关。
'category_translations',
'categories',
'album_assets',
'album_translations',
'album_members',
'albums',
// 身份域(Task 8)。
'audit_logs',
'sessions',
'users',
];
/** e2e 对象隔离边界:与 src/uploads/support.ts 的 OBJECT_KEY_ROOT 一致。 */
const RESET_OBJECT_PREFIX = 'photo-live/dev/';
/** 删除 MinIO 中 RESET_OBJECT_PREFIX 前缀的全部对象(分页列举 + 批量删除)。 */
async function resetMinioObjects(): Promise<void> {
const required = [
'S3_ENDPOINT',
'S3_REGION',
'S3_ACCESS_KEY',
'S3_SECRET_KEY',
'S3_BUCKET',
] as const;
const missing = required.filter((key) => !process.env[key]);
if (missing.length > 0) {
throw new Error(`resetMinioObjects: 缺少环境变量 ${missing.join(', ')}`);
}
const client = new S3Client({
endpoint: process.env.S3_ENDPOINT,
region: process.env.S3_REGION,
credentials: {
accessKeyId: process.env.S3_ACCESS_KEY as string,
secretAccessKey: process.env.S3_SECRET_KEY as string,
},
forcePathStyle: true,
});
const bucket = process.env.S3_BUCKET as string;
try {
for (;;) {
const listed = await client.send(
new ListObjectsV2Command({
Bucket: bucket,
Prefix: RESET_OBJECT_PREFIX,
MaxKeys: 1000,
}),
);
const objects = (listed.Contents ?? []).filter(
(obj) => typeof obj.Key === 'string',
);
if (objects.length === 0) {
break;
}
await client.send(
new DeleteObjectsCommand({
Bucket: bucket,
Delete: {
Objects: objects.map((obj) => ({ Key: obj.Key as string })),
Quiet: true,
},
}),
);
if (!listed.IsTruncated) {
break;
}
}
} finally {
await client.destroy();
}
}
export async function resetDb(): Promise<void> {
await resetMinioObjects();
const prisma = new PrismaClient();
try {
await prisma.$executeRawUnsafe('SET FOREIGN_KEY_CHECKS = 0');
for (const table of RESET_TABLES) {
await prisma.$executeRawUnsafe(`TRUNCATE TABLE ${table}`);
}
// 用量阈值单行被 TRUNCATE 后重种(迁移同值种子;保证「单行恒存在」
// 不变量对后续 spec 成立——发布路径的评估检查点依赖它,缺行时兜底
// 视作双维停用,跨 spec 的阈值残留则被清除)。
await prisma.$executeRaw`INSERT INTO usage_thresholds (id, storageBytes, perAlbumBytes, updatedAt)
VALUES (${USAGE_THRESHOLDS_SINGLETON_ID}, NULL, NULL, CURRENT_TIMESTAMP(6))`;
// 用户同步单行状态同理重种(status 端点直读本行;缺行 = 从未运行)。
// 列名恒反引号:`cursor` 是 MySQL 保留字(`trigger` 同理,Prisma 客户端
// 查询自带反引号,raw SQL 必须显式)。
await prisma.$executeRaw`INSERT INTO user_sync_state (\`id\`, \`lastRunAt\`, \`lastSuccessAt\`, \`cursor\`, \`runCount\`, \`updatedAt\`)
VALUES (${USER_SYNC_STATE_SINGLETON_ID}, NULL, NULL, NULL, 0, CURRENT_TIMESTAMP(6))`;
} finally {
await prisma.$executeRawUnsafe('SET FOREIGN_KEY_CHECKS = 1');
await prisma.$disconnect();
}
}