feat(server): 上游客户端与 key 管理（阶段 2）

- 上游客户端接口化：UpstreamClient + HTTP / fixture 两实现（env UPSTREAM_SOURCE 切换；fixture 读仓库根样例快照，错误分支由 key 内容约定触发）
- 失败分类：401 / 403 → key_rejected；其余非 2xx / 网络故障 / 超时 / 响应异常 → upstream_unavailable
- /api/manage/api-keys 三端点：清单（按 id 升序，敏感列不出接口）· 登记（上游验证 → 上游字段全组入库）· 启停（幂等、启用需验证并清 consecutive_401s）
- 登记成功触发「立即异步补一轮」hook（引擎骨架占位，阶段 4 填充）
- 统一错误信封：onError → 500 internal · notFound → 404
- 修正：load-env 未被服务端接线（此前只有 drizzle-kit 接了），env.ts 顶部补副作用导入
- 实施计划：进度表阶段 2 ✅ · 实现假设 #11–#14

.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=

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 {

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

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 }

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

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<string, unknown>,
+    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),
+  }
+}

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) =>
+apiKeysRoutes.get('/', async (c) => {
-  c.json(
+  const rows = await getDb().select().from(apiKeys).orderBy(asc(apiKeys.id))
-    {
+  return c.json(rows.map(toApiKeyRow))
-      error: {
+})
-        code: 'internal',
-        message: '未实现：key 清单（见 docs/design/API契约.md §2.2）',
-      },
-    },
-    501,
-  ),
-)
 
 /** 登记：body `{ key }`；上游验证通过 → 入库（不阻塞首轮同步） */
-apiKeysRoutes.post('/', (c) =>
+apiKeysRoutes.post('/', async (c) => {
-  c.json(
+  let body: RegisterApiKeyRequest
-    {
+  try {
-      error: {
+    body = (await c.req.json()) as RegisterApiKeyRequest
-        code: 'internal',
+  } catch {
-        message: '未实现：登记 key（见 docs/design/API契约.md §2.2）',
+    return apiError(c, 400, 'invalid_params', '请求体必须是 JSON')
-      },
+  }
-    },
+  const plainKey = typeof body?.key === 'string' ? body.key.trim() : ''
-    501,
+  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) =>
+apiKeysRoutes.patch('/:id', async (c) => {
-  c.json(
+  const id = Number(c.req.param('id'))
-    {
+  if (!Number.isInteger(id) || id <= 0) {
-      error: {
+    return apiError(c, 400, 'invalid_params', 'id 非法')
-        code: 'internal',
+  }
-        message: '未实现：启停（见 docs/design/API契约.md §2.2）',
+  let body: PatchApiKeyRequest
-      },
+  try {
-    },
+    body = (await c.req.json()) as PatchApiKeyRequest
-    501,
+  } 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!))
+})

apps/server/src/sync/index.ts

@@ -37,9 +37,19 @@ async function tick(): Promise<void> {
   }
 }
 
-/** 启动引擎：立即跑一轮，随后每 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()
+}

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<UpstreamTokenUsage>
+  /** 接口⑨：最近 1000 条调用日志（同步引擎唯一数据源） */
+  fetchLogs(plainKey: string): Promise<UpstreamLogRecord[]>
+  /** 接口④：站点配置（QPU 采样；无需 key） */
+  fetchStatus(): Promise<UpstreamStatus>
+}
+
+let client: UpstreamClient | null = null
+
+/** 进程级单例（实现选择只发生在首次调用） */
+export function getUpstreamClient(): UpstreamClient {
+  if (!client) {
+    client =
+      env.upstreamSource === 'fixture' ? createFixtureClient() : createHttpClient()
+  }
+  return client
+}
+
+export type { UpstreamLogRecord, UpstreamStatus, UpstreamTokenUsage }

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

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<T>(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<UpstreamTokenUsage> {
+      simulate(plainKey)
+      const first = loadLogs()[0]
+      let other: Record<string, unknown> = {}
+      if (first?.other) {
+        try {
+          other = JSON.parse(first.other) as Record<string, unknown>
+        } 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<UpstreamLogRecord[]> {
+      simulate(plainKey)
+      return loadLogs()
+    },
+    async fetchStatus(): Promise<UpstreamStatus> {
+      const snap = readSnapshot<{ data?: Record<string, unknown> }>(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
+    },
+  }
+}

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<Response>
+
+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<string, unknown>
+    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<unknown> {
+    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<UpstreamTokenUsage> {
+      return (await request('/api/usage/token', plainKey)) as UpstreamTokenUsage
+    },
+    async fetchLogs(plainKey: string): Promise<UpstreamLogRecord[]> {
+      return (await request('/api/log/token', plainKey)) as UpstreamLogRecord[]
+    },
+    async fetchStatus(): Promise<UpstreamStatus> {
+      return (await request('/api/status')) as UpstreamStatus
+    },
+  }
+}

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<string, unknown>
+  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
+}

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

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 定） | ☐ |