From 3681999667d325947d642fed31dcbe32f3070a73 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Sun, 20 Sep 2026 20:01:17 +0800 Subject: [PATCH] feat(api): scheduled user sync with mock source (T30 semantics) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../migration.sql | 70 ++ packages/db-cn/prisma/schema.prisma | 55 ++ packages/db-cn/src/index.ts | 3 + services/api/src/app-cn.module.ts | 4 + services/api/src/auth/auth.service.ts | 13 +- .../api/src/user-sync/mock-user-source.ts | 180 +++++ services/api/src/user-sync/sync-lock.ts | 83 ++ .../api/src/user-sync/user-source.port.ts | 45 ++ .../src/user-sync/user-sync-admin.service.ts | 176 +++++ .../api/src/user-sync/user-sync.controller.ts | 81 ++ .../api/src/user-sync/user-sync.module.ts | 28 + .../api/src/user-sync/user-sync.service.ts | 527 +++++++++++++ services/api/test/reset.ts | 14 + services/api/test/user-sync.e2e-spec.ts | 726 ++++++++++++++++++ 14 files changed, 2004 insertions(+), 1 deletion(-) create mode 100644 packages/db-cn/prisma/migrations/20260920114149_user_sync_domain/migration.sql create mode 100644 services/api/src/user-sync/mock-user-source.ts create mode 100644 services/api/src/user-sync/sync-lock.ts create mode 100644 services/api/src/user-sync/user-source.port.ts create mode 100644 services/api/src/user-sync/user-sync-admin.service.ts create mode 100644 services/api/src/user-sync/user-sync.controller.ts create mode 100644 services/api/src/user-sync/user-sync.module.ts create mode 100644 services/api/src/user-sync/user-sync.service.ts create mode 100644 services/api/test/user-sync.e2e-spec.ts diff --git a/packages/db-cn/prisma/migrations/20260920114149_user_sync_domain/migration.sql b/packages/db-cn/prisma/migrations/20260920114149_user_sync_domain/migration.sql new file mode 100644 index 0000000..c071b5c --- /dev/null +++ b/packages/db-cn/prisma/migrations/20260920114149_user_sync_domain/migration.sql @@ -0,0 +1,70 @@ +-- P4a Task 7：用户同步域（A29/T30 语义，mock 源先行）。 +-- 校对手决：Prisma 生成 DDL 固定 utf8mb4_unicode_ci，逐表改写为库级约定的 +-- utf8mb4_0900_bin（sourceSystem/externalUserId/code 等字符串列的大小写 +-- 敏感唯一/比较语义依赖它）。 +-- · user_sync_runs.cursor 为 VarChar(128)：G4 真实源可能出现非数值游标； +-- mock 源恒为数值串（源版本高水位），API 层投影为契约 int64。 +-- · external_user_mappings.userId 为扁平字符串外键（无 FK，同相册域约定： +-- 用户注销等跨域删除语义由应用层决策，且减轻 TRUNCATE 隔离负担）。 +-- · user_sync_state 单行状态表：迁移内种子行（固定 id 见 +-- USER_SYNC_STATE_SINGLETON_ID；reset.ts TRUNCATE 后重种同值）。 +-- `trigger` 为 MySQL 保留字：Prisma 查询恒反引号标识符，本迁移同样反引号。 + +-- CreateTable +CREATE TABLE `user_sync_runs` ( + `id` VARCHAR(191) NOT NULL, + `trigger` VARCHAR(16) NOT NULL, + `state` VARCHAR(16) NOT NULL DEFAULT 'running', + `cursor` VARCHAR(128) NULL, + `counts` JSON NOT NULL, + `startedAt` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + `finishedAt` DATETIME(6) NULL, + `error` VARCHAR(1024) NULL, + + INDEX `user_sync_runs_startedAt_idx`(`startedAt`), + PRIMARY KEY (`id`) +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin; + +-- CreateTable +CREATE TABLE `user_sync_errors` ( + `id` VARCHAR(191) NOT NULL, + `runId` VARCHAR(64) NOT NULL, + `externalUserId` VARCHAR(64) NULL, + `code` VARCHAR(64) NOT NULL, + `detail` VARCHAR(1024) NOT NULL, + `createdAt` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + + INDEX `user_sync_errors_runId_idx`(`runId`), + PRIMARY KEY (`id`) +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin; + +-- CreateTable +CREATE TABLE `external_user_mappings` ( + `id` VARCHAR(191) NOT NULL, + `userId` VARCHAR(64) NOT NULL, + `sourceSystem` VARCHAR(32) NOT NULL DEFAULT 'mock', + `externalUserId` VARCHAR(64) NOT NULL, + `sourceVersion` INTEGER NOT NULL DEFAULT 0, + `createdAt` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + + INDEX `external_user_mappings_userId_idx`(`userId`), + UNIQUE INDEX `external_user_mappings_sourceSystem_externalUserId_key`(`sourceSystem`, `externalUserId`), + PRIMARY KEY (`id`) +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin; + +-- CreateTable +CREATE TABLE `user_sync_state` ( + `id` VARCHAR(191) NOT NULL, + `lastRunAt` DATETIME(6) NULL, + `lastSuccessAt` DATETIME(6) NULL, + `cursor` VARCHAR(128) NULL, + `runCount` INTEGER NOT NULL DEFAULT 0, + `updatedAt` DATETIME(6) NOT NULL, + + PRIMARY KEY (`id`) +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin; + +-- 单行种子（引擎 upsert 兜底，但迁移种子保证常规路径下不变量成立—— +-- status 端点直接读本行，缺行视作「从未运行」）。 +INSERT INTO `user_sync_state` (`id`, `lastRunAt`, `lastSuccessAt`, `cursor`, `runCount`, `updatedAt`) +VALUES ('user-sync-state-single-row', NULL, NULL, NULL, 0, CURRENT_TIMESTAMP(6)); diff --git a/packages/db-cn/prisma/schema.prisma b/packages/db-cn/prisma/schema.prisma index df0e274..e7f53c6 100644 --- a/packages/db-cn/prisma/schema.prisma +++ b/packages/db-cn/prisma/schema.prisma @@ -277,3 +277,58 @@ model usage_alerts { resolvedAt DateTime? @db.DateTime(6) // 非空=已解除（阈值上调/用量回落/维度停用） @@index([resolvedAt]) } +// —— 用户同步域（P4a Task 7，A29/T30 语义，mock 源先行）—— +// 三条红线（技术方案 3.1A）：同步绝不写 password_hash/role/album_members/ +// localLocked；空响应/部分页绝不停用任何人；上游恢复（disabled→active） +// 不解除本地锁定。外部身份经 external_user_mappings 隔离（稳定外部 ID 锚点， +// 不在 users 上加列）。 +// · user_sync_runs：批次账本。cursor = 已提交源水位（VarChar 承载 G4 真实 +// 源可能出现的非数值游标；mock 源恒为数值串，API 层投影为契约 int）， +// 逐页提交（分页中断时停在最后成功页）；counts Json = {added,updated, +// disabled,failed}；state partial=有坏行但其余照常应用。 +// · user_sync_errors：坏行明细（partial/failed 批次的证据链）。 +// · external_user_mappings：本地用户 ↔ 外部身份。userId 为扁平字符串外键 +// （同相册域约定：跨域删除语义由应用层决策，且减轻 TRUNCATE 隔离负担）。 +// · user_sync_state：单行引擎状态（lastRunAt/lastSuccessAt/已提交水位/ +// runCount——runCount 驱动「每 N 批强制全量对账」节奏）。 +model user_sync_runs { + id String @id @default(cuid()) + trigger String @db.VarChar(16) // cron|manual（MySQL 保留字，Prisma 恒反引号） + state String @default("running") @db.VarChar(16) // running|success|partial|failed + cursor String? @db.VarChar(128) // 已提交源水位；未提交为 null + counts Json // {added,updated,disabled,failed} + startedAt DateTime @default(now()) @db.DateTime(6) + finishedAt DateTime? @db.DateTime(6) + error String? @db.VarChar(1024) // 批次级失败摘要（分页中断等） + + @@index([startedAt]) +} +model user_sync_errors { + id String @id @default(cuid()) + runId String @db.VarChar(64) + externalUserId String? @db.VarChar(64) // 坏行可能连外部 ID 都不合规 → 可空 + code String @db.VarChar(64) // invalid_record|apply_failed|... + detail String @db.VarChar(1024) + createdAt DateTime @default(now()) @db.DateTime(6) + + @@index([runId]) +} +model external_user_mappings { + id String @id @default(cuid()) + userId String @db.VarChar(64) // 扁平字符串外键（无 relation/FK，同相册域） + sourceSystem String @default("mock") @db.VarChar(32) // G4 后换真实源标识 + externalUserId String @db.VarChar(64) + sourceVersion Int @default(0) // 幂等比较位：仅更高版本应用 + createdAt DateTime @default(now()) @db.DateTime(6) + + @@unique([sourceSystem, externalUserId]) + @@index([userId]) +} +model user_sync_state { + id String @id // 固定单行 id（USER_SYNC_STATE_SINGLETON_ID；迁移种子） + lastRunAt DateTime? @db.DateTime(6) + lastSuccessAt DateTime? @db.DateTime(6) // 仅 state=success 批次推进（partial 不算完全成功） + cursor String? @db.VarChar(128) // 已提交源水位（与 runs.cursor 同形） + runCount Int @default(0) // 已启动批次数（含失败），驱动周期全量对账 + updatedAt DateTime @updatedAt @db.DateTime(6) +} diff --git a/packages/db-cn/src/index.ts b/packages/db-cn/src/index.ts index cd5595f..f1ef722 100644 --- a/packages/db-cn/src/index.ts +++ b/packages/db-cn/src/index.ts @@ -17,3 +17,6 @@ export type { UsageAggregates, UsageAlertKind, } from './usage'; +// P4a Task 7：用户同步域单行状态表固定 id（迁移种子；reset.ts TRUNCATE 后 +// 重种同值。引擎与 e2e 共用，避免字面量漂移）。 +export const USER_SYNC_STATE_SINGLETON_ID = 'user-sync-state-single-row'; diff --git a/services/api/src/app-cn.module.ts b/services/api/src/app-cn.module.ts index ee8d8dd..7ee9392 100644 --- a/services/api/src/app-cn.module.ts +++ b/services/api/src/app-cn.module.ts @@ -25,6 +25,7 @@ import { PublicModule } from './public/public.module'; import { RequestIdMiddleware } from './request-id.middleware'; import { SyncModule } from './sync/sync.module'; import { UploadsModule } from './uploads/uploads.module'; +import { UserSyncModule } from './user-sync/user-sync.module'; // P2 Task 2：logger 装配选项（genReqId/redact/level）收敛至 shared，与 ov 入口 // 共用单一来源；此处仅消费，行为不变（P1 e2e 套件守护）。 import { createLoggerOptions } from './shared/logger-options'; @@ -48,6 +49,9 @@ import { createLoggerOptions } from './shared/logger-options'; // 内部同步域（P2 Task 3：/internal/sync/bootstrap|reconcile，HMAC 鉴权， // 仅 cn 入口装配——cn 是内容同步的源端）。 SyncModule, + // 用户同步域（P4a Task 7：mock 源定期同步 + /admin/user-sync 管理端点 + + // setInterval 调度器——仅 cn 装配，ov 入口不碰）。 + UserSyncModule, LoggerModule.forRoot(createLoggerOptions()), ], controllers: [HealthController], diff --git a/services/api/src/auth/auth.service.ts b/services/api/src/auth/auth.service.ts index 8de55da..11f62f7 100644 --- a/services/api/src/auth/auth.service.ts +++ b/services/api/src/auth/auth.service.ts @@ -132,7 +132,18 @@ export class AuthService { // ③ 先验密码再查状态：状态不合格与密码错误同样不可区分（同 401 同文案； // argon2 在两条路径都真实跑过，时延也一致）。 - if (!(await verifyPassword(user.passwordHash, plainPassword))) { + // 哨兵哈希（P4a Task 7：同步创建的待激活账号 passwordHash='!:pending:…'） + // 不是合法 argon2 编码——verify 会抛错而非返回 false。此处按密码错误 + // 处理并补跑等代价 dummy verify（时延与密码错误路径一致，不泄露 + // 「该账号待激活」），杜绝 500。 + let passwordOk: boolean; + try { + passwordOk = await verifyPassword(user.passwordHash, plainPassword); + } catch { + await verifyPassword(await dummyHash(), plainPassword); + passwordOk = false; + } + if (!passwordOk) { this.rateLimiter.recordFailure(rateKey); throw invalidCredentials(); } diff --git a/services/api/src/user-sync/mock-user-source.ts b/services/api/src/user-sync/mock-user-source.ts new file mode 100644 index 0000000..6057412 --- /dev/null +++ b/services/api/src/user-sync/mock-user-source.ts @@ -0,0 +1,180 @@ +// services/api/src/user-sync/mock-user-source.ts —— MockUserSource（P4a Task 7； +// G4：真实接口到位后仅替换本 provider，引擎/契约/表不变）。 +// 场景可编程注入：e2e 经 app.get(MockUserSource) 直接改 scenario（比 env 开关 +// 更细粒度——单 spec 内可逐用例切换分页/版本/改名/停用/新增/空页/门闸/中断 +// 模式）；生产默认 = 健康场景（稳定演示用户集，见 DEFAULT_SCENARIO）。 +// +// 游标协议（引擎侧不透明，此处为权威定义）： +// · cursor=null/undefined → 全量快照第 0 页（首轮同步与周期全量对账）； +// · cursor="W"（纯数值串，引擎传入已提交源水位）→ 增量第 0 页： +// 仅返回 sourceVersion > W 的行； +// · cursor="W#P" →（增量基数 W 的）第 P 页续翻； +// · cursor="*#P" →（全量的）第 P 页续翻（'*' 基数与数值水位区分）； +// · nextCursor：还有后续页时为 `<基数>#`（增量基数=数值 W、全量 +// 基数='*'），本页即末页时为 null。 +// · 水位基数 W 在一轮内保持首调用值——引擎逐页提交的高水位（行内最大 +// sourceVersion）与续翻游标互不干扰。 +// 行序：sourceVersion 升序、同版本按场景数组序（确定性分页）。 +// 故障注入：failFromPage（服务第 ≥ 该下标的页时抛错 = 分页中断）； +// gateFirstPage（第 0 页等待该 Promise，e2e 用它制造「运行中」批次以测互斥）。 +// receivedCursors 记录每次 listPage 收到的游标（e2e 断言增量/全量形态）。 +import { Injectable } from '@nestjs/common'; +import type { ExternalUserRecord, UserSourcePage, UserSourcePort } from './user-source.port'; + +/** 场景行：全字段可选——坏行（缺 displayName/非法 sourceStatus 等）原样透传，校验在引擎。 */ +export interface MockExternalRecord { + externalUserId?: string; + loginAccount?: string; + displayName?: string; + department?: string; + sourceStatus?: string; + sourceVersion?: number; + updatedAt?: Date; +} + +export interface MockScenario { + users: MockExternalRecord[]; + /** 每页行数（默认 2：3 用户即两页，分页语义最小可测）。 */ + pageSize?: number; + /** 从该页下标起 listPage 抛错（分页中断注入；≥1 才有意义，第 0 页恒成功）。 */ + failFromPage?: number; + /** 第 0 页等待此 Promise（互斥测试的运行中批次制造器）。 */ + gateFirstPage?: Promise; +} + +/** 生产默认健康场景：3 个稳定演示用户（mock 源先行，A29）。 */ +export const DEFAULT_SCENARIO: MockScenario = { + users: [ + { + externalUserId: 'mock-u-001', + loginAccount: 'demo.li', + displayName: '李演示', + department: '运营部', + sourceStatus: 'active', + sourceVersion: 1, + updatedAt: new Date('2026-09-20T00:00:00Z'), + }, + { + externalUserId: 'mock-u-002', + loginAccount: 'demo.wang', + displayName: '王演示', + department: '摄制部', + sourceStatus: 'active', + sourceVersion: 1, + updatedAt: new Date('2026-09-20T00:00:00Z'), + }, + { + externalUserId: 'mock-u-003', + loginAccount: 'demo.zhao', + displayName: '赵演示', + department: '摄制部', + sourceStatus: 'active', + sourceVersion: 1, + updatedAt: new Date('2026-09-20T00:00:00Z'), + }, + ], +}; + +interface ParsedCursor { + /** null = 全量快照（首轮同步 / 周期对账）；数值 = 增量基数（仅 v > W）。 */ + watermark: number | null; + page: number; +} + +/** + * null/"" → 全量第 0 页；"W" → 增量第 0 页（仅 v > W）；"W#P" → 增量第 P 页； + * "*#P" → 全量第 P 页（全量续翻编码；基数 '*' 与数值水位区分）。 + */ +function parseCursor(raw: string | undefined): ParsedCursor { + if (raw === undefined || raw === '') { + return { watermark: null, page: 0 }; + } + const hash = raw.indexOf('#'); + const base = hash === -1 ? raw : raw.slice(0, hash); + const pagePart = hash === -1 ? '0' : raw.slice(hash + 1); + const page = Number.parseInt(pagePart, 10); + if (!Number.isInteger(page) || page < 0) { + return { watermark: null, page: 0 }; + } + if (base === '*') { + return { watermark: null, page }; + } + const watermark = Number.parseInt(base, 10); + if (!Number.isInteger(watermark) || watermark < 0) { + return { watermark: null, page: 0 }; + } + return { watermark, page }; +} + +/** 续翻游标编码：增量携带基数 W，全量用 '*' 基数。 */ +function encodePageCursor(watermark: number | null, page: number): string { + return `${watermark === null ? '*' : watermark}#${page}`; +} + +@Injectable() +export class MockUserSource implements UserSourcePort { + private scenario: MockScenario = DEFAULT_SCENARIO; + /** e2e 断言用：历次 listPage 收到的游标（null=全量形态）。 */ + readonly receivedCursors: Array = []; + + /** 整场景替换（浅拷贝隔离，避免用例间通过引用串改）。 */ + setScenario(next: MockScenario): void { + this.scenario = { ...next }; + this.receivedCursors.length = 0; + } + + /** 恢复生产默认健康场景。 */ + reset(): void { + this.setScenario(DEFAULT_SCENARIO); + } + + async listPage(cursor?: string): Promise { + const parsed = parseCursor(cursor); + this.receivedCursors.push(cursor ?? null); + + if (parsed.page === 0 && this.scenario.gateFirstPage !== undefined) { + await this.scenario.gateFirstPage; + } + if ( + this.scenario.failFromPage !== undefined && + parsed.page >= this.scenario.failFromPage + ) { + throw new Error( + `mock upstream failure at page ${parsed.page}（分页中断注入）`, + ); + } + + // 增量基数即游标携带的 W（首轮/对账 null → 全量：不做版本过滤）。 + const { watermark } = parsed; + const ordered = [...this.scenario.users] + .map((user, index) => ({ user, index })) + .sort((a, b) => { + const va = a.user.sourceVersion ?? 0; + const vb = b.user.sourceVersion ?? 0; + return va !== vb ? va - vb : a.index - b.index; + }) + .filter( + ({ user }) => + watermark === null || (user.sourceVersion ?? 0) > watermark, + ); + + const pageSize = this.scenario.pageSize ?? 2; + const start = parsed.page * pageSize; + const pageRecords = ordered.slice(start, start + pageSize); + const hasMore = start + pageSize < ordered.length; + + const items: ExternalUserRecord[] = pageRecords.map(({ user }) => ({ + externalUserId: user.externalUserId ?? '', + loginAccount: user.loginAccount, + displayName: user.displayName ?? '', + department: user.department, + sourceStatus: user.sourceStatus as ExternalUserRecord['sourceStatus'], + sourceVersion: user.sourceVersion ?? Number.NaN, + updatedAt: user.updatedAt ?? new Date(), + })); + return { + nextCursor: hasMore ? encodePageCursor(watermark, parsed.page + 1) : null, + items, + }; + } +} diff --git a/services/api/src/user-sync/sync-lock.ts b/services/api/src/user-sync/sync-lock.ts new file mode 100644 index 0000000..3904be9 --- /dev/null +++ b/services/api/src/user-sync/sync-lock.ts @@ -0,0 +1,83 @@ +// services/api/src/user-sync/sync-lock.ts —— 用户同步互斥锁（P4a Task 7）。 +// Redis（cn，REDIS_URL_CN）SETNX + TTL + 值比对释放（Lua CAS）： +// · 锁值 = 进行中批次 runId——手动补拉撞锁时直接复用该批次（契约 +// UserSyncRunCreated 语义），无需二段查询； +// · TTL 10 分钟：进程崩溃后的自愈上界（mock 全量分钟级完成，TTL 远大于 +// 单批时长；TTL 中途过期导致的僵尸批次并发是接受的边界——登记偏差）； +// · 释放 = Lua「值等于自己的 runId 才 DEL」：绝不误删他人批次的新锁。 +// 连接策略与 watermark-queue.provider 同一 lazy-safe 约定：首次使用才建 +// ioredis 连接；'error' 挂 no-op 监听（未监听的 error 事件崩进程）； +// onModuleDestroy quit。Redis 整体不可用时锁不可得 → 同步不启动（fail-closed， +// 宁可不同步也不并发跑批次）；status/runs/errors 查询纯 DB，不受影响。 +import { Injectable, Logger, type OnModuleDestroy } from '@nestjs/common'; +import IORedis from 'ioredis'; + +/** 锁键（cn Redis；同步域唯一互斥点）。 */ +export const USER_SYNC_LOCK_KEY = 'user-sync:lock'; + +/** 锁 TTL：崩溃自愈上界（毫秒）。 */ +const LOCK_TTL_MS = 10 * 60 * 1000; + +/** 值比对释放（CAS）：只有锁值仍是自己的 runId 才删除。 */ +const RELEASE_SCRIPT = ` +if redis.call('GET', KEYS[1]) == ARGV[1] then + return redis.call('DEL', KEYS[1]) +else + return 0 +end +`; + +@Injectable() +export class SyncLock implements OnModuleDestroy { + private readonly logger = new Logger(SyncLock.name); + private connection: IORedis | null = null; + + private getClient(): IORedis { + if (this.connection === null) { + const url = process.env.REDIS_URL_CN ?? 'redis://localhost:6379'; + this.connection = new IORedis(url, { + maxRetriesPerRequest: 1, // 锁操作快速失败：Redis 抖动时本轮同步放弃 + retryStrategy: (times) => Math.min(times * 200, 2000), + }); + this.connection.on('error', (err: Error) => { + this.logger.warn(`同步锁连接异常（将由 ioredis 自动重连）: ${err.message}`); + }); + } + return this.connection; + } + + /** 尝试获取锁（锁值 = runId）；成功 true，已被持有 false。 */ + async acquire(runId: string): Promise { + const result = await this.getClient().set( + USER_SYNC_LOCK_KEY, + runId, + 'PX', + LOCK_TTL_MS, + 'NX', + ); + return result === 'OK'; + } + + /** 当前锁值（持有中的 runId）；无锁为 null。 */ + async currentRunId(): Promise { + const value = await this.getClient().get(USER_SYNC_LOCK_KEY); + return value; + } + + /** 值比对删除（仅锁值仍为 runId 时）；用于陈旧锁清理与正常释放。 */ + async releaseIfValue(runId: string): Promise { + await this.getClient().eval( + RELEASE_SCRIPT, + 1, + USER_SYNC_LOCK_KEY, + runId, + ); + } + + async onModuleDestroy(): Promise { + if (this.connection !== null) { + await this.connection.quit().catch(() => undefined); + this.connection = null; + } + } +} diff --git a/services/api/src/user-sync/user-source.port.ts b/services/api/src/user-sync/user-source.port.ts new file mode 100644 index 0000000..aefd147 --- /dev/null +++ b/services/api/src/user-sync/user-source.port.ts @@ -0,0 +1,45 @@ +// services/api/src/user-sync/user-source.port.ts —— 用户源适配器接口 +// （P4a Task 7，A29；G4：真实接口到位后仅替换 provider 实现，引擎/契约不变）。 +// 方法签名与 D08 预期对齐：游标分页的 listPage。游标对引擎是不透明字符串—— +// 语义由源定义： +// · cursor=null → 全量快照（首轮同步与周期全量对账共用此形态）； +// · cursor="W"（已提交源水位）→ 增量：仅返回版本高于 W 的行（mock 语义； +// 真实源为变更游标）； +// · 页内续翻游标（nextCursor 回传值）由源自行编码（mock 用 "W#P"）。 +// 引擎侧不变量（技术方案 3.1A 红线，见 user-sync.service.ts 文件头）： +// · items 为空且 nextCursor=null → 正常结束，绝不停用任何本地用户； +// · listPage 抛错 → 分页中断：批次 failed，水位停在最后成功页。 +// department 字段：users 表无此列，端口保留以对齐真实源自形状，引擎忽略 +// （登记偏差，见验收记录）。 + +/** 外部用户行（源侧形状；字段可能不合规——校验是引擎的职责，见 validateItem）。 */ +export interface ExternalUserRecord { + externalUserId: string; + /** 登录账号；缺省时引擎回退用 externalUserId 作账号。 */ + loginAccount?: string; + displayName: string; + department?: string; + /** 上游状态：disabled → users.sourceStatus=disabled + 撤全部会话。 */ + sourceStatus: 'active' | 'disabled'; + /** 幂等比较位：仅高于映射行 sourceVersion 的行会被应用。 */ + sourceVersion: number; + updatedAt: Date; +} + +export interface UserSourcePage { + /** null = 本轮分页结束（非 null 时引擎以该值续翻下一页）。 */ + nextCursor: string | null; + items: ExternalUserRecord[]; +} + +/** 用户源适配器端口（mock 实现见 mock-user-source.ts；G4 后替换为真实适配器）。 */ +export interface UserSourcePort { + /** + * 拉取一页。cursor 语义见文件头；抛错 = 分页中断（引擎记 failed 批次， + * 不动任何用户，水位不前进）。 + */ + listPage(cursor?: string): Promise; +} + +/** DI token（user-sync.module 装配 MockUserSource；e2e 经 app.get 注入场景）。 */ +export const USER_SOURCE = Symbol('USER_SOURCE'); diff --git a/services/api/src/user-sync/user-sync-admin.service.ts b/services/api/src/user-sync/user-sync-admin.service.ts new file mode 100644 index 0000000..c32b0d6 --- /dev/null +++ b/services/api/src/user-sync/user-sync-admin.service.ts @@ -0,0 +1,176 @@ +// services/api/src/user-sync/user-sync-admin.service.ts —— 同步管理查询服务 +// （P4a Task 7；契约 getUserSyncStatus / listUserSyncRuns / listUserSyncErrors， +// tag [admin]，全部仅 admin——403 AUTH_FORBIDDEN）。纯 DB 读： +// · status：单行状态（lastRunAt/lastSuccessAt/已提交水位）+ 最近批次的 +// state/counts；引擎状态 = 无批次 idle、最近批次 running 或其终态； +// · runs：startedAt+ID 稳定倒序键集游标（shared/cursor，同管理列表先例）； +// · errors：单批次坏行明细（createdAt+ID 倒序；runId 必填，未知批次=空数组 +// ——契约无 404，空列表即「无失败行」与「批次不存在」的共同安全答案）。 +// Redis 故障不影响本服务（互斥锁只在触发/执行路径上，见 sync-lock.ts）。 +import { HttpStatus, Injectable } from '@nestjs/common'; +import { ErrorCodes, type components } from '@photo/contracts'; +import { ApiException } from '../api-exception'; +import type { AuthUser } from '../auth/session.guard'; +import { PrismaService } from '../prisma/prisma.service'; +import { + decodeKeysetCursor, + encodeKeysetCursor, + type KeysetCursor, +} from '../shared/cursor'; +import { validationFailed } from '../albums/support'; +import { USER_SYNC_STATE_SINGLETON_ID } from '@photo/db-cn'; +import { watermarkToInt, type UserSyncCounts } from './user-sync.service'; + +export type UserSyncStatusResponse = components['schemas']['UserSyncStatus']; +export type UserSyncRunListResponse = components['schemas']['UserSyncRunList']; +export type UserSyncRunItem = components['schemas']['UserSyncRun']; +export type UserSyncErrorListResponse = components['schemas']['UserSyncErrorList']; +export type UserSyncStateValue = components['schemas']['UserSyncState']; + +/** 坏行明细单批读取上限（契约无分页；防异常场景拖垮响应的护栏）。 */ +const ERRORS_LIST_LIMIT = 500; + +function requireAdmin(user: AuthUser): void { + if (user.role !== 'admin') { + throw new ApiException( + ErrorCodes.AUTH_FORBIDDEN, + '仅管理员可查看用户同步信息', + HttpStatus.FORBIDDEN, + ); + } +} + +/** counts Json 列 → 契约形状（缺字段兜底 0——历史行防御）。 */ +function toCounts(raw: unknown): UserSyncCounts { + const source = (raw ?? {}) as Record; + return { + added: typeof source.added === 'number' ? source.added : 0, + updated: typeof source.updated === 'number' ? source.updated : 0, + disabled: typeof source.disabled === 'number' ? source.disabled : 0, + failed: typeof source.failed === 'number' ? source.failed : 0, + }; +} + +type RunRow = { + id: string; + trigger: string; + state: string; + cursor: string | null; + counts: unknown; + startedAt: Date; + finishedAt: Date | null; + error: string | null; +}; + +function toRunItem(row: RunRow): UserSyncRunItem { + return { + id: row.id, + trigger: row.trigger as UserSyncRunItem['trigger'], + state: row.state as UserSyncRunItem['state'], + cursor: watermarkToInt(row.cursor), + counts: toCounts(row.counts), + startedAt: row.startedAt.toISOString(), + finishedAt: row.finishedAt?.toISOString() ?? null, + error: row.error, + }; +} + +@Injectable() +export class UserSyncAdminService { + constructor(private readonly prisma: PrismaService) {} + + /** 引擎当前状态 + 最近批次计数（getUserSyncStatus）。 */ + async status(user: AuthUser): Promise { + requireAdmin(user); + const [stateRow, latestRun] = await Promise.all([ + this.prisma.user_sync_state.findUnique({ + where: { id: USER_SYNC_STATE_SINGLETON_ID }, + }), + this.prisma.user_sync_runs.findFirst({ + orderBy: [{ startedAt: 'desc' }, { id: 'desc' }], + }), + ]); + const state: UserSyncStateValue = + latestRun === null + ? 'idle' + : latestRun.state === 'running' + ? 'running' + : (latestRun.state as UserSyncStateValue); + return { + lastRunAt: stateRow?.lastRunAt?.toISOString() ?? null, + lastSuccessAt: stateRow?.lastSuccessAt?.toISOString() ?? null, + state, + sourceVersion: watermarkToInt(stateRow?.cursor ?? null), + counts: latestRun === null ? null : toCounts(latestRun.counts), + }; + } + + /** 运行历史（listUserSyncRuns）：startedAt+ID 倒序键集游标分页。 */ + async listRuns( + user: AuthUser, + cursor: string | undefined, + limit: number, + ): Promise { + requireAdmin(user); + let decoded: KeysetCursor | null = null; + if (cursor !== undefined) { + decoded = decodeKeysetCursor(cursor); + if (decoded === null) { + throw validationFailed('cursor 不合法'); + } + } + const cursorSort = decoded === null ? null : new Date(decoded.sort); + const cursorId = decoded === null ? '' : decoded.id; + const rows = await this.prisma.user_sync_runs.findMany({ + where: + cursorSort === null + ? undefined + : { + OR: [ + { startedAt: { lt: cursorSort } }, + { startedAt: cursorSort, id: { lt: cursorId } }, + ], + }, + orderBy: [{ startedAt: 'desc' }, { id: 'desc' }], + take: limit + 1, + }); + const hasMore = rows.length > limit; + const page = hasMore ? rows.slice(0, limit) : rows; + const last = page.at(-1); + // CursorPageMeta 不变式：hasMore ⟺ nextCursor≠null（同 albums.list 推导）。 + const nextCursor = + hasMore && last !== undefined + ? encodeKeysetCursor({ + sort: last.startedAt.toISOString(), + id: last.id, + }) + : null; + return { + items: page.map((row) => toRunItem(row)), + nextCursor, + hasMore: nextCursor !== null, + }; + } + + /** 单批次坏行明细（listUserSyncErrors）。 */ + async listErrors( + user: AuthUser, + runId: string, + ): Promise { + requireAdmin(user); + const rows = await this.prisma.user_sync_errors.findMany({ + where: { runId }, + orderBy: [{ createdAt: 'desc' }, { id: 'desc' }], + take: ERRORS_LIST_LIMIT, + }); + return { + items: rows.map((row) => ({ + runId: row.runId, + externalUserId: row.externalUserId ?? '', + code: row.code, + detail: row.detail, + createdAt: row.createdAt.toISOString(), + })), + }; + } +} diff --git a/services/api/src/user-sync/user-sync.controller.ts b/services/api/src/user-sync/user-sync.controller.ts new file mode 100644 index 0000000..72e2283 --- /dev/null +++ b/services/api/src/user-sync/user-sync.controller.ts @@ -0,0 +1,81 @@ +// services/api/src/user-sync/user-sync.controller.ts —— /admin/user-sync 端点 +// （P4a Task 7；契约 getUserSyncStatus / listUserSyncRuns / triggerUserSyncRun / +// listUserSyncErrors，tag [admin]）。方法名与 operationId 一一对应；控制器级 +// SessionGuard：全端点需认证，admin 判定在服务层（403 AUTH_FORBIDDEN）。 +// POST /runs 契约响应 202（受理/复用进行中批次，异步执行——结果经 status/runs +// 查询），显式 @HttpCode。查询参数规整：limit/cursor 经 shared/pagination 同一 +// bounds；runId 必填非空（缺失/空串 → 400 VALIDATION_FAILED）。 +import { + Controller, + Get, + HttpCode, + HttpStatus, + Post, + Query, + Req, + UseGuards, +} from '@nestjs/common'; +import { userOf } from '../albums/albums.controller'; +import { AuthenticatedRequest, SessionGuard } from '../auth/session.guard'; +import { parseCursor, parseLimit } from '../shared/pagination'; +import { validationFailed } from '../albums/support'; +import { + UserSyncAdminService, + type UserSyncErrorListResponse, + type UserSyncRunListResponse, + type UserSyncStatusResponse, +} from './user-sync-admin.service'; +import { UserSyncEngine } from './user-sync.service'; + +@Controller('admin/user-sync') +@UseGuards(SessionGuard) +export class UserSyncController { + constructor( + private readonly adminService: UserSyncAdminService, + private readonly engine: UserSyncEngine, + ) {} + + /** 同步引擎状态 + 最近批次计数。 */ + @Get('status') + getUserSyncStatus( + @Req() req: AuthenticatedRequest, + ): Promise { + return this.adminService.status(userOf(req)); + } + + /** 运行历史（startedAt+ID 倒序游标分页）。 */ + @Get('runs') + listUserSyncRuns( + @Req() req: AuthenticatedRequest, + @Query('cursor') cursor: string | undefined, + @Query('limit') limit: string | undefined, + ): Promise { + return this.adminService.listRuns( + userOf(req), + parseCursor(cursor), + parseLimit(limit), + ); + } + + /** 手动触发同步补拉（与定时共用互斥锁；运行中批次复用其 runId）。 */ + @Post('runs') + @HttpCode(HttpStatus.ACCEPTED) + async triggerUserSyncRun(@Req() req: AuthenticatedRequest): Promise<{ + runId: string; + }> { + const runId = await this.engine.triggerManual(userOf(req)); + return { runId }; + } + + /** 单批次失败明细（runId 必填）。 */ + @Get('errors') + listUserSyncErrors( + @Req() req: AuthenticatedRequest, + @Query('runId') runId: string | undefined, + ): Promise { + if (runId === undefined || runId === '') { + throw validationFailed('runId 必填（批次 ID）'); + } + return this.adminService.listErrors(userOf(req), runId); + } +} diff --git a/services/api/src/user-sync/user-sync.module.ts b/services/api/src/user-sync/user-sync.module.ts new file mode 100644 index 0000000..75fb08b --- /dev/null +++ b/services/api/src/user-sync/user-sync.module.ts @@ -0,0 +1,28 @@ +// services/api/src/user-sync/user-sync.module.ts —— 用户同步域模块（P4a Task 7）。 +// 仅 cn 装配（AppCnModule import；ov 入口不碰——用户身份是 cn 单域事实）。 +// 装配：SyncLock（cn Redis 互斥）、MockUserSource（USER_SOURCE 端口实现， +// G4 后替换 provider 即切真实源）、UserSyncEngine（批次引擎）、 +// UserSyncScheduler（setInterval 定时触发，仅 cn bootstrap 启动）、 +// UserSyncAdminService + UserSyncController（/admin/user-sync 端点）。 +// useExisting 装配端口：引擎只依赖 UserSourcePort 符号，e2e 经 +// app.get(MockUserSource) 注入场景（mock-user-source.ts 文件头）。 +import { Module } from '@nestjs/common'; +import { MockUserSource } from './mock-user-source'; +import { SyncLock } from './sync-lock'; +import { UserSyncAdminService } from './user-sync-admin.service'; +import { UserSyncController } from './user-sync.controller'; +import { UserSyncEngine, UserSyncScheduler } from './user-sync.service'; +import { USER_SOURCE } from './user-source.port'; + +@Module({ + controllers: [UserSyncController], + providers: [ + SyncLock, + MockUserSource, + { provide: USER_SOURCE, useExisting: MockUserSource }, + UserSyncEngine, + UserSyncScheduler, + UserSyncAdminService, + ], +}) +export class UserSyncModule {} diff --git a/services/api/src/user-sync/user-sync.service.ts b/services/api/src/user-sync/user-sync.service.ts new file mode 100644 index 0000000..6e72125 --- /dev/null +++ b/services/api/src/user-sync/user-sync.service.ts @@ -0,0 +1,527 @@ +// services/api/src/user-sync/user-sync.service.ts —— 用户同步引擎（P4a Task 7， +// A29；T30 语义全量，mock 源先行——G4 真实接口到位后仅替换 USER_SOURCE provider）。 +// +// 红线（技术方案 3.1A，e2e 全量守护）： +// 1. 同步绝不写 password_hash / role / album_members / localLocked（display +// 维度）；映射经 external_user_mappings 隔离，users 上不加源身份列。 +// 2. 空响应 / 部分页绝不停用任何人：停用只来自上游行级 sourceStatus=disabled。 +// 3. 上游恢复（disabled→active）不解除 localLocked：本地锁定独立维度， +// 解锁只有管理员 PATCH localLocked=false 一条路。 +// 4. 稳定外部 ID：外部身份锚点是 (sourceSystem, externalUserId) 唯一映射， +// 本地 account 仅是展示/登录维度；loginAccount 与本地已有账号同名时 +// 走「关联」而非冲突（建映射 + 应用源属性，不动该账号既有凭据/角色）。 +// +// 批次流程（run）： +// Redis SETNX 互斥（锁值=runId，TTL 10min，Lua CAS 释放——见 sync-lock.ts） +// → 建 running 批次行 + 单行状态推进（lastRunAt/runCount++） +// → 游标：每 RECONCILE_EVERY_N_RUNS 批强制 cursor=null（周期全量对账， +// 修补增量水位漏掉的行内版本更新）；其余批次从已提交水位增量拉取 +// → 逐页：逐行校验/应用（坏行 → user_sync_errors + counts.failed++ + +// 批次 partial，其余行照常）；**每页提交一次**水位（run.cursor + +// state.cursor 同事务）——分页中断时水位停在最后成功页，已应用行不回滚 +// （幂等重入：版本比较 + account 关联使下批安全重放） +// → 终态：success（无坏行）/ partial（有坏行）/ failed（listPage 抛错等 +// 批次级故障：error 摘要落账，会话绝不被批次故障牵连撤销）。 +// +// 待激活闭环（§10）：新建用户 credentialStatus=pending + 哨兵哈希 +// `!:pending:`（非法 argon2 编码，任何口令都无法通过校验——auth.service +// 对 verify 抛错按密码错误处理并跑等代价 dummy verify）。初始凭据由管理员 +// activate + reset-password 出具（Task 6 端点）。 +// +// department 字段：users 表无此列，忽略（端口保留对齐真实源形状，登记偏差）。 +import { randomBytes } from 'node:crypto'; +import { + Inject, + Injectable, + Logger, + HttpStatus, + type OnApplicationBootstrap, + type OnModuleDestroy, +} from '@nestjs/common'; +import { USER_SYNC_STATE_SINGLETON_ID } from '@photo/db-cn'; +import { ErrorCodes } from '@photo/contracts'; +import { ApiException } from '../api-exception'; +import type { AuthUser } from '../auth/session.guard'; +import { PrismaService } from '../prisma/prisma.service'; +import { SyncLock } from './sync-lock'; +import { + USER_SOURCE, + type ExternalUserRecord, + type UserSourcePort, +} from './user-source.port'; + +/** 同步源系统标识（external_user_mappings.sourceSystem；G4 后换真实源标识）。 */ +export const USER_SYNC_SOURCE_SYSTEM = 'mock'; + +/** 哨兵哈希前缀：同步创建的待激活账号（非法 argon2 编码，不可被猜中登录）。 */ +export const PENDING_PASSWORD_HASH_PREFIX = '!:pending:'; + +/** 定时同步间隔缺省 5 分钟（env SYNC_USERS_INTERVAL_MS 可调）。 */ +const DEFAULT_SYNC_INTERVAL_MS = 300_000; + +/** 周期全量对账节奏缺省：每 10 批强制 cursor=null（env RECONCILE_EVERY_N_RUNS）。 */ +const DEFAULT_RECONCILE_EVERY_N_RUNS = 10; + +/** 单批页数护栏（源实现 bug 导致的无限分页保护）。 */ +const MAX_PAGES_PER_RUN = 10_000; + +export type UserSyncTrigger = 'cron' | 'manual'; +export type UserSyncRunStateValue = 'running' | 'success' | 'partial' | 'failed'; + +export interface UserSyncCounts { + added: number; + updated: number; + disabled: number; + failed: number; +} + +const ZERO_COUNTS: UserSyncCounts = { added: 0, updated: 0, disabled: 0, failed: 0 }; + +/** 单行应用结果：bucket=null 即幂等跳过（版本不高于映射行）。 */ +interface ItemOutcome { + bucket: 'added' | 'updated' | 'disabled' | 'failed' | null; + /** 可信行版本（坏行为 null——其版本不可采信，不进水位）。 */ + version: number | null; + code?: string; + detail?: string; +} + +function pendingPasswordHash(): string { + return `${PENDING_PASSWORD_HASH_PREFIX}${randomBytes(16).toString('hex')}`; +} + +/** 引擎侧批次 ID：预生成以兼作锁值（24 hex，撞库概率可忽略）。 */ +function generateRunId(): string { + return randomBytes(12).toString('hex'); +} + +/** 水位串 → 契约 int64（mock 恒数值串；非数值游标——G4 真实源——投影为 null）。 */ +export function watermarkToInt(cursor: string | null): number | null { + if (cursor === null || cursor === '') { + return null; + } + const value = Number(cursor); + return Number.isSafeInteger(value) && value >= 0 ? value : null; +} + +/** 坏行校验：外部行字段可能不合规（mock 场景透传），引擎是唯一校验点。 */ +function validateItem( + item: ExternalUserRecord, +): { ok: true; version: number } | { ok: false; detail: string } { + if (typeof item.externalUserId !== 'string' || item.externalUserId === '' || item.externalUserId.length > 64) { + return { ok: false, detail: 'externalUserId 缺失或超长（须 1-64 字符）' }; + } + if (typeof item.displayName !== 'string' || item.displayName === '' || item.displayName.length > 128) { + return { ok: false, detail: `用户 ${item.externalUserId}：displayName 缺失或超长（须 1-128 字符）` }; + } + if (item.sourceStatus !== 'active' && item.sourceStatus !== 'disabled') { + return { ok: false, detail: `用户 ${item.externalUserId}：sourceStatus 非法（${String(item.sourceStatus)}）` }; + } + if (!Number.isSafeInteger(item.sourceVersion) || item.sourceVersion < 0) { + return { ok: false, detail: `用户 ${item.externalUserId}：sourceVersion 非法（${String(item.sourceVersion)}）` }; + } + if (item.loginAccount !== undefined && (item.loginAccount === '' || item.loginAccount.length > 64)) { + return { ok: false, detail: `用户 ${item.externalUserId}：loginAccount 非法（须 1-64 字符）` }; + } + return { ok: true, version: item.sourceVersion }; +} + +@Injectable() +export class UserSyncEngine { + private readonly logger = new Logger(UserSyncEngine.name); + private readonly reconcileEveryNRuns: number; + + constructor( + private readonly prisma: PrismaService, + @Inject(USER_SOURCE) private readonly source: UserSourcePort, + private readonly lock: SyncLock, + ) { + const raw = Number.parseInt(process.env.RECONCILE_EVERY_N_RUNS ?? '', 10); + this.reconcileEveryNRuns = + Number.isFinite(raw) && raw >= 1 ? raw : DEFAULT_RECONCILE_EVERY_N_RUNS; + } + + /** + * 手动补拉（POST /admin/user-sync/runs → 202 {runId}）：仅 admin（403 + * AUTH_FORBIDDEN——判定须在触引擎**之前**，非 admin 不产生任何批次副作用）。 + * 与定时共用互斥锁；撞锁且持有中批次仍在 running → 复用该 runId； + * 自启批次则后台异步执行（受理即返回，结果经 status/runs 查询）。 + */ + async triggerManual(user: AuthUser): Promise { + if (user.role !== 'admin') { + throw new ApiException( + ErrorCodes.AUTH_FORBIDDEN, + '仅管理员可手动触发用户同步', + HttpStatus.FORBIDDEN, + ); + } + for (let attempt = 0; attempt < 2; attempt++) { + const started = await this.tryStartRun('manual'); + if (started !== null) { + if (started.own) { + void this.execute(started.runId).catch((err: unknown) => { + this.logger.error(`手动同步批次 ${started.runId} 执行异常: ${errText(err)}`); + }); + } + return started.runId; + } + // GET 与 SET 之间被并发抢锁：重试一次，仍失败则读回赢家复用。 + } + const winner = await this.lock.currentRunId(); + if (winner !== null) { + return winner; + } + throw new Error('user-sync: 无法获取同步互斥锁（连续两次撞锁后锁已释放，异常状态）'); + } + + /** 定时批次入口（UserSyncScheduler 间隔调用；e2e 亦直接调用以测 cron 语义）。 */ + async tick(): Promise { + try { + const started = await this.tryStartRun('cron'); + if (started === null || !started.own) { + return; // 撞锁（复用语义对定时无意义）→ 本轮跳过 + } + await this.execute(started.runId); + } catch (err) { + this.logger.error(`定时同步批次失败: ${errText(err)}`); + } + } + + /** + * 启动一批：撞锁时若持有中批次仍 running → 复用（own=false）； + * 陈旧锁（批次已终结未释放 / 建行前崩溃）→ CAS 清理后重新获取。 + * 返回 null = 竞争失败（本轮放弃）。 + */ + private async tryStartRun( + trigger: UserSyncTrigger, + ): Promise<{ runId: string; own: boolean } | null> { + const held = await this.lock.currentRunId(); + if (held !== null) { + const running = await this.prisma.user_sync_runs.findUnique({ + where: { id: held }, + select: { state: true }, + }); + if (running !== null && running.state === 'running') { + return { runId: held, own: false }; + } + await this.lock.releaseIfValue(held); // 陈旧锁清理 + } + const runId = generateRunId(); + if (!(await this.lock.acquire(runId))) { + return null; + } + try { + await this.prisma.user_sync_runs.create({ + data: { + id: runId, + trigger, + state: 'running', + counts: { ...ZERO_COUNTS }, + startedAt: new Date(), + }, + }); + await this.prisma.user_sync_state.upsert({ + where: { id: USER_SYNC_STATE_SINGLETON_ID }, + create: { + id: USER_SYNC_STATE_SINGLETON_ID, + lastRunAt: new Date(), + runCount: 1, + }, + update: { lastRunAt: new Date(), runCount: { increment: 1 } }, + }); + return { runId, own: true }; + } catch (err) { + await this.lock.releaseIfValue(runId).catch(() => undefined); + throw err; + } + } + + /** 批次主体（锁已持有；execute 负责终态落账与锁释放）。 */ + private async execute(runId: string): Promise { + let finalState: UserSyncRunStateValue = 'success'; + let errorMessage: string | null = null; + try { + const [stateRow, runRow] = await Promise.all([ + this.prisma.user_sync_state.findUnique({ + where: { id: USER_SYNC_STATE_SINGLETON_ID }, + }), + this.prisma.user_sync_runs.findUnique({ where: { id: runId } }), + ]); + if (runRow === null) { + throw new Error(`批次 ${runId} 不存在（启动事务异常）`); + } + const startedAt = runRow.startedAt; + const runNumber = stateRow?.runCount ?? 1; + // 周期全量对账：每 N 批忽略增量水位（cursor=null → 源返回全量快照）， + // 修补增量流漏掉的行内版本更新（水位是全局最大版本，低版本行的新变更 + // 对增量不可见——对账页以映射行版本逐行比较，幂等安全）。 + const fullSnapshot = runNumber % this.reconcileEveryNRuns === 0; + const counts = { ...ZERO_COUNTS }; + let watermark: string | null = stateRow?.cursor ?? null; + let pageCursor: string | null = fullSnapshot ? null : watermark; + + let pages = 0; + for (;;) { + const page = await this.source.listPage(pageCursor ?? undefined); + let pageMaxVersion = watermarkToInt(watermark) ?? 0; + for (const item of page.items) { + const outcome = await this.applyItem(item); + if (outcome.version !== null && outcome.version > pageMaxVersion) { + pageMaxVersion = outcome.version; + } + if (outcome.bucket === 'failed') { + counts.failed += 1; + if (finalState === 'success') { + finalState = 'partial'; + } + await this.prisma.user_sync_errors.create({ + data: { + runId, + externalUserId: + typeof item.externalUserId === 'string' && item.externalUserId !== '' + ? item.externalUserId.slice(0, 64) + : null, + code: outcome.code ?? 'apply_failed', + detail: (outcome.detail ?? '未知失败').slice(0, 1024), + }, + }); + } else if (outcome.bucket !== null) { + counts[outcome.bucket] += 1; + } + } + // 每页提交：水位与批次计数同事务推进（分页中断时停在最后成功页）。 + if (pageMaxVersion > (watermarkToInt(watermark) ?? 0)) { + watermark = String(pageMaxVersion); + } + await this.prisma.$transaction([ + this.prisma.user_sync_runs.update({ + where: { id: runId }, + data: { counts: { ...counts }, cursor: watermark }, + }), + this.prisma.user_sync_state.updateMany({ + where: { id: USER_SYNC_STATE_SINGLETON_ID }, + data: { cursor: watermark }, + }), + ]); + if (page.nextCursor === null) { + break; + } + pageCursor = page.nextCursor; + pages += 1; + if (pages > MAX_PAGES_PER_RUN) { + throw new Error(`源分页超出护栏（>${MAX_PAGES_PER_RUN} 页），疑似游标死循环`); + } + } + await this.finalize(runId, finalState, startedAt); + } catch (err) { + // 批次级故障（分页中断/护栏/DB）：failed 终态；水位停在最后提交页， + // 已应用行不回滚（幂等重入），任何会话不被批次故障牵连。 + errorMessage = errText(err).slice(0, 1024); + this.logger.warn(`同步批次 ${runId} 失败: ${errorMessage}`); + try { + await this.finalize(runId, 'failed', undefined, errorMessage); + } catch (finalizeErr) { + this.logger.error(`同步批次 ${runId} 终态落账失败: ${errText(finalizeErr)}`); + } + } finally { + await this.lock.releaseIfValue(runId).catch((err: unknown) => { + this.logger.warn(`同步批次 ${runId} 锁释放失败（TTL 自愈）: ${errText(err)}`); + }); + } + } + + /** 终态落账：批次行 + （success 时）单行状态 lastSuccessAt。 */ + private async finalize( + runId: string, + state: UserSyncRunStateValue, + startedAt: Date | undefined, + error: string | null = null, + ): Promise { + await this.prisma.$transaction([ + this.prisma.user_sync_runs.update({ + where: { id: runId }, + data: { state, finishedAt: new Date(), error }, + }), + ...(state === 'success' && startedAt !== undefined + ? [ + this.prisma.user_sync_state.updateMany({ + where: { id: USER_SYNC_STATE_SINGLETON_ID }, + data: { lastSuccessAt: startedAt }, + }), + ] + : []), + ]); + } + + /** 单行应用（映射存在 → 幂等更新；不存在 → account 关联或建待激活用户）。 */ + private async applyItem(item: ExternalUserRecord): Promise { + const validated = validateItem(item); + if (!validated.ok) { + return { bucket: 'failed', version: null, code: 'invalid_record', detail: validated.detail }; + } + try { + const mappingWhere = { + sourceSystem_externalUserId: { + sourceSystem: USER_SYNC_SOURCE_SYSTEM, + externalUserId: item.externalUserId, + }, + }; + const mapping = await this.prisma.external_user_mappings.findUnique({ + where: mappingWhere, + }); + if (mapping !== null) { + const user = await this.prisma.users.findUnique({ + where: { id: mapping.userId }, + }); + if (user === null) { + // 本地用户已被应用层决策删除而映射残留：按新建路径重建并改指映射 + //（稳定外部 ID 语义：同一外部身份恒指向同一个本地用户行）。 + const created = await this.prisma.users.create({ + data: { + account: item.loginAccount ?? item.externalUserId, + passwordHash: pendingPasswordHash(), + displayName: item.displayName, + role: 'staff', + sourceStatus: item.sourceStatus, + credentialStatus: 'pending', + }, + }); + await this.prisma.external_user_mappings.update({ + where: { id: mapping.id }, + data: { userId: created.id, sourceVersion: validated.version }, + }); + return { bucket: 'added', version: validated.version }; + } + if (validated.version <= mapping.sourceVersion) { + return { bucket: null, version: validated.version }; // 幂等跳过 + } + const becomingDisabled = + item.sourceStatus === 'disabled' && user.sourceStatus !== 'disabled'; + const revokedAt = new Date(); + await this.prisma.$transaction([ + this.prisma.users.update({ + where: { id: user.id }, + data: { + displayName: item.displayName, + sourceStatus: item.sourceStatus, // 红线 3：绝不动 localLocked + }, + }), + this.prisma.external_user_mappings.update({ + where: { id: mapping.id }, + data: { sourceVersion: validated.version }, + }), + ...(becomingDisabled + ? [ + this.prisma.sessions.updateMany({ + where: { userId: user.id, revokedAt: null }, + data: { revokedAt }, + }), + ] + : []), + ]); + return { + bucket: becomingDisabled ? 'disabled' : 'updated', + version: validated.version, + }; + } + + // 无映射：优先 account 关联（本地已直建同账号用户——T30 并存语义）。 + const account = item.loginAccount ?? item.externalUserId; + const existing = await this.prisma.users.findUnique({ where: { account } }); + if (existing !== null) { + const becomingDisabled = + item.sourceStatus === 'disabled' && existing.sourceStatus !== 'disabled'; + const revokedAt = new Date(); + await this.prisma.$transaction([ + this.prisma.users.update({ + where: { id: existing.id }, + data: { + displayName: item.displayName, + sourceStatus: item.sourceStatus, + }, + }), + this.prisma.external_user_mappings.create({ + data: { + userId: existing.id, + sourceSystem: USER_SYNC_SOURCE_SYSTEM, + externalUserId: item.externalUserId, + sourceVersion: validated.version, + }, + }), + ...(becomingDisabled + ? [ + this.prisma.sessions.updateMany({ + where: { userId: existing.id, revokedAt: null }, + data: { revokedAt }, + }), + ] + : []), + ]); + return { + bucket: becomingDisabled ? 'disabled' : 'updated', + version: validated.version, + }; + } + + // 全新外部身份 → 建待激活用户（§10 闭环：activate + reset-password 出凭据）。 + const created = await this.prisma.users.create({ + data: { + account, + passwordHash: pendingPasswordHash(), + displayName: item.displayName, + role: 'staff', + sourceStatus: item.sourceStatus, + credentialStatus: 'pending', + }, + }); + await this.prisma.external_user_mappings.create({ + data: { + userId: created.id, + sourceSystem: USER_SYNC_SOURCE_SYSTEM, + externalUserId: item.externalUserId, + sourceVersion: validated.version, + }, + }); + return { bucket: 'added', version: validated.version }; + } catch (err) { + return { + bucket: 'failed', + version: null, + code: 'apply_failed', + detail: errText(err), + }; + } + } +} + +/** 定时调度（仅 cn 装配——UserSyncModule 只进 AppCnModule）：间隔触发引擎 + * tick；锁被持有时 tick 自行跳过。interval unref：不阻塞进程正常退出。 */ +@Injectable() +export class UserSyncScheduler implements OnApplicationBootstrap, OnModuleDestroy { + private readonly logger = new Logger(UserSyncScheduler.name); + private timer: NodeJS.Timeout | null = null; + + constructor(private readonly engine: UserSyncEngine) {} + + onApplicationBootstrap(): void { + const raw = Number.parseInt(process.env.SYNC_USERS_INTERVAL_MS ?? '', 10); + const intervalMs = + Number.isFinite(raw) && raw > 0 ? raw : DEFAULT_SYNC_INTERVAL_MS; + this.timer = setInterval(() => { + void this.engine.tick(); + }, intervalMs); + this.timer.unref(); + this.logger.log(`用户同步定时器就绪：每 ${intervalMs}ms（撞锁自动跳过）`); + } + + onModuleDestroy(): void { + if (this.timer !== null) { + clearInterval(this.timer); + this.timer = null; + } + } +} + +function errText(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} diff --git a/services/api/test/reset.ts b/services/api/test/reset.ts index 8f23cfe..32c4307 100644 --- a/services/api/test/reset.ts +++ b/services/api/test/reset.ts @@ -17,6 +17,7 @@ import { import { PrismaClient, USAGE_THRESHOLDS_SINGLETON_ID, + USER_SYNC_STATE_SINGLETON_ID, } from '@photo/db-cn'; export const RESET_TABLES: readonly string[] = [ @@ -27,6 +28,14 @@ export const RESET_TABLES: readonly string[] = [ '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', @@ -120,6 +129,11 @@ export async function resetDb(): Promise { // 视作双维停用，跨 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(); diff --git a/services/api/test/user-sync.e2e-spec.ts b/services/api/test/user-sync.e2e-spec.ts new file mode 100644 index 0000000..0e1824a --- /dev/null +++ b/services/api/test/user-sync.e2e-spec.ts @@ -0,0 +1,726 @@ +// services/api/test/user-sync.e2e-spec.ts —— P4a Task 7 用户同步 e2e（先写后 +// 实现）。真实 app + MySQL + cn Redis（互斥锁）。MockUserSource 场景经 +// app.get(MockUserSource) 程序化注入（mock-user-source.ts 文件头约定）。 +// 覆盖（T30 语义全量，红线逐条断言）： +// ① 首轮全量：3 用户 = 1 account 关联（本地直建账号建映射，不动其 +// 凭据/角色，会话保留）+ 2 新增待激活（credentialStatus=pending + +// 哨兵哈希 !:pending:）；counts/水位/映射行； +// ② 改名幂等：版本更高 → displayName 更新；同版本重跑 → 全跳过（counts 0）； +// ③ 上游停用与恢复：disabled → users.sourceStatus=disabled + 撤全部会话 +// （令牌 401 + 正确密码登录 401）且 **localLocked 不被写入**；恢复 +// active → 原密码可再登录； +// ④ 红线「上游恢复不解除 localLocked」：手动停用（localLocked=true）+ +// 上游停/恢复 → sourceStatus=active 而 localLocked 仍 true → 登录 401； +// ⑤ 增量新增第 4 用户：mock 收到已提交水位游标（receivedCursors）； +// ⑥ 空响应：items=[]+nextCursor=null → 正常结束、counts 全 0、无人被动、 +// 水位不进； +// ⑦ 分页中断：page2 抛错 → 批次 failed、水位停在 page1、未拉到的行不出现、 +// 无用户被动停用；修复后增量补齐； +// ⑧ 部分无效记录：1 坏行 → partial + errors 明细（invalid_record）+ 其余 +// 照常 + lastSuccessAt 不推进（partial 不是完全成功）； +// ⑨ 手动补拉与定时互斥：门闸制造运行中批次 → 第二次 POST 复用 runId、 +// cron tick 跳过（无新批次）；放行后完成； +// ⑩ 增量 + 周期全量对账（RECONCILE_EVERY_N_RUNS=10）：低版本行的新变更 +// 对增量不可见 → 第 10 批 cursor=null 全量对账应用；缺席用户绝不停用； +// ⑪ §10 闭环（Task 6 联动）：同步创建的 pending 用户登录 401（哨兵哈希 +// 不 500）→ activate → 仍 401（哈希不可猜）→ reset-password → 可登录； +// ⑫ status/runs/errors API 形状 + runs 游标分页走全量 + 参数 400 + 非 +// admin 403 / 匿名 401。 +import { randomBytes } from 'node:crypto'; +import { type INestApplication } from '@nestjs/common'; +import { Test } from '@nestjs/testing'; +import request from 'supertest'; +import IORedis from 'ioredis'; +import { PrismaClient, USER_SYNC_STATE_SINGLETON_ID } from '@photo/db-cn'; +import { ErrorCodes } from '@photo/contracts'; +import { AppCnModule } from '../src/app-cn.module'; +import { hashPassword } from '../src/auth/password'; +import { hashSessionToken } from '../src/auth/session.guard'; +import { API_PREFIX } from '../src/shared/constants'; +import { + MockUserSource, + type MockExternalRecord, + type MockScenario, +} from '../src/user-sync/mock-user-source'; +import { USER_SYNC_LOCK_KEY } from '../src/user-sync/sync-lock'; +import { UserSyncEngine } from '../src/user-sync/user-sync.service'; +import { resetDb } from './reset'; + +jest.setTimeout(120_000); + +/** 批次行（断言用字段子集）。 */ +interface RunRow { + id: string; + trigger: string; + state: string; + cursor: string | null; + counts: { added: number; updated: number; disabled: number; failed: number }; + error: string | null; + startedAt: Date; + finishedAt: Date | null; +} + +describe('用户定期同步 /admin/user-sync (e2e)', () => { + let app: INestApplication; + let prisma: PrismaClient; + let mock: MockUserSource; + let engine: UserSyncEngine; + + let adminToken: string; + let staffToken: string; + let u1Id: string; + let u1Token: string; + let u1OriginalHash: string; + + beforeAll(async () => { + await resetDb(); + // Redis 互斥锁键不在 resetDb 清理范围（见 reset.ts 注释）：spec 自清， + // 防上一轮残留锁让本 spec 首批直接「复用幽灵批次」。 + const redis = new IORedis(process.env.REDIS_URL_CN ?? 'redis://localhost:6379'); + try { + await redis.del(USER_SYNC_LOCK_KEY); + } finally { + await redis.quit().catch(() => undefined); + } + + const moduleRef = await Test.createTestingModule({ + imports: [AppCnModule], + }).compile(); + app = moduleRef.createNestApplication(); + app.setGlobalPrefix(API_PREFIX); + await app.init(); + + prisma = new PrismaClient(); + mock = app.get(MockUserSource); + engine = app.get(UserSyncEngine); + + const mkToken = async (userId: string): Promise => { + const token = randomBytes(32).toString('hex'); + await prisma.sessions.create({ + data: { + userId, + tokenHash: hashSessionToken(token), + expiresAt: new Date(Date.now() + 3_600_000), + }, + }); + return token; + }; + + // ① 的关联目标：本地直建账号（真实密码 + 既有会话——同步只许建映射和 + // 应用 displayName/sourceStatus，凭据/角色/会话必须原样）。 + u1OriginalHash = await hashPassword('U1-Original-77'); + u1Id = ( + await prisma.users.create({ + data: { + account: 'sync-u1', + passwordHash: u1OriginalHash, + displayName: '本地直建', + role: 'staff', + }, + select: { id: true }, + }) + ).id; + u1Token = await mkToken(u1Id); + + const adminId = ( + await prisma.users.create({ + data: { + account: 'sync-admin', + passwordHash: await hashPassword('Admin-Sync-1'), + displayName: '同步管理员', + role: 'admin', + }, + select: { id: true }, + }) + ).id; + adminToken = await mkToken(adminId); + const staffId = ( + await prisma.users.create({ + data: { + account: 'sync-staff', + passwordHash: await hashPassword('Staff-Sync-1'), + displayName: '同步员工', + }, + select: { id: true }, + }) + ).id; + staffToken = await mkToken(staffId); + }); + + afterAll(async () => { + await app.close(); + await prisma.$disconnect(); + }); + + // —— 工具 —— + + const ext = ( + externalUserId: string, + loginAccount: string, + displayName: string, + sourceVersion: number, + sourceStatus: string = 'active', + ): MockExternalRecord => ({ + externalUserId, + loginAccount, + displayName, + sourceStatus, + sourceVersion, + updatedAt: new Date(), + }); + + function scenario(users: MockExternalRecord[], extra: Partial = {}): void { + mock.setScenario({ users, ...extra }); + } + + async function waitRunDone(runId: string): Promise { + const deadline = Date.now() + 10_000; + for (;;) { + const row = await prisma.user_sync_runs.findUnique({ where: { id: runId } }); + if (row !== null && row.state !== 'running') { + return row as unknown as RunRow; + } + if (Date.now() > deadline) { + throw new Error(`批次 ${runId} 未在时限内完成`); + } + await new Promise((resolve) => setTimeout(resolve, 25)); + } + } + + /** 手动补拉并等待完成（POST 202 → 轮询批次终态）。 */ + async function triggerRun(): Promise<{ runId: string; run: RunRow }> { + const res = await request(app.getHttpServer()) + .post(`/${API_PREFIX}/admin/user-sync/runs`) + .set('Authorization', `Bearer ${adminToken}`) + .send(undefined); + expect(res.status).toBe(202); + expect(typeof res.body.runId).toBe('string'); + const run = await waitRunDone(res.body.runId); + return { runId: res.body.runId, run }; + } + + function getStatus(token: string | null = adminToken) { + const req = request(app.getHttpServer()) + .get(`/${API_PREFIX}/admin/user-sync/status`); + return token === null ? req : req.set('Authorization', `Bearer ${token}`); + } + + function listRuns( + query: Record = {}, + token: string | null = adminToken, + ) { + const req = request(app.getHttpServer()) + .get(`/${API_PREFIX}/admin/user-sync/runs`); + return (token === null ? req : req.set('Authorization', `Bearer ${token}`)) + .query(query); + } + + function listErrors( + query: Record, + token: string | null = adminToken, + ) { + const req = request(app.getHttpServer()) + .get(`/${API_PREFIX}/admin/user-sync/errors`); + return (token === null ? req : req.set('Authorization', `Bearer ${token}`)) + .query(query); + } + + function postRun(token: string) { + return request(app.getHttpServer()) + .post(`/${API_PREFIX}/admin/user-sync/runs`) + .set('Authorization', `Bearer ${token}`) + .send(undefined); + } + + function login(account: string, password: string) { + return request(app.getHttpServer()) + .post(`/${API_PREFIX}/auth/password`) + .send({ account, password }); + } + + /** 任意受保护 staff 端点探针（前缀查询）：验证令牌是否仍被守卫放行。 */ + function probeToken(token: string, accountPrefix = 'sync-u') { + return request(app.getHttpServer()) + .get(`/${API_PREFIX}/admin/users`) + .set('Authorization', `Bearer ${token}`) + .query({ accountPrefix }); + } + + const userByAccount = (account: string) => + prisma.users.findUnique({ where: { account } }); + + // —— 场景基线（版本账本见各用例注释）—— + const U1 = (version: number, status = 'active') => + ext('ext-1', 'sync-u1', '李一味', version, status); + const U2 = (version: number, displayName = '王二') => + ext('ext-2', 'sync-u2', displayName, version); + const U3 = (version: number) => ext('ext-3', 'sync-u3', '赵三', version); + + it('① 首轮全量：1 account 关联 + 2 新增待激活；counts/水位/映射/红线（不动凭据角色会话）', async () => { + scenario([U1(1), U2(1), U3(1)]); // pageSize 2 → 两页（1,2 | 3），全 v1 + const { run } = await triggerRun(); + + expect(run.state).toBe('success'); + expect(run.trigger).toBe('manual'); + expect(run.counts).toEqual({ added: 2, updated: 1, disabled: 0, failed: 0 }); + expect(run.cursor).toBe('1'); // 水位 = 已提交最大版本 + expect(run.finishedAt).not.toBeNull(); + + // 新增：待激活 + 哨兵哈希 + staff 角色 + 源状态。 + for (const account of ['sync-u2', 'sync-u3']) { + const user = await userByAccount(account); + expect(user).not.toBeNull(); + expect(user?.credentialStatus).toBe('pending'); + expect(user?.role).toBe('staff'); + expect(user?.sourceStatus).toBe('active'); + expect(user?.passwordHash.startsWith('!:pending:')).toBe(true); + } + + // 关联：本地直建账号建映射 + 应用源 displayName/sourceStatus； + // 凭据/角色/会话原样（红线 1：绝不写 password_hash/role）。 + const u1 = await userByAccount('sync-u1'); + expect(u1?.displayName).toBe('李一味'); + expect(u1?.sourceStatus).toBe('active'); + expect(u1?.passwordHash).toBe(u1OriginalHash); + expect(u1?.role).toBe('staff'); + expect(u1?.localLocked).toBe(false); + const liveU1Sessions = await prisma.sessions.count({ + where: { userId: u1Id, revokedAt: null }, + }); + expect(liveU1Sessions).toBe(1); + const probed = await probeToken(u1Token); + expect(probed.status).toBe(200); + + // 映射：稳定外部 ID 锚点 (sourceSystem, externalUserId) 唯一。 + const mappings = await prisma.external_user_mappings.findMany(); + expect(mappings).toHaveLength(3); + expect( + new Set(mappings.map((m) => `${m.sourceSystem}:${m.externalUserId}`)), + ).toEqual(new Set(['mock:ext-1', 'mock:ext-2', 'mock:ext-3'])); + for (const m of mappings) { + expect(m.sourceVersion).toBe(1); + } + + // status 端点：首轮即 success + 水位 + 计数。 + const status = await getStatus(); + expect(status.status).toBe(200); + expect(status.body.state).toBe('success'); + expect(status.body.sourceVersion).toBe(1); + expect(status.body.counts).toEqual({ + added: 2, + updated: 1, + disabled: 0, + failed: 0, + }); + expect(status.body.lastRunAt).not.toBeNull(); + expect(status.body.lastSuccessAt).not.toBeNull(); + }); + + it('② 改名幂等：版本更高 → displayName 更新；同版本重跑 → 全跳过', async () => { + scenario([U1(1), U2(2, '王二改'), U3(1)]); + const first = await triggerRun(); + expect(first.run.state).toBe('success'); + expect(first.run.counts).toEqual({ added: 0, updated: 1, disabled: 0, failed: 0 }); + expect((await userByAccount('sync-u2'))?.displayName).toBe('王二改'); + const mapping = await prisma.external_user_mappings.findUnique({ + where: { + sourceSystem_externalUserId: { sourceSystem: 'mock', externalUserId: 'ext-2' }, + }, + }); + expect(mapping?.sourceVersion).toBe(2); + + // 同版本重跑：幂等跳过，counts 全 0，displayName 不动。 + const again = await triggerRun(); + expect(again.run.state).toBe('success'); + expect(again.run.counts).toEqual({ added: 0, updated: 0, disabled: 0, failed: 0 }); + expect((await userByAccount('sync-u2'))?.displayName).toBe('王二改'); + expect(again.run.cursor).toBe('2'); + }); + + it('③ 上游停用与恢复：disabled → sourceStatus + 撤全部会话 + 登录 401；恢复 → 原密码可登录', async () => { + scenario([U1(3, 'disabled'), U2(2, '王二改'), U3(1)]); + const disabled = await triggerRun(); + expect(disabled.run.state).toBe('success'); + expect(disabled.run.counts).toEqual({ added: 0, updated: 0, disabled: 1, failed: 0 }); + + const u1 = await userByAccount('sync-u1'); + expect(u1?.sourceStatus).toBe('disabled'); + expect(u1?.localLocked).toBe(false); // 红线：同步不写 localLocked + const liveSessions = await prisma.sessions.count({ + where: { userId: u1Id, revokedAt: null }, + }); + expect(liveSessions).toBe(0); + const probed = await probeToken(u1Token); + expect(probed.status).toBe(401); + expect(probed.body.code).toBe(ErrorCodes.AUTH_TOKEN_EXPIRED); + const lockedLogin = await login('sync-u1', 'U1-Original-77'); + expect(lockedLogin.status).toBe(401); + expect(lockedLogin.body.code).toBe(ErrorCodes.AUTH_INVALID_CREDENTIALS); + + // 上游恢复（普通账号）：sourceStatus=active，原密码可再登录。 + scenario([U1(4), U2(2, '王二改'), U3(1)]); + const recovered = await triggerRun(); + expect(recovered.run.counts).toEqual({ added: 0, updated: 1, disabled: 0, failed: 0 }); + expect((await userByAccount('sync-u1'))?.sourceStatus).toBe('active'); + const relogin = await login('sync-u1', 'U1-Original-77'); + expect(relogin.status).toBe(200); + }); + + it('④ 红线：上游恢复不解除 localLocked（本地锁定独立维度）', async () => { + // 管理员手动停用（Task 6 deactivate：localLocked=true + 撤会话）。 + const deact = await request(app.getHttpServer()) + .post(`/${API_PREFIX}/admin/users/${u1Id}/deactivate`) + .set('Authorization', `Bearer ${adminToken}`) + .send({}); + expect(deact.status).toBe(200); + + // 上游同时停用：sourceStatus=disabled，localLocked 不被触碰（已为 true）。 + scenario([U1(5, 'disabled'), U2(2, '王二改'), U3(1)]); + await triggerRun(); + const mid = await userByAccount('sync-u1'); + expect(mid?.sourceStatus).toBe('disabled'); + expect(mid?.localLocked).toBe(true); + + // 上游恢复：sourceStatus=active 但 localLocked **仍为 true**（3.1A 红线）。 + scenario([U1(6), U2(2, '王二改'), U3(1)]); + const recovered = await triggerRun(); + expect(recovered.run.counts).toEqual({ added: 0, updated: 1, disabled: 0, failed: 0 }); + const after = await userByAccount('sync-u1'); + expect(after?.sourceStatus).toBe('active'); + expect(after?.localLocked).toBe(true); + const stillLocked = await login('sync-u1', 'U1-Original-77'); + expect(stillLocked.status).toBe(401); + expect(stillLocked.body.code).toBe(ErrorCodes.AUTH_INVALID_CREDENTIALS); + + // 解锁唯一路径 = 管理员 PATCH localLocked=false（恢复 u1 供后续用例）。 + const unlock = await request(app.getHttpServer()) + .patch(`/${API_PREFIX}/admin/users/${u1Id}`) + .set('Authorization', `Bearer ${adminToken}`) + .send({ localLocked: false }); + expect(unlock.status).toBe(200); + expect((await userByAccount('sync-u1'))?.localLocked).toBe(false); + }); + + it('⑤ 增量新增第 4 用户：mock 收到已提交水位游标，仅新行入账', async () => { + scenario([ext('ext-4', 'sync-u4', '钱四', 7)]); // 水位 6 → v7 唯一可见 + const { run } = await triggerRun(); + expect(run.state).toBe('success'); + expect(run.counts).toEqual({ added: 1, updated: 0, disabled: 0, failed: 0 }); + expect(run.cursor).toBe('7'); + + // 增量形态证据：listPage 收到裸水位串 "6"（非 null 全量、非续翻页）。 + expect(mock.receivedCursors).toEqual(['6']); + + const u4 = await userByAccount('sync-u4'); + expect(u4).not.toBeNull(); + expect(u4?.credentialStatus).toBe('pending'); + expect(u4?.displayName).toBe('钱四'); + }); + + it('⑥ 空响应：items=[] + nextCursor=null → 正常结束不误伤，水位不进', async () => { + const usersBefore = await prisma.users.count(); + scenario([]); + const { run } = await triggerRun(); + expect(run.state).toBe('success'); + expect(run.counts).toEqual({ added: 0, updated: 0, disabled: 0, failed: 0 }); + expect(run.cursor).toBe('7'); // 水位不动 + expect(run.error).toBeNull(); + + expect(await prisma.users.count()).toBe(usersBefore); + const disabledCount = await prisma.users.count({ + where: { sourceStatus: 'disabled' }, + }); + expect(disabledCount).toBe(0); // 空响应绝不停用任何人（红线 2） + + const status = await getStatus(); + expect(status.body.state).toBe('success'); + expect(status.body.sourceVersion).toBe(7); + const errors = await listErrors({ runId: run.id }); + expect(errors.status).toBe(200); + expect(errors.body.items).toEqual([]); + }); + + it('⑦ 分页中断：page2 抛错 → failed、水位停 page1、未拉到的行不出现；修复后增量补齐', async () => { + scenario( + [ + ext('ext-5', 'sync-u5', '周五', 8), + ext('ext-6', 'sync-u6', '吴六', 8), + ext('ext-7', 'sync-u7', '郑七', 8), + ], + { pageSize: 2, failFromPage: 1 }, + ); + const { run } = await triggerRun(); + + expect(run.state).toBe('failed'); + expect(run.error).not.toBeNull(); + expect(run.error).toContain('mock upstream failure'); + expect(run.counts).toEqual({ added: 2, updated: 0, disabled: 0, failed: 0 }); + expect(run.cursor).toBe('8'); // 水位停在最后成功页（page1 提交值） + + // page1 已应用行保留（幂等重入前提）；page2 的行不出现。 + expect(await userByAccount('sync-u5')).not.toBeNull(); + expect(await userByAccount('sync-u6')).not.toBeNull(); + expect(await userByAccount('sync-u7')).toBeNull(); + // 批次故障不牵连任何会话/停用（无用户被动）。 + expect( + await prisma.users.count({ where: { sourceStatus: 'disabled' } }), + ).toBe(0); + + const status = await getStatus(); + expect(status.body.state).toBe('failed'); + expect(status.body.sourceVersion).toBe(8); + + // 修复（撤故障注入）+ 版本推进 → 增量从水位 8 续上，补齐 ext-7。 + scenario([ext('ext-7', 'sync-u7', '郑七', 9)]); + const healed = await triggerRun(); + expect(healed.run.state).toBe('success'); + expect(healed.run.counts).toEqual({ added: 1, updated: 0, disabled: 0, failed: 0 }); + expect(await userByAccount('sync-u7')).not.toBeNull(); + }); + + it('⑧ 部分无效记录：1 坏行 → partial + errors 明细，其余照常；lastSuccessAt 不推进', async () => { + const before = await getStatus(); + const lastSuccessAtBefore = before.body.lastSuccessAt; + + scenario([ + ext('ext-8', 'sync-u8', '郑八', 10), + { ...ext('ext-bad', 'sync-bad', '坏行', 10), displayName: '' }, // 坏行：displayName 缺失 + ]); + const { run } = await triggerRun(); + + expect(run.state).toBe('partial'); + expect(run.counts).toEqual({ added: 1, updated: 0, disabled: 0, failed: 1 }); + expect(run.cursor).toBe('10'); // 坏行版本不可采信，水位取有效行最大版本 + expect(run.error).toBeNull(); + + // 坏行明细：invalid_record + 诊断详情；其余行照常应用。 + expect(await userByAccount('sync-u8')).not.toBeNull(); + const errors = await listErrors({ runId: run.id }); + expect(errors.status).toBe(200); + expect(errors.body.items).toHaveLength(1); + expect(errors.body.items[0]).toEqual( + expect.objectContaining({ + runId: run.id, + externalUserId: 'ext-bad', + code: 'invalid_record', + }), + ); + expect(errors.body.items[0].detail).toContain('displayName'); + expect(typeof errors.body.items[0].createdAt).toBe('string'); + + // partial 不是完全成功：lastSuccessAt 不推进。 + const after = await getStatus(); + expect(after.body.state).toBe('partial'); + expect(after.body.lastSuccessAt).toBe(lastSuccessAtBefore); + }); + + it('⑨ 手动补拉与定时互斥：运行中批次复用 runId；cron tick 撞锁跳过', async () => { + let releaseGate!: () => void; + const gate = new Promise((resolve) => { + releaseGate = resolve; + }); + scenario([ext('ext-9', 'sync-u9', '冯九', 11)], { gateFirstPage: gate }); + + const runsBefore = await prisma.user_sync_runs.count(); + + const first = await postRun(adminToken); + expect(first.status).toBe(202); + const runId = first.body.runId as string; + + // 批次卡在 page0（门闸）：第二次手动补拉复用同一 runId（202 语义）。 + const second = await postRun(adminToken); + expect(second.status).toBe(202); + expect(second.body.runId).toBe(runId); + + // 定时 tick 撞锁：静默跳过，不产生新批次。 + await engine.tick(); + expect(await prisma.user_sync_runs.count()).toBe(runsBefore + 1); + const inflight = await prisma.user_sync_runs.findUnique({ + where: { id: runId }, + }); + expect(inflight?.state).toBe('running'); + const inflightStatus = await getStatus(); + expect(inflightStatus.body.state).toBe('running'); + + // 放行 → 批次完成并落账。 + releaseGate(); + const done = await waitRunDone(runId); + expect(done.state).toBe('success'); + expect(done.counts).toEqual({ added: 1, updated: 0, disabled: 0, failed: 0 }); + expect(await userByAccount('sync-u9')).not.toBeNull(); + }); + + it('⑩ 增量 + 周期全量对账（每 10 批）：低版本行新变更增量不可见，对账批应用；缺席用户绝不停用', async () => { + // 对齐节奏：使「下下批」恰为第 10 批（对账批）。 + const stateRow = await prisma.user_sync_state.findUniqueOrThrow({ + where: { id: USER_SYNC_STATE_SINGLETON_ID }, + }); + const fillers = (10 - ((stateRow.runCount + 2) % 10)) % 10; + scenario([]); + for (let i = 0; i < fillers; i++) { + await triggerRun(); + } + + // 漂移构造：ext-2 改名到 v11 == 当前水位（低于/等于水位的行内变更 + // 对增量不可见——增量只回 v > 水位）。 + const usersBefore = await prisma.users.count(); + scenario([U2(11, '王二对账名')]); + + // 增量批：收到裸水位 "11"，过滤后空 → 全跳过，displayName 不变。 + const incremental = await triggerRun(); + expect(incremental.run.state).toBe('success'); + expect(incremental.run.counts).toEqual({ + added: 0, + updated: 0, + disabled: 0, + failed: 0, + }); + expect(mock.receivedCursors).toEqual(['11']); + expect((await userByAccount('sync-u2'))?.displayName).toBe('王二改'); + + // 对账批（第 10 批）：cursor=null 全量快照 → 按映射行版本比较应用。 + // （场景未重设 → receivedCursors 跨两批累计：['11', null] 恰为增量→全量 + // 形态切换的完整证据。） + const reconcile = await triggerRun(); + expect(mock.receivedCursors).toEqual(['11', null]); + expect(reconcile.run.state).toBe('success'); + expect(reconcile.run.counts).toEqual({ + added: 0, + updated: 1, + disabled: 0, + failed: 0, + }); + expect((await userByAccount('sync-u2'))?.displayName).toBe('王二对账名'); + expect(reconcile.run.cursor).toBe('11'); + + // 红线 2：快照里缺席的用户绝不被停用/删除。 + expect(await prisma.users.count()).toBe(usersBefore); + expect( + await prisma.users.count({ where: { sourceStatus: 'disabled' } }), + ).toBe(0); + }); + + it('⑪ §10 闭环：pending 登录 401（哨兵哈希不 500）→ activate → 仍 401 → reset-password → 可登录', async () => { + const u2 = await userByAccount('sync-u2'); + expect(u2?.credentialStatus).toBe('pending'); + + // 待激活 + 哨兵哈希：任何口令 401（而非 argon2 解析 500），文案同密码错误。 + const guess = await login('sync-u2', 'Guessing-123'); + expect(guess.status).toBe(401); + expect(guess.body.code).toBe(ErrorCodes.AUTH_INVALID_CREDENTIALS); + + // 激活（credentialStatus 维度）：哈希仍为哨兵 → 依旧 401。 + const activate = await request(app.getHttpServer()) + .post(`/${API_PREFIX}/admin/users/${u2?.id}/activate`) + .set('Authorization', `Bearer ${adminToken}`) + .send({}); + expect(activate.status).toBe(200); + expect(activate.body.credentialStatus).toBe('active'); + const still = await login('sync-u2', 'Guessing-123'); + expect(still.status).toBe(401); + expect(still.body.code).toBe(ErrorCodes.AUTH_INVALID_CREDENTIALS); + + // 初始凭据唯一出口 = 管理员 reset-password（生成）。 + const reset = await request(app.getHttpServer()) + .post(`/${API_PREFIX}/admin/users/${u2?.id}/reset-password`) + .set('Authorization', `Bearer ${adminToken}`) + .send({}); + expect(reset.status).toBe(200); + const initial = reset.body.newPassword as string; + expect(initial.length).toBeGreaterThanOrEqual(12); + const ok = await login('sync-u2', initial); + expect(ok.status).toBe(200); + expect(typeof ok.body.sessionToken).toBe('string'); + }); + + it('⑫ status/runs/errors API 形状 + runs 游标分页走全量 + 参数 400 + 权限矩阵', async () => { + // 形状：status 全键集 + counts 全键集。 + const status = await getStatus(); + expect(status.status).toBe(200); + expect(Object.keys(status.body).sort()).toEqual([ + 'counts', + 'lastRunAt', + 'lastSuccessAt', + 'sourceVersion', + 'state', + ]); + expect(Object.keys(status.body.counts).sort()).toEqual([ + 'added', + 'disabled', + 'failed', + 'updated', + ]); + + // 形状：run 条目全键集（cursor 投影为 int：水位数值串 → number）。 + const firstPage = await listRuns({ limit: 3 }); + expect(firstPage.status).toBe(200); + expect(firstPage.body.items.length).toBeGreaterThan(0); + expect(Object.keys(firstPage.body.items[0]).sort()).toEqual([ + 'counts', + 'cursor', + 'error', + 'finishedAt', + 'id', + 'startedAt', + 'state', + 'trigger', + ]); + expect(typeof firstPage.body.items[0].cursor).toBe('number'); + expect(typeof firstPage.body.items[0].startedAt).toBe('string'); + + // 游标分页走全量：与 DB 总数一致、无重复、startedAt 倒序、元信息成对。 + const total = await prisma.user_sync_runs.count(); + const seen: string[] = []; + let cursor: string | undefined; + for (;;) { + const page = await listRuns( + cursor === undefined ? { limit: '3' } : { limit: '3', cursor }, + ); + expect(page.status).toBe(200); + expect(page.body.hasMore).toBe(page.body.nextCursor !== null); + seen.push(...(page.body.items as { id: string; startedAt: string }[]).map((i) => i.id)); + const items = page.body.items as { startedAt: string }[]; + for (let i = 1; i < items.length; i++) { + expect(items[i - 1].startedAt >= items[i].startedAt).toBe(true); + } + cursor = page.body.nextCursor; + if (cursor === null || cursor === undefined) break; + } + expect(seen).toHaveLength(total); + expect(new Set(seen).size).toBe(total); + + // 参数校验：limit/cursor 非法 400；errors 缺 runId 400、未知 runId 空数组。 + expect((await listRuns({ limit: 'abc' })).status).toBe(400); + expect((await listRuns({ limit: '0' })).status).toBe(400); + expect((await listRuns({ cursor: '%%not-base64%%' })).status).toBe(400); + const noRunId = await listErrors({}); + expect(noRunId.status).toBe(400); + expect(noRunId.body.code).toBe(ErrorCodes.VALIDATION_FAILED); + const emptyRunId = await listErrors({ runId: '' }); + expect(emptyRunId.status).toBe(400); + const unknown = await listErrors({ runId: 'no-such-run' }); + expect(unknown.status).toBe(200); + expect(unknown.body).toEqual({ items: [] }); + + // 权限矩阵：staff 全端点 403（POST 不产生批次副作用）；匿名 401。 + expect((await getStatus(staffToken)).status).toBe(403); + const staffStatus = await getStatus(staffToken); + expect(staffStatus.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN); + expect((await listRuns({}, staffToken)).status).toBe(403); + expect((await listErrors({ runId: 'x' }, staffToken)).status).toBe(403); + const runsBeforePost = await prisma.user_sync_runs.count(); + const staffPost = await postRun(staffToken); + expect(staffPost.status).toBe(403); + expect(staffPost.body.code).toBe(ErrorCodes.AUTH_FORBIDDEN); + expect(await prisma.user_sync_runs.count()).toBe(runsBeforePost); + + expect((await getStatus(null)).status).toBe(401); + expect((await listRuns({}, null)).status).toBe(401); + expect((await listErrors({ runId: 'x' }, null)).status).toBe(401); + const anonPost = await request(app.getHttpServer()) + .post(`/${API_PREFIX}/admin/user-sync/runs`) + .send(undefined); + expect(anonPost.status).toBe(401); + expect(anonPost.body.code).toBe(ErrorCodes.AUTH_TOKEN_EXPIRED); + }); +});