diff --git a/.env.example b/.env.example index 28016f9..665073a 100644 --- a/.env.example +++ b/.env.example @@ -19,3 +19,10 @@ KEYS_MASTER_KEY=change-me-32-bytes-base64 # 上游 NewAPI 站点地址（同步引擎调用的 BaseURL，不带尾部斜杠） NEWAPI_BASE=https://your-newapi-instance.example.com + +# 上游客户端实现：http（默认，调真实站点）/ fixture（开发期读仓库根样例快照，见实施计划 §6） +# 仅在离线开发 / 本地验证时设为 fixture——生产必须保持 http +# UPSTREAM_SOURCE=fixture + +# 样例文件目录（仅 fixture 实现使用；留空 = 仓库根） +# NEWAPI_FIXTURE_DIR= diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index a33288b..2091221 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -29,6 +29,13 @@ app.route('/api/manage/qpu', qpuRoutes) // 同步引擎：随进程启动，启动即跑一轮（同步引擎设计 §3.2） startSyncEngine() +// 统一错误信封（契约 约定 20）：未捕获异常 → 500 internal，未知路由 → 404 not_found +app.onError((err, c) => { + console.error('[server] 未处理异常', err) + return c.json({ error: { code: 'internal', message: '服务端内部错误' } }, 500) +}) +app.notFound((c) => c.json({ error: { code: 'not_found', message: '无此端点' } }, 404)) + console.log(`[server] Hono 已就绪，监听 :${env.port}`) export default { diff --git a/apps/server/src/lib/api-error.ts b/apps/server/src/lib/api-error.ts new file mode 100644 index 0000000..5c81dce --- /dev/null +++ b/apps/server/src/lib/api-error.ts @@ -0,0 +1,34 @@ +/** + * 错误响应的统一构造（API 契约 约定 20/21：非 2xx + `{ error: { code, message } }` 信封）。 + */ +import type { ContentfulStatusCode } from 'hono/utils/http-status' +import type { Context } from 'hono' +import type { ApiErrorCode } from '@naku/shared' + +import { UpstreamError } from '../upstream/errors' + +export function apiError( + c: Context, + status: ContentfulStatusCode, + code: ApiErrorCode, + message: string, +) { + return c.json({ error: { code, message } }, status) +} + +/** + * 上游错误 → 契约错误码：`key_rejected` → 400 · `upstream_unavailable` → 502。 + * 非上游错误返回 null（调用方 rethrow，交给 onError）。 + */ +export function upstreamErrorResponse(c: Context, err: unknown) { + if (!(err instanceof UpstreamError)) return null + if (err.kind === 'key_rejected') return apiError(c, 400, 'key_rejected', err.message) + return apiError(c, 502, 'upstream_unavailable', err.message) +} + +/** PostgreSQL 唯一约束冲突（并发登记同一 key 时的兜底判定，SQLSTATE 23505） */ +export function isUniqueViolation(err: unknown): boolean { + return ( + typeof err === 'object' && err !== null && (err as { code?: string }).code === '23505' + ) +} diff --git a/apps/server/src/lib/env.ts b/apps/server/src/lib/env.ts index 82b153a..4877bdc 100644 --- a/apps/server/src/lib/env.ts +++ b/apps/server/src/lib/env.ts @@ -2,6 +2,9 @@ * 环境变量（部署时经 env 注入；全量清单见仓库根 `.env.example`）。 * 缺关键变量时在「首次使用处」报错，不在进程启动时硬崩（便于本地起骨架）。 */ +// 副作用导入：显式加载仓库根 `.env`（Bun 只自动加载 cwd 的 .env，见 load-env.ts） +import './load-env' + function required(name: string): string { const v = process.env[name] if (!v) { @@ -21,6 +24,11 @@ export const env = { keysMasterKey: process.env.KEYS_MASTER_KEY ?? '', /** 上游 NewAPI 站点地址（同步引擎调用的 BaseURL） */ newapiBase: process.env.NEWAPI_BASE ?? '', + /** 上游客户端实现：`http`（默认，真实站点）/ `fixture`（开发期样例文件，见实施计划 §6） */ + upstreamSource: + process.env.UPSTREAM_SOURCE === 'fixture' ? ('fixture' as const) : ('http' as const), + /** 样例文件目录（仅 fixture 实现使用；空 = 仓库根） */ + newapiFixtureDir: process.env.NEWAPI_FIXTURE_DIR ?? '', } export { required } diff --git a/apps/server/src/lib/keys.ts b/apps/server/src/lib/keys.ts new file mode 100644 index 0000000..9072641 --- /dev/null +++ b/apps/server/src/lib/keys.ts @@ -0,0 +1,18 @@ +/** + * 凭证密码学单例：从 `KEYS_MASTER_KEY` 派生（懒初始化，缺变量的报错发生在首次使用处）。 + * 主密钥丢失 = 全部 key 需重新登记（见 api_keys 表设计 §3.1 与部署文档）。 + */ +import { createKeySecrets, type KeySecrets } from './crypto' +import { env } from './env' + +let secrets: KeySecrets | null = null + +export function getKeySecrets(): KeySecrets { + if (!secrets) { + if (!env.keysMasterKey) { + throw new Error('缺少环境变量 KEYS_MASTER_KEY（见 .env.example）') + } + secrets = createKeySecrets(env.keysMasterKey) + } + return secrets +} diff --git a/apps/server/src/lib/serialize.ts b/apps/server/src/lib/serialize.ts new file mode 100644 index 0000000..9ed5613 --- /dev/null +++ b/apps/server/src/lib/serialize.ts @@ -0,0 +1,54 @@ +/** + * 库行 → 契约响应行的序列化（API 契约 约定 3：字段 snake_case 与表列同名同形； + * 约定 9/11：时间戳一律北京时间 `+08:00`）。 + * 各表 1:1 映射，敏感列（`key_hash` / `key_ciphertext`）永不出接口。 + */ +import type { ApiKeyRow, PurchaseRow, QpuRow } from '@naku/shared' + +import { toBeijingIso } from './time' +import type { apiKeys, purchases, qpuHistory } from '../db/schema' + +/** 行内可空时间戳 → 契约输出（null 保持 null） */ +const iso = (d: Date | null): string | null => (d ? toBeijingIso(d) : null) + +export function toApiKeyRow(row: typeof apiKeys.$inferSelect): ApiKeyRow { + return { + id: row.id, + token_name: row.tokenName, + status: row.status as ApiKeyRow['status'], + disabled_source: row.disabledSource as ApiKeyRow['disabled_source'], + consecutive_401s: row.consecutive401s, + registered_at: toBeijingIso(row.registeredAt), + log_synced_at: iso(row.logSyncedAt), + usage_synced_at: iso(row.usageSyncedAt), + total_granted: row.totalGranted, + total_used: row.totalUsed, + total_available: row.totalAvailable, + unlimited_quota: row.unlimitedQuota, + model_limits: row.modelLimits as Record, + model_limits_enabled: row.modelLimitsEnabled, + expires_at: iso(row.expiresAt), + } +} + +export function toPurchaseRow(row: typeof purchases.$inferSelect): PurchaseRow { + return { + id: row.id, + kind: row.kind as PurchaseRow['kind'], + paid_at: toBeijingIso(row.paidAt), + paid_amount: row.paidAmount, + note: row.note, + period_start: iso(row.periodStart), + period_end: iso(row.periodEnd), + subscription_id: row.subscriptionId, + quota_granted: row.quotaGranted, + } +} + +export function toQpuRow(row: typeof qpuHistory.$inferSelect): QpuRow { + return { + id: row.id, + qpu: row.qpu, + effective_from: toBeijingIso(row.effectiveFrom), + } +} diff --git a/apps/server/src/routes/manage/api-keys.ts b/apps/server/src/routes/manage/api-keys.ts index f1cc427..f10ac62 100644 --- a/apps/server/src/routes/manage/api-keys.ts +++ b/apps/server/src/routes/manage/api-keys.ts @@ -1,6 +1,17 @@ import { Hono } from 'hono' +import { asc, eq } from 'drizzle-orm' +import type { PatchApiKeyRequest, RegisterApiKeyRequest } from '@naku/shared' +import { apiError, isUniqueViolation, upstreamErrorResponse } from '../../lib/api-error' +import { getKeySecrets } from '../../lib/keys' +import { toApiKeyRow } from '../../lib/serialize' +import { expiresAtFromUnix } from '../../lib/time' import { manageAuth } from '../../middleware/manage-auth' +import { getDb } from '../../db/client' +import { apiKeys } from '../../db/schema' +import { requestImmediateSync } from '../../sync' +import { getUpstreamClient } from '../../upstream/client' +import type { UpstreamTokenUsage } from '../../upstream/types' /** * 管理面 · key 管理（契约 §2.2）：清单 / 登记 / 启停。 @@ -10,41 +21,152 @@ export const apiKeysRoutes = new Hono() apiKeysRoutes.use(manageAuth) +/** 上游字段全组（登记 / 启用验证成功时一起写，见 api_keys 表设计 §3.4/§3.5） */ +function upstreamColumns(usage: UpstreamTokenUsage) { + return { + tokenName: usage.name, + totalGranted: usage.total_granted, + totalUsed: usage.total_used, + totalAvailable: usage.total_available, + unlimitedQuota: usage.unlimited_quota, + modelLimits: usage.model_limits, + modelLimitsEnabled: usage.model_limits_enabled, + expiresAt: expiresAtFromUnix(usage.expires_at), + } +} + /** 清单（按 id 升序，不分页） */ -apiKeysRoutes.get('/', (c) => - c.json( - { - error: { - code: 'internal', - message: '未实现：key 清单（见 docs/design/API契约.md §2.2）', - }, - }, - 501, - ), -) +apiKeysRoutes.get('/', async (c) => { + const rows = await getDb().select().from(apiKeys).orderBy(asc(apiKeys.id)) + return c.json(rows.map(toApiKeyRow)) +}) /** 登记：body `{ key }`；上游验证通过 → 入库（不阻塞首轮同步） */ -apiKeysRoutes.post('/', (c) => - c.json( - { - error: { - code: 'internal', - message: '未实现：登记 key（见 docs/design/API契约.md §2.2）', - }, - }, - 501, - ), -) +apiKeysRoutes.post('/', async (c) => { + let body: RegisterApiKeyRequest + try { + body = (await c.req.json()) as RegisterApiKeyRequest + } catch { + return apiError(c, 400, 'invalid_params', '请求体必须是 JSON') + } + const plainKey = typeof body?.key === 'string' ? body.key.trim() : '' + if (!plainKey) { + return apiError(c, 400, 'invalid_params', 'key 缺失或为空') + } + + const db = getDb() + const secrets = getKeySecrets() + const keyHash = secrets.keyHash(plainKey) + + const existing = await db + .select({ id: apiKeys.id }) + .from(apiKeys) + .where(eq(apiKeys.keyHash, keyHash)) + .limit(1) + if (existing.length > 0) { + return apiError(c, 409, 'already_registered', '该 key 已登记') + } + + // 上游验证（GET /api/usage/token）——同一次成功调用即一次额度采集： + // 上游字段全组 + usage_synced_at 随登记写入；log_synced_at 由引擎首轮补同步（契约 §2.2） + const requestedAt = new Date() + let usage: UpstreamTokenUsage + try { + usage = await getUpstreamClient().fetchTokenUsage(plainKey) + } catch (err) { + const res = upstreamErrorResponse(c, err) + if (res) return res + throw err + } + + try { + const [row] = await db + .insert(apiKeys) + .values({ + keyHash, + keyCiphertext: secrets.encryptKey(plainKey), + status: 'active', + disabledSource: null, + consecutive401s: 0, + usageSyncedAt: requestedAt, + ...upstreamColumns(usage), + }) + .returning() + requestImmediateSync() // 首轮日志同步：立即异步补一轮，不阻塞登记响应（假设 #7） + return c.json(toApiKeyRow(row!), 201) + } catch (err) { + if (isUniqueViolation(err)) { + return apiError(c, 409, 'already_registered', '该 key 已登记') + } + throw err + } +}) /** 启停：body `{ status: 'active' | 'disabled' }`；启用需上游验证、成功清零 consecutive_401s */ -apiKeysRoutes.patch('/:id', (c) => - c.json( - { - error: { - code: 'internal', - message: '未实现：启停（见 docs/design/API契约.md §2.2）', - }, - }, - 501, - ), -) +apiKeysRoutes.patch('/:id', async (c) => { + const id = Number(c.req.param('id')) + if (!Number.isInteger(id) || id <= 0) { + return apiError(c, 400, 'invalid_params', 'id 非法') + } + let body: PatchApiKeyRequest + try { + body = (await c.req.json()) as PatchApiKeyRequest + } catch { + return apiError(c, 400, 'invalid_params', '请求体必须是 JSON') + } + const status = body?.status + if (status !== 'active' && status !== 'disabled') { + return apiError(c, 400, 'invalid_params', "status 必须是 'active' 或 'disabled'") + } + + const db = getDb() + const [row] = await db.select().from(apiKeys).where(eq(apiKeys.id, id)).limit(1) + if (!row) { + return apiError(c, 404, 'not_found', `id=${id} 的 key 不存在`) + } + // 幂等：提交当前已是的状态直接返回（不重复验证、不覆盖停用来源） + if (row.status === status) { + return c.json(toApiKeyRow(row)) + } + + if (status === 'disabled') { + const [updated] = await db + .update(apiKeys) + .set({ status: 'disabled', disabledSource: 'manual' }) + .where(eq(apiKeys.id, id)) + .returning() + return c.json(toApiKeyRow(updated!)) + } + + // 启用：上游验证通过 → active + 清 disabled_source + consecutive_401s 清零 + // （验证调用本身即一次成功调用，否则下一轮若 401 会立刻再次摘除——契约 §2.2） + let plainKey: string + try { + plainKey = getKeySecrets().decryptKey(Buffer.from(row.keyCiphertext)) + } catch (err) { + console.error(`[manage] key_id=${id} 解密失败（KEYS_MASTER_KEY 是否变更？）`, err) + return apiError(c, 500, 'internal', '凭证解密失败：KEYS_MASTER_KEY 变更后需重新登记该 key') + } + const requestedAt = new Date() + let usage: UpstreamTokenUsage + try { + usage = await getUpstreamClient().fetchTokenUsage(plainKey) + } catch (err) { + const res = upstreamErrorResponse(c, err) + if (res) return res + throw err + } + + const [updated] = await db + .update(apiKeys) + .set({ + status: 'active', + disabledSource: null, + consecutive401s: 0, + usageSyncedAt: requestedAt, + ...upstreamColumns(usage), + }) + .where(eq(apiKeys.id, id)) + .returning() + return c.json(toApiKeyRow(updated!)) +}) diff --git a/apps/server/src/sync/index.ts b/apps/server/src/sync/index.ts index bc5f673..a8eb408 100644 --- a/apps/server/src/sync/index.ts +++ b/apps/server/src/sync/index.ts @@ -37,9 +37,19 @@ async function tick(): Promise { } } -/** 启动引擎：立即跑一轮，随后每 5 分钟一次 */ +/** + * 启动引擎：立即跑一轮，随后每 5 分钟一次 + */ export function startSyncEngine(): void { console.log('[sync] 引擎启动：立即跑一轮，随后每 5 分钟一次') void tick() setInterval(() => void tick(), TICK_MS) } + +/** + * 「立即异步补一轮」——登记 / 启用成功后触发（API 契约 §2.2：不阻塞响应、不等 tick）。 + * 防重入由 tick 自身承担：轮次进行中触发时本轮自然跳过。 + */ +export function requestImmediateSync(): void { + void tick() +} diff --git a/apps/server/src/upstream/client.ts b/apps/server/src/upstream/client.ts new file mode 100644 index 0000000..d4bd596 --- /dev/null +++ b/apps/server/src/upstream/client.ts @@ -0,0 +1,31 @@ +/** + * 上游客户端：接口 + 工厂（实施计划 §6「上游客户端接口化」）。 + * 引擎与路由依赖本接口；实现按 `UPSTREAM_SOURCE` 选择——`http`（默认，真实站点） + * / `fixture`（开发期样例文件，见 fixture.ts）。 + */ +import { env } from '../lib/env' +import { createFixtureClient } from './fixture' +import { createHttpClient } from './http' +import type { UpstreamLogRecord, UpstreamStatus, UpstreamTokenUsage } from './types' + +export interface UpstreamClient { + /** 接口⑧：token 级用量汇总（登记 / 启用验证 + 额度采集）；失败抛 UpstreamError */ + fetchTokenUsage(plainKey: string): Promise + /** 接口⑨：最近 1000 条调用日志（同步引擎唯一数据源） */ + fetchLogs(plainKey: string): Promise + /** 接口④：站点配置（QPU 采样；无需 key） */ + fetchStatus(): Promise +} + +let client: UpstreamClient | null = null + +/** 进程级单例（实现选择只发生在首次调用） */ +export function getUpstreamClient(): UpstreamClient { + if (!client) { + client = + env.upstreamSource === 'fixture' ? createFixtureClient() : createHttpClient() + } + return client +} + +export type { UpstreamLogRecord, UpstreamStatus, UpstreamTokenUsage } diff --git a/apps/server/src/upstream/errors.ts b/apps/server/src/upstream/errors.ts new file mode 100644 index 0000000..ab676af --- /dev/null +++ b/apps/server/src/upstream/errors.ts @@ -0,0 +1,26 @@ +/** + * 上游调用失败的两种对外语义（API 契约 约定 21）： + * - `key_rejected`：上游明确拒绝该 key（HTTP 401 / 403——key 不存在、显式禁用、账号被封禁） + * - `upstream_unavailable`：上游故障 / 超时 / 响应异常（无法完成验证，非 key 的问题） + * + * 唯一 401 信号的使用见同步引擎设计 §3.4（摘除计数只认 401，故 `status` 原样带上）。 + */ +export type UpstreamErrorKind = 'key_rejected' | 'upstream_unavailable' + +export class UpstreamError extends Error { + readonly kind: UpstreamErrorKind + /** 上游 HTTP 状态码；网络层故障（超时 / 拒连）时为 undefined */ + readonly status?: number + + constructor(kind: UpstreamErrorKind, message: string, status?: number) { + super(message) + this.name = 'UpstreamError' + this.kind = kind + this.status = status + } +} + +/** 把任意抛出物转成一行可读消息（用于日志 / 错误 message） */ +export function errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err) +} diff --git a/apps/server/src/upstream/fixture.ts b/apps/server/src/upstream/fixture.ts new file mode 100644 index 0000000..e9d5d79 --- /dev/null +++ b/apps/server/src/upstream/fixture.ts @@ -0,0 +1,107 @@ +/** + * 上游客户端 · 样例文件实现（实施计划 §6：仓库根实测响应快照 —— `api.log.token.json` + * / `api.status.json`，未入库、绝不提交）。开发期以 `UPSTREAM_SOURCE=fixture` 启用。 + * + * 快照缺失的接口（`/api/usage/token` 无快照）就地按实测结构合成：总量类字段取自 + * `api.log.token.json` 首条日志的 `other.subscription_*`（真实形状），保证 fixture + * 下的登记结果与 seed 数据（阶段 4）同一形状。 + * + * 错误分支的仿真约定（key 内容大小写不敏感；仅 fixture 模式生效，见实施计划「实现假设」）： + * 含 `reject` → 401 key_rejected · 含 `forbidden` → 403 key_rejected · 含 `unavailable` → 网络故障 + */ +import { readFileSync } from 'node:fs' + +import { DEFAULT_QUOTA_PER_UNIT } from '@naku/shared' + +import { env } from '../lib/env' +import { errorMessage, UpstreamError } from './errors' +import type { UpstreamClient } from './client' +import type { UpstreamLogRecord, UpstreamStatus, UpstreamTokenUsage } from './types' + +/** 仓库根（本文件位于 apps/server/src/upstream/）——样例快照的默认目录 */ +const REPO_ROOT = new URL('../../../../', import.meta.url) + +export interface FixtureUpstreamOptions { + /** 样例文件目录；缺省取 env.newapiFixtureDir，再退到仓库根 */ + dir?: string | URL +} + +function readSnapshot(dir: string | URL, name: string): T { + try { + return JSON.parse(readFileSync(new URL(name, dir), 'utf8')) as T + } catch (err) { + throw new UpstreamError( + 'upstream_unavailable', + `样例文件不可用（${name}）：${errorMessage(err)}`, + ) + } +} + +function asNumber(v: unknown, fallback: number): number { + return typeof v === 'number' && Number.isFinite(v) ? v : fallback +} + +export function createFixtureClient(options: FixtureUpstreamOptions = {}): UpstreamClient { + // env.newapiFixtureDir 为「空 = 仓库根」的语义，故用 ||（空字符串要退到 REPO_ROOT） + const dir = options.dir || env.newapiFixtureDir || REPO_ROOT + + /** 仿真约定：错误分支由 key 内容触发 */ + function simulate(key: string): void { + const k = key.toLowerCase() + if (k.includes('reject')) throw new UpstreamError('key_rejected', 'fixture：模拟上游拒绝该 key（401）', 401) + if (k.includes('forbidden')) throw new UpstreamError('key_rejected', 'fixture：模拟账号被封禁（403）', 403) + if (k.includes('unavailable')) { + throw new UpstreamError('upstream_unavailable', 'fixture：模拟上游不可达（网络故障）') + } + } + + function loadLogs(): UpstreamLogRecord[] { + const snap = readSnapshot<{ data?: UpstreamLogRecord[] }>(dir, 'api.log.token.json') + if (!Array.isArray(snap.data)) { + throw new UpstreamError('upstream_unavailable', '样例文件结构异常（api.log.token.json 缺 data[]）') + } + return snap.data + } + + return { + async fetchTokenUsage(plainKey: string): Promise { + simulate(plainKey) + const first = loadLogs()[0] + let other: Record = {} + if (first?.other) { + try { + other = JSON.parse(first.other) as Record + } catch { + other = {} // 样例 other 解析失败不致命：退到兜底值 + } + } + const totalGranted = asNumber(other.subscription_total, 800_000_000) + const totalUsed = asNumber(other.subscription_used, 85_579_458) + return { + name: first?.token_name ?? 'fixture-token', + total_granted: totalGranted, + total_used: totalUsed, + total_available: totalGranted - totalUsed, + unlimited_quota: false, + model_limits: {}, + model_limits_enabled: false, + expires_at: 0, // 永不过期 + } + }, + async fetchLogs(plainKey: string): Promise { + simulate(plainKey) + return loadLogs() + }, + async fetchStatus(): Promise { + const snap = readSnapshot<{ data?: Record }>(dir, 'api.status.json') + const data = snap.data + if (!data || typeof data !== 'object') { + throw new UpstreamError('upstream_unavailable', '样例文件结构异常（api.status.json 缺 data）') + } + return { + ...data, + quota_per_unit: asNumber(data.quota_per_unit, DEFAULT_QUOTA_PER_UNIT), + } as UpstreamStatus + }, + } +} diff --git a/apps/server/src/upstream/http.ts b/apps/server/src/upstream/http.ts new file mode 100644 index 0000000..3060ad3 --- /dev/null +++ b/apps/server/src/upstream/http.ts @@ -0,0 +1,92 @@ +/** + * 上游客户端 · 真实 HTTP 实现（实施计划 §6「接口化」的两套实现之一）。 + * 三个调用（usage / logs / status）共用同一套失败分类：401 / 403 → key_rejected， + * 其余非 2xx、网络故障、超时、响应结构异常 → upstream_unavailable（同步引擎设计 §3.4）。 + */ +import { env } from '../lib/env' +import { errorMessage, UpstreamError } from './errors' +import type { UpstreamClient } from './client' +import type { UpstreamLogRecord, UpstreamStatus, UpstreamTokenUsage } from './types' + +/** 上游请求超时（实施计划「实现假设」#3：30 秒；失败即本轮失败，不单轮重试） */ +export const UPSTREAM_TIMEOUT_MS = 30_000 + +/** 只用到 fetch 的最小子集（便于单测注入打桩） */ +export type FetchLike = (url: string, init?: RequestInit) => Promise + +export interface HttpUpstreamOptions { + /** BaseURL；缺省取 env.newapiBase */ + base?: string + /** 便于单测注入；缺省用全局 fetch */ + fetchImpl?: FetchLike + timeoutMs?: number +} + +/** 剥掉 `{ code/success, message, data }` 传输层包装；业务失败或结构异常抛 upstream_unavailable */ +function unwrapData(body: unknown, path: string, status: number): unknown { + if (body && typeof body === 'object') { + const b = body as Record + if (b.code === false || b.success === false) { + throw new UpstreamError( + 'upstream_unavailable', + `上游业务失败（${path}）：${typeof b.message === 'string' ? b.message : '未知原因'}`, + status, + ) + } + if ('data' in b) return b.data + } + throw new UpstreamError('upstream_unavailable', `上游响应结构异常（${path}）`, status) +} + +export function createHttpClient(options: HttpUpstreamOptions = {}): UpstreamClient { + const base = (options.base ?? env.newapiBase).replace(/\/+$/, '') + const doFetch = options.fetchImpl ?? fetch + const timeoutMs = options.timeoutMs ?? UPSTREAM_TIMEOUT_MS + + async function request(path: string, key?: string): Promise { + if (!base) { + throw new UpstreamError('upstream_unavailable', '未配置 NEWAPI_BASE（见 .env.example）') + } + let res: Response + try { + res = await doFetch(`${base}${path}`, { + headers: key ? { Authorization: `Bearer ${key}` } : undefined, + signal: AbortSignal.timeout(timeoutMs), + }) + } catch (err) { + throw new UpstreamError( + 'upstream_unavailable', + `请求上游失败（${path}）：${errorMessage(err)}`, + ) + } + if (res.status === 401 || res.status === 403) { + throw new UpstreamError('key_rejected', `上游拒绝该 key（HTTP ${res.status}）`, res.status) + } + if (!res.ok) { + throw new UpstreamError('upstream_unavailable', `上游返回 HTTP ${res.status}（${path}）`, res.status) + } + let body: unknown + try { + body = await res.json() + } catch (err) { + throw new UpstreamError( + 'upstream_unavailable', + `上游响应不是合法 JSON（${path}）：${errorMessage(err)}`, + res.status, + ) + } + return unwrapData(body, path, res.status) + } + + return { + async fetchTokenUsage(plainKey: string): Promise { + return (await request('/api/usage/token', plainKey)) as UpstreamTokenUsage + }, + async fetchLogs(plainKey: string): Promise { + return (await request('/api/log/token', plainKey)) as UpstreamLogRecord[] + }, + async fetchStatus(): Promise { + return (await request('/api/status')) as UpstreamStatus + }, + } +} diff --git a/apps/server/src/upstream/types.ts b/apps/server/src/upstream/types.ts new file mode 100644 index 0000000..5b85285 --- /dev/null +++ b/apps/server/src/upstream/types.ts @@ -0,0 +1,53 @@ +/** + * 上游 NewAPI 响应类型（字段级事实见 `docs/upstream/NewAPI-sk-key接口文档.md`）。 + * 只声明本项目消费的字段，其余原样透传（索引签名兜底）。 + */ + +/** 接口⑧ `GET /api/usage/token` → `data`（`api_keys` 表 8 个上游字段的来源） */ +export interface UpstreamTokenUsage { + name: string + total_granted: number + total_used: number + total_available: number + unlimited_quota: boolean + model_limits: Record + model_limits_enabled: boolean + /** Unix 秒；0 = 永不过期（落库转 NULL，见 api_keys 表设计 §3.5） */ + expires_at: number +} + +/** + * 接口⑨ `GET /api/log/token` → `data[]` 的一行：上游 21 字段原形。 + * `other` 是 JSON **字符串**（解析入库归同步引擎）；`upstream_request_id` 为 omitempty 可选字段。 + */ +export interface UpstreamLogRecord { + id: number + user_id: number + created_at: number + type: number + content: string + username: string + token_name: string + model_name: string + quota: number + prompt_tokens: number + completion_tokens: number + use_time: number + is_stream: boolean + channel: number + channel_name: string + token_id: number + group: string + ip: string + request_id: string + upstream_request_id?: string + other: string +} + +/** 接口④ `GET /api/status` → `data`（QPU 采样只需个别字段，其余透传） */ +export interface UpstreamStatus { + quota_per_unit: number + quota_display_type?: string + usd_exchange_rate?: number + [key: string]: unknown +} diff --git a/apps/server/src/upstream/upstream.test.ts b/apps/server/src/upstream/upstream.test.ts new file mode 100644 index 0000000..46f0868 --- /dev/null +++ b/apps/server/src/upstream/upstream.test.ts @@ -0,0 +1,129 @@ +/** + * 上游客户端单测（bun test）：失败分类 + 样例实现。 + * HTTP 实现用注入的 fetch 打桩；样例实现读仓库根快照（缺失时跳过文件相关断言）。 + */ +import { describe, expect, test } from 'bun:test' +import { existsSync } from 'node:fs' + +import { UpstreamError } from './errors' +import { createFixtureClient } from './fixture' +import { createHttpClient, type FetchLike } from './http' + +const jsonRes = (body: unknown, status = 200): Response => + new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json' }, + }) + +function httpWith(fetchImpl: FetchLike) { + return createHttpClient({ base: 'https://upstream.test', fetchImpl }) +} + +describe('HTTP 实现 · 失败分类', () => { + test('401 → key_rejected（带状态码）', async () => { + const client = httpWith(async () => jsonRes({ message: 'invalid token' }, 401)) + const err = await client.fetchTokenUsage('sk-x').catch((e: unknown) => e) + expect(err).toBeInstanceOf(UpstreamError) + expect((err as UpstreamError).kind).toBe('key_rejected') + expect((err as UpstreamError).status).toBe(401) + }) + + test('403 → key_rejected（账号封禁同样视为「上游拒绝该 key」）', async () => { + const client = httpWith(async () => jsonRes({ message: 'banned' }, 403)) + const err = await client.fetchLogs('sk-x').catch((e: unknown) => e) + expect((err as UpstreamError).kind).toBe('key_rejected') + expect((err as UpstreamError).status).toBe(403) + }) + + test('500 → upstream_unavailable', async () => { + const client = httpWith(async () => jsonRes({ message: 'db error' }, 500)) + const err = await client.fetchStatus().catch((e: unknown) => e) + expect((err as UpstreamError).kind).toBe('upstream_unavailable') + expect((err as UpstreamError).status).toBe(500) + }) + + test('网络故障 → upstream_unavailable（无状态码）', async () => { + const client = httpWith(async () => { + throw new TypeError('fetch failed') + }) + const err = await client.fetchLogs('sk-x').catch((e: unknown) => e) + expect((err as UpstreamError).kind).toBe('upstream_unavailable') + expect((err as UpstreamError).status).toBeUndefined() + expect((err as UpstreamError).message).toContain('fetch failed') + }) + + test('HTTP 200 但 code/success 为 false → upstream_unavailable（取上游 message）', async () => { + const client = httpWith(async () => jsonRes({ success: false, message: '内部错误', data: null })) + const err = await client.fetchTokenUsage('sk-x').catch((e: unknown) => e) + expect((err as UpstreamError).kind).toBe('upstream_unavailable') + expect((err as UpstreamError).message).toContain('内部错误') + }) + + test('成功：剥掉传输层包装返回 data；请求头带 Bearer', async () => { + let seenAuth: string | undefined + const client = httpWith(async (_url, init) => { + const headers = new Headers(init?.headers) + seenAuth = headers.get('authorization') ?? undefined + return jsonRes({ code: true, message: 'ok', data: { total_used: 42 } }) + }) + const usage = await client.fetchTokenUsage('sk-abc') + expect(seenAuth).toBe('Bearer sk-abc') + expect((usage as unknown as { total_used: number }).total_used).toBe(42) + }) + + test('未配置 BaseURL → upstream_unavailable', async () => { + const client = createHttpClient({ base: '', fetchImpl: async () => jsonRes({}) }) + const err = await client.fetchStatus().catch((e: unknown) => e) + expect((err as UpstreamError).kind).toBe('upstream_unavailable') + }) +}) + +const FIXTURE_DIR = new URL('../../../../', import.meta.url) +const hasSnapshots = existsSync(new URL('api.log.token.json', FIXTURE_DIR)) + +describe('样例实现 · 错误仿真约定', () => { + test('含 reject → 401 key_rejected', async () => { + const client = createFixtureClient({ dir: FIXTURE_DIR }) + const err = await client.fetchTokenUsage('sk-fixture-reject-1').catch((e: unknown) => e) + expect((err as UpstreamError).kind).toBe('key_rejected') + expect((err as UpstreamError).status).toBe(401) + }) + + test('含 unavailable → upstream_unavailable', async () => { + const client = createFixtureClient({ dir: FIXTURE_DIR }) + const err = await client.fetchLogs('sk-fixture-unavailable').catch((e: unknown) => e) + expect((err as UpstreamError).kind).toBe('upstream_unavailable') + }) + + test('不传 dir 时退到仓库根（env.newapiFixtureDir 为空串的兜底）', async () => { + const client = createFixtureClient() + const status = await client.fetchStatus() + expect(status.quota_per_unit).toBe(500000) + }) +}) + +describe.if(hasSnapshots)('样例实现 · 快照读取', () => { + test('fetchLogs 返回 1000 条且 other 保持 JSON 字符串', async () => { + const client = createFixtureClient({ dir: FIXTURE_DIR }) + const logs = await client.fetchLogs('sk-fixture-ok') + expect(logs.length).toBe(1000) + expect(typeof logs[0]!.other).toBe('string') + expect(typeof logs[0]!.request_id).toBe('string') + }) + + test('fetchTokenUsage 按实测形状合成（总量取自快照 other.subscription_*）', async () => { + const client = createFixtureClient({ dir: FIXTURE_DIR }) + const usage = await client.fetchTokenUsage('sk-fixture-ok') + expect(usage.unlimited_quota).toBe(false) + expect(usage.expires_at).toBe(0) + expect(usage.total_granted).toBeGreaterThan(0) + expect(usage.total_available).toBe(usage.total_granted - usage.total_used) + expect(typeof usage.name).toBe('string') + }) + + test('fetchStatus 读取 quota_per_unit', async () => { + const client = createFixtureClient({ dir: FIXTURE_DIR }) + const status = await client.fetchStatus() + expect(status.quota_per_unit).toBe(500000) + }) +}) diff --git a/docs/design/实施计划.md b/docs/design/实施计划.md index 76a4bae..c6a4ae6 100644 --- a/docs/design/实施计划.md +++ b/docs/design/实施计划.md @@ -7,7 +7,7 @@ | 状态 | 定稿（2026-09-14；**进度表随 loop 持续更新**） | | 关联 | `docs/design/API契约.md` · `docs/design/同步引擎设计.md` · `docs/design/交互设计.md` · 四张表字段设计 · `AGENTS.md`（命令） | | 范围 | 阶段划分、每批验收标准、loop 机制与推进规则；技术细节一律以设计文档为准 | -| 当前进度 | 阶段 0 / 1 已完成；下一批：**阶段 2**（上游客户端 + key 管理） | +| 当前进度 | 阶段 0 / 1 / 2 已完成；下一批：**阶段 3**（管理面其余端点：purchases + qpu） | --- @@ -36,6 +36,10 @@ | 8 | 前端提交时间用 `toISOString()`（`Z` 形式） | 契约 约定 9 允许 | | 9 | drizzle-kit 的连库驱动用 `postgres`（postgres.js，纯 JS）——仅 CLI 需要；运行时仍走 Bun 内建 SQL（`drizzle-orm/bun-sql`） | devDependency，可换 `pg` | | 10 | 上游客户端以**接口 + 可注入实现**组织（真实 HTTP / 样例文件两种实现），引擎与路由依赖接口 | 见 §6；可改 | +| 11 | 上游实现的注入开关：env `UPSTREAM_SOURCE`（`http` 默认 / `fixture` 读仓库根快照）；fixture 的错误分支由 key 内容约定触发（含 `reject` → 401、`forbidden` → 403、`unavailable` → 网络故障） | 阶段 2 落地；仅开发期使用，生产保持默认 `http` | +| 12 | 登记 / 启用验证成功即写 `usage_synced_at` + 上游字段全组（该次 `GET /api/usage/token` 本身即一次额度采集）；`log_synced_at` 仍只由日志同步写入 | 契约 §2.2 已定登记响应 `log_synced_at: null`；额度侧取「数据已入库就不撒谎」口径 | +| 13 | 失败分类：401 / 403 → `key_rejected`（上游明确拒绝）；其余非 2xx / 网络故障 / 超时 / 响应结构异常 → `upstream_unavailable` | 上游事实：403 = 账号封禁；可调 | +| 14 | `PATCH` 幂等分支不重复调上游验证（直接返回当前行）；启用时凭证解密失败 → `500 internal`（提示需重新登记） | 与「解密失败不摘除、需人工重新登记」一致；可调 | ## 2. 前提（2026-09-14 定） @@ -49,7 +53,7 @@ |------|------|---------|------| | 0 | 脚手架 + 契约类型 + 501 骨架 | 命令链全绿（type-check / test / build） | ✅ 2026-09-14 | | 1 | **数据层与密钥**：crypto（HKDF 派生 / HMAC-SHA256 / AES-256-GCM）· 时间工具（+08:00 序列化、UTC 秒转换）· 建表 | 单测：哈希确定性 + 加解密往返；`db:push` 后四张表在库 | ✅ 2026-09-14（15 项单测绿；四表在库——`api_keys` 17 列 / `usage_logs` 23 列 / 索引 8 个） | -| 2 | **上游客户端 + key 管理**（`GET` / `POST` / `PATCH` 三端点）· 客户端接口化（§6） | curl 实测：登记 → 清单 → 停用/启用（**注入样例实现**：登记成功路径用样例响应驱动）；三类错误码分支正确（`key_rejected` / `already_registered` / `upstream_unavailable`） | ☐ | +| 2 | **上游客户端 + key 管理**（`GET` / `POST` / `PATCH` 三端点）· 客户端接口化（§6） | curl 实测：登记 → 清单 → 停用/启用（**注入样例实现**：登记成功路径用样例响应驱动）；三类错误码分支正确（`key_rejected` / `already_registered` / `upstream_unavailable`） | ✅ 2026-09-15（28 项单测绿；curl 16 步——登记 201 / 重复 409 / `key_rejected` 400 / `upstream_unavailable` 502 / 停用与幂等 / 启用清 `consecutive_401s` / `not_found` 404 / 非法参数 400；顺带修正「`load-env` 未被服务端接线」） | | 3 | **管理面其余**：purchases 五端点 + qpu 一端点 | curl 实测：CRUD 全流程 + 观测聚合（`matched_purchase_id` 判定正确） | ☐ | | 4 | **同步引擎**：tick（5 分钟 + 防重入 + 启动即跑）· 日志整批事务入库 · 额度采集 · QPU 采样 · 401 状态机（N=3）· **seed 脚本**（§6） | 库中出现数据（seed 或真实采集）；`[sync]` 日志符合引擎设计；401 状态机单测（连续 401 → 摘除且 `disabled_source = auto_401`） | ☐ | | 5 | **查看面数据面**：`logs/query` + `summary` + 折算链（按购买期间分段取尺） | curl 实测：字段齐全、`meta.data_until` 取最慢、金额按购买分段累加；**无归属购买的行 `amount_rmb = null` 且计入 `unpriced_count`**（2026-09-14 定） | ☐ |