feat(api): HttpUserSource adapter for uniservice open/v1/users (G4)
全量拉取+phone 全局去重（一人多档跨页）、水位游标不当页码回归守护、 分块续翻缓存；对齐 UserSourcePort 类型。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -0,0 +1,184 @@
|
||||
// services/api/src/user-sync/http-user-source.ts —— uniservice /open/v1/users
|
||||
// 真实源适配器（G4，docs/open-v1-users-api.md；USER_SYNC_SOURCE=http 装配）。
|
||||
//
|
||||
// 游标协议（引擎侧不透明，此处为权威定义）：
|
||||
// · cursor 空/未传/纯数值（引擎传入的已提交水位 = updated_at 秒级时间戳）
|
||||
// → 全量第 1 块。上游 API 无增量过滤参数（仅 page/page_size/company_id/
|
||||
// status/keyword/ids），增量不可行——每批全量重拉，幂等由引擎行级
|
||||
// sourceVersion 比较保证（213 用户/200 每页 ≈ 2 页，成本可忽略）。
|
||||
// 纯数值水位绝不当页码用（历史 bug：水位 1758520000 曾被 parseInt 成页号）。
|
||||
// · cursor="p:N"（N≥2）→ 本轮缓存的第 N 块续翻（不发上游请求）。
|
||||
// 拉取策略：首调用拉全上游页 → 全局按 phone 去重（一人多档：同自然人多公司
|
||||
// 多行，保留 updated_at 最新，平手取后见者）→ 按 CHUNK_SIZE 切块缓存。
|
||||
// 引擎互斥锁保证 listPage 串行，实例缓存无并发风险。
|
||||
// 稳定外部 ID：接口 id（int → string）。版本：updated_at epoch 秒（单调可比；
|
||||
// 解析失败得 NaN → 引擎 validateItem 记坏行，不污染水位）。
|
||||
// 状态映射：status=1 在职 → active；0 停用 / 2 离职 → disabled。
|
||||
// 无 phone 行跳过（V1.1 账号=手机号，无手机号无法登录；warn 落日志）。
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import type {
|
||||
ExternalUserRecord,
|
||||
UserSourcePage,
|
||||
UserSourcePort,
|
||||
} from './user-source.port';
|
||||
|
||||
/** 接口返回的用户对象（docs/open-v1-users-api.md §3 字段表；未用字段保留声明）。 */
|
||||
interface OpenV1User {
|
||||
id: number;
|
||||
sn: string;
|
||||
name: string;
|
||||
sex: number | null;
|
||||
email: string;
|
||||
phone: string;
|
||||
avatar: string | null;
|
||||
status: number;
|
||||
user_type: number;
|
||||
manager_id: number | null;
|
||||
company_id: number;
|
||||
union_id: string | null;
|
||||
created_at: string;
|
||||
updated_at: string;
|
||||
company_name: string;
|
||||
company_abbr: string;
|
||||
company_tenant_key: string;
|
||||
feishu_open_id: string | null;
|
||||
feishu_user_id: string | null;
|
||||
feishu_tenant_key: string | null;
|
||||
last_login_at: string | null;
|
||||
roles: string[];
|
||||
}
|
||||
|
||||
interface OpenV1Response {
|
||||
success: boolean;
|
||||
message: string;
|
||||
data: {
|
||||
items: OpenV1User[];
|
||||
total: number;
|
||||
page: number;
|
||||
page_size: number;
|
||||
};
|
||||
}
|
||||
|
||||
/** 上游分页上限（文档 §5：200/页）。 */
|
||||
const UPSTREAM_PAGE_SIZE = 200;
|
||||
|
||||
/** 喂引擎的块大小（引擎逐页提交水位——保留分页中断恢复点语义）。 */
|
||||
const CHUNK_SIZE = 200;
|
||||
|
||||
/** 上游页数护栏（total 与分页行为不一致时的死循环保护，1000 页 = 20 万用户）。 */
|
||||
const MAX_UPSTREAM_PAGES = 1_000;
|
||||
|
||||
/** 单请求超时。 */
|
||||
const REQUEST_TIMEOUT_MS = 15_000;
|
||||
|
||||
/** 本源命名空间（external_user_mappings.sourceSystem）。 */
|
||||
export const HTTP_USER_SOURCE_SYSTEM = 'uniservice';
|
||||
|
||||
/** "p:N"（N≥2）→ 第 N 块下标；空/数值水位/非法 → null（全量第 1 块）。 */
|
||||
function parseChunkCursor(raw: string | undefined): number | null {
|
||||
if (raw === undefined || raw === '' || !raw.startsWith('p:')) {
|
||||
return null;
|
||||
}
|
||||
const n = Number.parseInt(raw.slice(2), 10);
|
||||
return Number.isInteger(n) && n >= 2 ? n - 1 : null;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class HttpUserSource implements UserSourcePort {
|
||||
readonly sourceSystem = HTTP_USER_SOURCE_SYSTEM;
|
||||
|
||||
private readonly logger = new Logger(HttpUserSource.name);
|
||||
|
||||
/** 本轮全量块缓存（null = 未拉取；续翻直接下发，不再回源）。 */
|
||||
private chunks: ExternalUserRecord[][] | null = null;
|
||||
|
||||
async listPage(cursor?: string): Promise<UserSourcePage> {
|
||||
const index = parseChunkCursor(cursor);
|
||||
if (index === null) {
|
||||
this.chunks = await this.fetchAllChunks();
|
||||
}
|
||||
// 续翻时缓存理论恒在（同批互斥串行）；异常缺失则空页收尾——安全失败。
|
||||
const chunks = this.chunks ?? [];
|
||||
const i = index ?? 0;
|
||||
return {
|
||||
items: chunks[i] ?? [],
|
||||
nextCursor: i + 1 < chunks.length ? `p:${i + 2}` : null,
|
||||
};
|
||||
}
|
||||
|
||||
/** 拉全上游页 → phone 全局去重 → 切块。任何上游故障整批抛错（引擎记 failed 批次）。 */
|
||||
private async fetchAllChunks(): Promise<ExternalUserRecord[][]> {
|
||||
const { baseUrl, apiKey } = this.config();
|
||||
const rawUsers: OpenV1User[] = [];
|
||||
let exhausted = false;
|
||||
for (let page = 1; page <= MAX_UPSTREAM_PAGES; page++) {
|
||||
const params = new URLSearchParams({
|
||||
page: String(page),
|
||||
page_size: String(UPSTREAM_PAGE_SIZE),
|
||||
status: 'all',
|
||||
});
|
||||
const res = await fetch(`${baseUrl}?${params}`, {
|
||||
headers: { 'X-Api-Key': apiKey },
|
||||
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
||||
});
|
||||
if (res.status === 401) {
|
||||
throw new Error('用户源鉴权失败（Invalid API Key）——检查 OPEN_API_USERS_KEY');
|
||||
}
|
||||
if (!res.ok) {
|
||||
throw new Error(`用户源 HTTP ${res.status}: ${(await res.text()).slice(0, 200)}`);
|
||||
}
|
||||
const body = (await res.json()) as OpenV1Response;
|
||||
if (!body.success) {
|
||||
throw new Error(`用户源返回失败: ${body.message || 'success=false'}`);
|
||||
}
|
||||
rawUsers.push(...body.data.items);
|
||||
if (body.data.page * body.data.page_size >= body.data.total) {
|
||||
exhausted = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!exhausted) {
|
||||
throw new Error(`上游分页超出护栏（>${MAX_UPSTREAM_PAGES} 页），疑似 total 异常`);
|
||||
}
|
||||
|
||||
// 一人多档：同一自然人多公司各一行（不同 id），按 phone 归并——保留
|
||||
// updated_at 最新行（平手取后见者，上游 id 序稳定 → 确定性）。
|
||||
const byPhone = new Map<string, OpenV1User>();
|
||||
for (const raw of rawUsers) {
|
||||
const phone = raw.phone?.trim();
|
||||
if (!phone) {
|
||||
this.logger.warn(`跳过上游用户 id=${raw.id} name=${raw.name}：无手机号（V1.1 账号=手机号）`);
|
||||
continue;
|
||||
}
|
||||
const existing = byPhone.get(phone);
|
||||
if (!existing || raw.updated_at >= existing.updated_at) {
|
||||
byPhone.set(phone, raw);
|
||||
}
|
||||
}
|
||||
|
||||
const records = [...byPhone.values()].map<ExternalUserRecord>((raw) => ({
|
||||
externalUserId: String(raw.id),
|
||||
loginAccount: raw.phone.trim(),
|
||||
displayName: raw.name || raw.sn || raw.phone,
|
||||
department: raw.company_abbr || raw.company_name || '',
|
||||
sourceStatus: raw.status === 1 ? 'active' : 'disabled',
|
||||
sourceVersion: Math.floor(Date.parse(raw.updated_at) / 1000),
|
||||
updatedAt: new Date(raw.updated_at),
|
||||
}));
|
||||
|
||||
const chunks: ExternalUserRecord[][] = [];
|
||||
for (let i = 0; i < records.length; i += CHUNK_SIZE) {
|
||||
chunks.push(records.slice(i, i + CHUNK_SIZE));
|
||||
}
|
||||
return chunks;
|
||||
}
|
||||
|
||||
private config(): { baseUrl: string; apiKey: string } {
|
||||
const baseUrl = process.env.OPEN_API_USERS_URL;
|
||||
const apiKey = process.env.OPEN_API_USERS_KEY;
|
||||
if (!baseUrl || !apiKey) {
|
||||
throw new Error('HttpUserSource: OPEN_API_USERS_URL / OPEN_API_USERS_KEY 未配置');
|
||||
}
|
||||
return { baseUrl, apiKey };
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user