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);
+ });
+});