feat(api): scheduled user sync with mock source (T30 semantics)
P4a Task 7 —— cn 侧用户定期同步（mock 源先行，G4 真实接口到位后仅替换
UserSourcePort provider，引擎/契约/表不变）。
数据模型（db-cn 迁移 20260920114149，utf8mb4_0900_bin 逐表校对）：
- user_sync_runs：批次账本（trigger cron|manual、state running|success|
partial|failed、cursor 水位、counts JSON、起止时间与错误摘要）；
- user_sync_errors：单批次坏行明细（runId 索引）；
- external_user_mappings：本地↔外部 ID 映射（@@unique([sourceSystem,
externalUserId])，sourceVersion 逐行高水位）；
- user_sync_state：单行状态（lastRunAt/lastSuccessAt/cursor/runCount）。
同步引擎（UserSyncEngine）：
- Redis SETNX 分布式互斥（TTL 10min，锁值=runId，Lua CAS 释放）；
定时 tick 与手动触发共用，运行中批次复用 runId（202）；
- 游标增量 + 周期全量对账（RECONCILE_EVERY_N_RUNS 默认 10）；
- 逐页单事务提交（run counts+cursor / state cursor），分页中断→failed
且水位不前进、会话不动；
- 红线遵守：sync 不写 password_hash/role/album_members/localLocked；
空/部分页不误伤停用；上游恢复不解除本地锁定；
- 新用户按 pending 建（随机不可用哨兵哈希，激活+重置后方可登录）；
- 停用即撤销该用户全部会话；本地预建同名账号走关联而非重建。
Admin API（/admin/user-sync）：GET status|runs|errors?runId=、POST runs
（202 {runId}）；runs 为 startedAt+ID 倒序游标分页。
auth.service：登录对哨兵/损坏哈希加固（等代价 dummy verify，
401 而非 500，时序平坦）。
测试：user-sync.e2e-spec 12 用例（首轮全量/改名幂等/停用撤会话/
恢复不解锁/新增/空页/分页中断/部分无效/手动互斥/增量+对账/API 形状
与权限/pending 登录门）；api e2e 22 套件 239 用例全绿。
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -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：还有后续页时为 `<基数>#<P+1>`（增量基数=数值 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<void>;
|
||||
}
|
||||
|
||||
/** 生产默认健康场景：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<string | null> = [];
|
||||
|
||||
/** 整场景替换（浅拷贝隔离，避免用例间通过引用串改）。 */
|
||||
setScenario(next: MockScenario): void {
|
||||
this.scenario = { ...next };
|
||||
this.receivedCursors.length = 0;
|
||||
}
|
||||
|
||||
/** 恢复生产默认健康场景。 */
|
||||
reset(): void {
|
||||
this.setScenario(DEFAULT_SCENARIO);
|
||||
}
|
||||
|
||||
async listPage(cursor?: string): Promise<UserSourcePage> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -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<boolean> {
|
||||
const result = await this.getClient().set(
|
||||
USER_SYNC_LOCK_KEY,
|
||||
runId,
|
||||
'PX',
|
||||
LOCK_TTL_MS,
|
||||
'NX',
|
||||
);
|
||||
return result === 'OK';
|
||||
}
|
||||
|
||||
/** 当前锁值（持有中的 runId）；无锁为 null。 */
|
||||
async currentRunId(): Promise<string | null> {
|
||||
const value = await this.getClient().get(USER_SYNC_LOCK_KEY);
|
||||
return value;
|
||||
}
|
||||
|
||||
/** 值比对删除（仅锁值仍为 runId 时）；用于陈旧锁清理与正常释放。 */
|
||||
async releaseIfValue(runId: string): Promise<void> {
|
||||
await this.getClient().eval(
|
||||
RELEASE_SCRIPT,
|
||||
1,
|
||||
USER_SYNC_LOCK_KEY,
|
||||
runId,
|
||||
);
|
||||
}
|
||||
|
||||
async onModuleDestroy(): Promise<void> {
|
||||
if (this.connection !== null) {
|
||||
await this.connection.quit().catch(() => undefined);
|
||||
this.connection = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<UserSourcePage>;
|
||||
}
|
||||
|
||||
/** DI token（user-sync.module 装配 MockUserSource；e2e 经 app.get 注入场景）。 */
|
||||
export const USER_SOURCE = Symbol('USER_SOURCE');
|
||||
@@ -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<string, unknown>;
|
||||
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<UserSyncStatusResponse> {
|
||||
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<UserSyncRunListResponse> {
|
||||
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<UserSyncErrorListResponse> {
|
||||
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(),
|
||||
})),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -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<UserSyncStatusResponse> {
|
||||
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<UserSyncRunListResponse> {
|
||||
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<UserSyncErrorListResponse> {
|
||||
if (runId === undefined || runId === '') {
|
||||
throw validationFailed('runId 必填（批次 ID）');
|
||||
}
|
||||
return this.adminService.listErrors(userOf(req), runId);
|
||||
}
|
||||
}
|
||||
@@ -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 {}
|
||||
@@ -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:<random>`（非法 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<string> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<ItemOutcome> {
|
||||
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);
|
||||
}
|
||||
Reference in New Issue
Block a user