3681999667
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>
142 lines
5.5 KiB
TypeScript
142 lines
5.5 KiB
TypeScript
// 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();
|
||
}
|
||
}
|