From d732caa739e43d0f9adc1236586e3806e37a0ef3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 22 Sep 2026 16:27:43 +0800 Subject: [PATCH] feat(api): HttpUserSource adapter for uniservice open/v1/users (G4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 全量拉取+phone 全局去重（一人多档跨页）、水位游标不当页码回归守护、 分块续翻缓存；对齐 UserSourcePort 类型。 Co-Authored-By: Claude Code --- .../api/src/user-sync/http-user-source.ts | 184 +++++++++++++++++ services/api/test/http-user-source.spec.ts | 185 ++++++++++++++++++ 2 files changed, 369 insertions(+) create mode 100644 services/api/src/user-sync/http-user-source.ts create mode 100644 services/api/test/http-user-source.spec.ts diff --git a/services/api/src/user-sync/http-user-source.ts b/services/api/src/user-sync/http-user-source.ts new file mode 100644 index 0000000..4ba2f56 --- /dev/null +++ b/services/api/src/user-sync/http-user-source.ts @@ -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 { + 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 { + 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(); + 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((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 }; + } +} diff --git a/services/api/test/http-user-source.spec.ts b/services/api/test/http-user-source.spec.ts new file mode 100644 index 0000000..cddae96 --- /dev/null +++ b/services/api/test/http-user-source.spec.ts @@ -0,0 +1,185 @@ +// services/api/test/http-user-source.spec.ts —— HttpUserSource 单测（G4 对接）。 +// 覆盖：跨页一人多档全局去重、数值水位游标不当页码（回归）、分块续翻 p:N、 +// 鉴权/失败路径、无手机号跳过。fetch 全程 mock，不发真实请求。 +import { HttpUserSource } from '../src/user-sync/http-user-source'; + +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[]; +} + +const BASE_USER: OpenV1User = { + id: 1, sn: 'S1', name: '张三', sex: null, email: 'z@example.com', + phone: '13800000001', avatar: null, status: 1, user_type: 1, manager_id: null, + company_id: 1, union_id: null, created_at: '2026-09-01T00:00:00Z', + updated_at: '2026-09-20T00:00:00Z', company_name: 'X公司', company_abbr: 'X', + company_tenant_key: 'tk', feishu_open_id: null, feishu_user_id: null, + feishu_tenant_key: null, last_login_at: null, roles: ['user'], +}; + +const user = (over: Partial = {}): OpenV1User => ({ ...BASE_USER, ...over }); + +const okBody = (items: OpenV1User[], total: number, page = 1, pageSize = 200) => ({ + success: true, + message: 'ok', + data: { items, total, page, page_size: pageSize }, +}); + +/** 最小 fetch Response 替身（仅适配器用到的字段）。 */ +function fakeRes(status: number, body: unknown) { + return { + status, + ok: status >= 200 && status < 300, + json: async () => body, + text: async () => JSON.stringify(body), + }; +} + +const originalFetch = global.fetch; +const originalEnv = { + url: process.env.OPEN_API_USERS_URL, + key: process.env.OPEN_API_USERS_KEY, +}; + +describe('HttpUserSource（uniservice /open/v1/users 适配器）', () => { + // 注意不能用大写 URL 作变量名——会遮蔽全局 URL 构造器。 + const SOURCE_URL = 'https://uniservice.example/open/v1/users'; + let source: HttpUserSource; + let fetchMock: jest.Mock; + + beforeEach(() => { + process.env.OPEN_API_USERS_URL = SOURCE_URL; + process.env.OPEN_API_USERS_KEY = 'test-key'; + source = new HttpUserSource(); + fetchMock = jest.fn(); + global.fetch = fetchMock as unknown as typeof fetch; + }); + + afterEach(() => { + global.fetch = originalFetch; + // 保存/恢复而非删除——避免清掉开发者本地 .env 装载的同名值。 + if (originalEnv.url === undefined) { + delete process.env.OPEN_API_USERS_URL; + } else { + process.env.OPEN_API_USERS_URL = originalEnv.url; + } + if (originalEnv.key === undefined) { + delete process.env.OPEN_API_USERS_KEY; + } else { + process.env.OPEN_API_USERS_KEY = originalEnv.key; + } + }); + + it('跨上游页的一人多档按 phone 全局去重，保留 updated_at 最新行；字段映射正确', async () => { + // echo page_size=2（服务端收窄）：page1=[id1,id3]、page2=[id2]——同 phone 两行分落两页。 + fetchMock + .mockResolvedValueOnce(fakeRes(200, okBody([ + user({ id: 1, name: '张三新档', updated_at: '2026-09-21T00:00:00Z' }), + user({ id: 3, name: '李四', phone: '13800000002', sn: 'S3', updated_at: '2026-09-18T00:00:00Z' }), + ], 3, 1, 2))) + .mockResolvedValueOnce(fakeRes(200, okBody([ + user({ id: 2, name: '张三旧档', phone: '13800000001', status: 2, updated_at: '2026-09-19T00:00:00Z' }), + ], 3, 2, 2))); + + const page = await source.listPage(); + + // 请求形状：page=1 起、page_size=200、status=all、X-Api-Key 头。 + const firstUrl = new URL(fetchMock.mock.calls[0][0] as string); + expect(firstUrl.searchParams.get('page')).toBe('1'); + expect(firstUrl.searchParams.get('page_size')).toBe('200'); + expect(firstUrl.searchParams.get('status')).toBe('all'); + expect((fetchMock.mock.calls[0][1] as { headers: Record }).headers['X-Api-Key']).toBe('test-key'); + + // 去重后 2 人；13800000001 保留 id=1（updated_at 最新），status=1 → active。 + expect(page.items).toHaveLength(2); + expect(page.nextCursor).toBeNull(); + const kept = page.items.find((i) => i.loginAccount === '13800000001'); + expect(kept).toMatchObject({ + externalUserId: '1', + displayName: '张三新档', + sourceStatus: 'active', + sourceVersion: Math.floor(Date.parse('2026-09-21T00:00:00Z') / 1000), + }); + expect(kept?.updatedAt).toEqual(new Date('2026-09-21T00:00:00Z')); + expect(page.items.find((i) => i.externalUserId === '2')).toBeUndefined(); + expect(source.sourceSystem).toBe('uniservice'); + }); + + it('数值水位游标不当页码：cursor="1758520000" 仍从第 1 页全量拉取（回归）', async () => { + fetchMock.mockResolvedValueOnce(fakeRes(200, okBody([user()], 1))); + + const page = await source.listPage('1758520000'); + + const url = new URL(fetchMock.mock.calls[0][0] as string); + expect(url.searchParams.get('page')).toBe('1'); + expect(page.items).toHaveLength(1); + expect(page.nextCursor).toBeNull(); + }); + + it('分块续翻：>200 行切两块，p:2 用缓存不再回源；新批（水位/空 cursor）重新全量拉', async () => { + const many = Array.from({ length: 205 }, (_, i) => + user({ id: i + 1, phone: `1380000${String(i).padStart(4, '0')}` }), + ); + fetchMock + .mockResolvedValueOnce(fakeRes(200, okBody(many.slice(0, 200), 205))) + .mockResolvedValueOnce(fakeRes(200, okBody(many.slice(200), 205, 2))); + + const first = await source.listPage(); + expect(first.items).toHaveLength(200); + expect(first.nextCursor).toBe('p:2'); + + const second = await source.listPage('p:2'); + expect(second.items).toHaveLength(5); + expect(second.nextCursor).toBeNull(); + expect(fetchMock).toHaveBeenCalledTimes(2); // 续翻走缓存 + + // 新一批（引擎传数值水位）→ 重新全量拉取。 + fetchMock.mockResolvedValueOnce(fakeRes(200, okBody([], 0))); + const fresh = await source.listPage('999'); + expect(fresh.items).toHaveLength(0); + expect(fetchMock).toHaveBeenCalledTimes(3); + }); + + it('失败路径：401 / HTTP 500 / success=false / env 缺失各自抛错', async () => { + fetchMock.mockResolvedValueOnce(fakeRes(401, { success: false, message: 'Invalid API Key', error_code: 'INVALID_API_KEY' })); + await expect(source.listPage()).rejects.toThrow(/Invalid API Key|鉴权失败/); + + fetchMock.mockResolvedValueOnce(fakeRes(500, { success: false, message: 'boom' })); + await expect(source.listPage()).rejects.toThrow(/HTTP 500/); + + fetchMock.mockResolvedValueOnce(fakeRes(200, { success: false, message: '查询失败' })); + await expect(source.listPage()).rejects.toThrow('查询失败'); + + delete process.env.OPEN_API_USERS_KEY; + await expect(source.listPage()).rejects.toThrow(/OPEN_API_USERS/); + }); + + it('无手机号行跳过（warn 不抛），有手机号行照常', async () => { + fetchMock.mockResolvedValueOnce( + fakeRes(200, okBody([user({ id: 1, phone: '' }), user({ id: 2, phone: '13900000002' })], 2)), + ); + const page = await source.listPage(); + expect(page.items).toHaveLength(1); + expect(page.items[0]?.externalUserId).toBe('2'); + }); +});