From e1c6d2153ece1d5c890b4f61868391df657bd759 Mon Sep 17 00:00:00 2001 From: skycurtain Date: Tue, 15 Sep 2026 00:46:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20=E5=90=8C=E6=AD=A5=E5=BC=95?= =?UTF-8?q?=E6=93=8E=E4=B8=8E=20seed=EF=BC=88=E9=98=B6=E6=AE=B5=204?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 引擎：tick（5 分钟 + 防重入 + 启动即跑）· 单轮串行遍历 active key · 日志整批单事务入库（ON CONFLICT DO NOTHING + log_synced_at 取请求发起时刻）· 额度采集（上游字段全组，独立 try）· QPU 采样（基线 / 值变化才追加）· 每轮摘要与超期 warning - 401 状态机：N=3 摘除（disabled_source=auto_401）；403/500/网络不计数；解密失败跳过不计数 - 纯逻辑抽层：policy（计数 / 合并判定）· dedupe（去重键与内容指纹）· log-row（日志映射，与 seed 共用） - seed 脚本（幂等）：假 key + 984 条日志 + QPU 基线 + 示例购买 - 实测：401 端到端 1→2→3 摘除且摘除后不再轮询；幂等重跑新增 0 条 - 实证发现：错误行 request_id 不唯一（8 组 ×3）——记入实施计划「待定问题 #5」与 upstream 文档，待作者裁决 - AGENTS.md / 根 package.json 增 seed 命令 --- AGENTS.md | 3 +- apps/server/package.json | 1 + apps/server/scripts/seed.ts | 127 +++++++++++++ apps/server/src/lib/upstream-columns.ts | 19 ++ apps/server/src/routes/manage/api-keys.ts | 16 +- apps/server/src/sync/dedupe.test.ts | 53 ++++++ apps/server/src/sync/dedupe.ts | 26 +++ apps/server/src/sync/index.ts | 219 ++++++++++++++++++++-- apps/server/src/sync/log-row.ts | 45 +++++ apps/server/src/sync/policy.test.ts | 79 ++++++++ apps/server/src/sync/policy.ts | 60 ++++++ docs/design/实施计划.md | 7 +- docs/upstream/NewAPI-sk-key接口文档.md | 2 +- package.json | 3 +- 14 files changed, 623 insertions(+), 37 deletions(-) create mode 100644 apps/server/scripts/seed.ts create mode 100644 apps/server/src/lib/upstream-columns.ts create mode 100644 apps/server/src/sync/dedupe.test.ts create mode 100644 apps/server/src/sync/dedupe.ts create mode 100644 apps/server/src/sync/log-row.ts create mode 100644 apps/server/src/sync/policy.test.ts create mode 100644 apps/server/src/sync/policy.ts diff --git a/AGENTS.md b/AGENTS.md index 2a15cfb..e6d8042 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,8 +35,9 @@ bun run type-check # 类型检查（vue-tsc + tsc） bun run db:generate # 生成 Drizzle 迁移（apps/server/drizzle） bun run db:migrate # 执行迁移 bun run db:push # 直接推 schema（开发期） +bun run seed # 灌入开发数据（样例快照 → 四张表；幂等，仅本地） ``` -- 环境变量：复制 `.env.example` 为 `.env` 并填值（`DATABASE_URL` / `MANAGE_TOKEN` / `KEYS_MASTER_KEY` / `NEWAPI_BASE`） +- 环境变量：复制 `.env.example` 为 `.env` 并填值（`DATABASE_URL` / `MANAGE_TOKEN` / `KEYS_MASTER_KEY` / `NEWAPI_BASE`；离线开发可加 `UPSTREAM_SOURCE=fixture` 走样例快照，见 `docs/design/实施计划.md` §6） - 目录速记：前端 `apps/web/src/{api,router,views}` · 后端 `apps/server/src/{db,routes,sync,lib}` · 契约类型在 `packages/shared` - 开发期前端 `/api` 由 Vite 代理到 :3000（`apps/web/vite.config.ts`）；生产由同域反代承担（见 README 工程注记） diff --git a/apps/server/package.json b/apps/server/package.json index 85b24eb..0fe2cb9 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -6,6 +6,7 @@ "dev": "bun run --hot src/index.ts", "start": "bun run src/index.ts", "test": "bun test", + "seed": "bun run scripts/seed.ts", "db:generate": "drizzle-kit generate", "db:migrate": "drizzle-kit migrate", "db:push": "drizzle-kit push" diff --git a/apps/server/scripts/seed.ts b/apps/server/scripts/seed.ts new file mode 100644 index 0000000..ff8641c --- /dev/null +++ b/apps/server/scripts/seed.ts @@ -0,0 +1,127 @@ +/** + * seed 脚本（实施计划 §6）：从仓库根的实测响应快照灌入开发数据——**幂等**，可重复执行。 + * + * 用法：在 `apps/server` 下 `bun run seed`（环境变量自动经 `src/lib/env.ts` 加载仓库根 `.env`）。 + * + * 灌入内容： + * 1. 一个占位 `api_keys` 行（假 key 常量 `SEED_KEY`，哈希 / 密文走真实 crypto 逻辑—— + * 查询面可用该常量 key 命中已登记 key） + * 2. `usage_logs`：`api.log.token.json` 的 1000 条样例日志（`ON CONFLICT DO NOTHING`） + * 3. `qpu_history` 基线：`api.status.json` 的 `quota_per_unit`（已有数据则跳过） + * 4. 一条示例 `purchases`（订阅路径；**金额为占位值**，用于验证折算链） + * + * 注：快照文件未入库（`.gitignore` 的 `/api*.json`），本脚本只在本地开发环境使用。 + */ +import { readFileSync } from 'node:fs' +import { eq } from 'drizzle-orm' + +import { DEFAULT_QUOTA_PER_UNIT } from '@naku/shared' + +import { getDb } from '../src/db/client' +import { apiKeys, purchases, qpuHistory, usageLogs } from '../src/db/schema' +import { getKeySecrets } from '../src/lib/keys' +import { fromUnixSeconds } from '../src/lib/time' +import { parseOther, toLogRow } from '../src/sync/log-row' +import type { UpstreamLogRecord } from '../src/upstream/types' + +/** 假 key 常量：查看面 / 管理面用它做本地验证（可用 env SEED_KEY 覆盖） */ +const SEED_KEY = process.env.SEED_KEY ?? 'sk-seed-demo-key-0001' + +/** 仓库根（本文件位于 apps/server/scripts/） */ +const REPO_ROOT = new URL('../../../', import.meta.url) + +function readSnapshot(name: string): T { + return JSON.parse(readFileSync(new URL(name, REPO_ROOT), 'utf8')) as T +} + +function asNumber(v: unknown, fallback: number): number { + return typeof v === 'number' && Number.isFinite(v) ? v : fallback +} + +const db = getDb() +const secrets = getKeySecrets() + +// ---------- 样例快照 ---------- +const logs = readSnapshot<{ data?: UpstreamLogRecord[] }>('api.log.token.json').data +if (!Array.isArray(logs) || logs.length === 0) { + console.error('[seed] api.log.token.json 缺失或结构异常（缺 data[]）——请确认样例文件在仓库根') + process.exit(1) +} +const status = readSnapshot<{ data?: Record }>('api.status.json').data ?? {} + +// ---------- 1. api_keys ---------- +const keyHash = secrets.keyHash(SEED_KEY) +let [keyRow] = await db.select().from(apiKeys).where(eq(apiKeys.keyHash, keyHash)).limit(1) +if (!keyRow) { + const first = logs[0]! + const other = parseOther(first.other) + const totalGranted = asNumber(other.subscription_total, 800_000_000) + const totalUsed = asNumber(other.subscription_used, 85_579_458) + const now = new Date() + ;[keyRow] = await db + .insert(apiKeys) + .values({ + keyHash, + keyCiphertext: secrets.encryptKey(SEED_KEY), + status: 'active', + disabledSource: null, + consecutive401s: 0, + logSyncedAt: now, + usageSyncedAt: now, + tokenName: first.token_name ?? 'seed-token', + totalGranted, + totalUsed, + totalAvailable: totalGranted - totalUsed, + unlimitedQuota: false, + modelLimits: {}, + modelLimitsEnabled: false, + expiresAt: null, + }) + .returning() + console.log(`[seed] api_keys：已插入 id=${keyRow!.id}（token_name=${keyRow!.tokenName}）`) +} else { + console.log(`[seed] api_keys：已存在（id=${keyRow.id}），跳过`) +} + +// ---------- 2. usage_logs ---------- +const rows = logs.map((log) => toLogRow(keyRow!.id, log)) +const inserted = await db + .insert(usageLogs) + .values(rows) + .onConflictDoNothing({ target: usageLogs.dedupeKey }) + .returning({ id: usageLogs.id }) +console.log(`[seed] usage_logs：新增 ${inserted.length} 条（样例共 ${logs.length} 条）`) + +// ---------- 3. qpu_history 基线 ---------- +const [qpuRow] = await db.select().from(qpuHistory).limit(1) +if (!qpuRow) { + const qpu = asNumber(status.quota_per_unit, DEFAULT_QUOTA_PER_UNIT) + await db.insert(qpuHistory).values({ qpu, effectiveFrom: new Date() }) + console.log(`[seed] qpu_history：已插入基线 qpu=${qpu}`) +} else { + console.log(`[seed] qpu_history：已有数据（qpu=${qpuRow.qpu}），跳过`) +} + +// ---------- 4. 示例 purchases（金额为占位值） ---------- +const [purchaseRow] = await db.select().from(purchases).limit(1) +if (!purchaseRow) { + const first = logs[0]! + const other = parseOther(first.other) + const earliestSeconds = Math.min(...logs.map((l) => l.created_at)) + const earliest = fromUnixSeconds(earliestSeconds) + await db.insert(purchases).values({ + kind: 'subscription', + paidAt: earliest, + paidAmount: 168.5, // 占位值：实付金额只有作者知道 + note: 'seed 示例（金额为占位值，非真实）', + periodStart: earliest, + periodEnd: null, + subscriptionId: asNumber(other.subscription_id, 90), + quotaGranted: asNumber(other.subscription_total, 800_000_000), + }) + console.log('[seed] purchases：已插入示例一条（折算链验证用）') +} else { + console.log(`[seed] purchases：已有数据（id=${purchaseRow.id}），跳过`) +} + +console.log(`[seed] 完成。查询面可用假 key「${SEED_KEY}」命中已登记 key。`) diff --git a/apps/server/src/lib/upstream-columns.ts b/apps/server/src/lib/upstream-columns.ts new file mode 100644 index 0000000..fef9b6d --- /dev/null +++ b/apps/server/src/lib/upstream-columns.ts @@ -0,0 +1,19 @@ +/** + * 上游字段全组（`api_keys` 表 8 个上游列）——登记 / 启用 / 每轮额度采集共用一处， + * 避免三处各写一份导致漏列（api_keys 表设计 §3.4/§3.5：整组一起写）。 + */ +import type { UpstreamTokenUsage } from '../upstream/types' +import { expiresAtFromUnix } from './time' + +export 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), + } +} diff --git a/apps/server/src/routes/manage/api-keys.ts b/apps/server/src/routes/manage/api-keys.ts index f10ac62..48ec2bb 100644 --- a/apps/server/src/routes/manage/api-keys.ts +++ b/apps/server/src/routes/manage/api-keys.ts @@ -5,7 +5,7 @@ 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 { upstreamColumns } from '../../lib/upstream-columns' import { manageAuth } from '../../middleware/manage-auth' import { getDb } from '../../db/client' import { apiKeys } from '../../db/schema' @@ -21,20 +21,6 @@ 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('/', async (c) => { const rows = await getDb().select().from(apiKeys).orderBy(asc(apiKeys.id)) diff --git a/apps/server/src/sync/dedupe.test.ts b/apps/server/src/sync/dedupe.test.ts new file mode 100644 index 0000000..19f9c7b --- /dev/null +++ b/apps/server/src/sync/dedupe.test.ts @@ -0,0 +1,53 @@ +/** + * 去重键单测（usage_logs 表设计 §3.1 + 实施计划「实现假设」#5）。 + */ +import { describe, expect, test } from 'bun:test' + +import { contentFingerprint, dedupeKeyFor } from './dedupe' +import type { UpstreamLogRecord } from '../upstream/types' + +const log = (over: Partial = {}): UpstreamLogRecord => ({ + id: 1, + user_id: 5, + created_at: 1789126642, + type: 2, + content: '', + username: 'chendong', + token_name: 'yangsiyue', + model_name: 'glm-5.3-flash', + quota: 702, + prompt_tokens: 29157, + completion_tokens: 1036, + use_time: 30, + is_stream: true, + channel: 7, + channel_name: '', + token_id: 29, + group: 'default', + ip: '', + request_id: '202609111136522416558528268d9d6CRxC30kh', + other: '{}', + ...over, +}) + +describe('dedupeKeyFor', () => { + test('request_id 存在 → `${key_id}:${request_id}`', () => { + expect(dedupeKeyFor(3, log())).toBe('3:202609111136522416558528268d9d6CRxC30kh') + }) + + test('跨 key 隔离：同一 request_id 在不同 key 下不同键', () => { + expect(dedupeKeyFor(3, log())).not.toBe(dedupeKeyFor(4, log())) + }) + + test('request_id 为空 → 内容指纹（同输入稳定）、前缀区分 `fp:`', () => { + const a = dedupeKeyFor(3, log({ request_id: '' })) + const b = dedupeKeyFor(3, log({ request_id: '' })) + expect(a).toBe(b) + expect(a.startsWith('fp:')).toBe(true) + }) + + test('指纹随内容变化（真实不同请求不同键）', () => { + expect(contentFingerprint(3, log())).not.toBe(contentFingerprint(3, log({ quota: 703 }))) + expect(contentFingerprint(3, log())).not.toBe(contentFingerprint(4, log())) + }) +}) diff --git a/apps/server/src/sync/dedupe.ts b/apps/server/src/sync/dedupe.ts new file mode 100644 index 0000000..810e9e6 --- /dev/null +++ b/apps/server/src/sync/dedupe.ts @@ -0,0 +1,26 @@ +/** + * 去重键生成（usage_logs 表设计 §3.1 + 实施计划「实现假设」#5）： + * `dedupe_key = ${key_id}:${request_id}`；`request_id` 为空时退化为内容指纹 + * `sha256(key_id｜created_at｜model_name｜quota｜prompt_tokens｜completion_tokens)`。 + */ +import { createHash } from 'node:crypto' + +import type { UpstreamLogRecord } from '../upstream/types' + +/** 内容指纹（仅 `request_id` 缺失时启用） */ +export function contentFingerprint(keyId: number, log: UpstreamLogRecord): string { + const parts = [ + keyId, + log.created_at, + log.model_name ?? '', + log.quota, + log.prompt_tokens, + log.completion_tokens, + ].join('|') + return createHash('sha256').update(parts, 'utf8').digest('hex') +} + +export function dedupeKeyFor(keyId: number, log: UpstreamLogRecord): string { + const requestId = (log.request_id ?? '').trim() + return requestId ? `${keyId}:${requestId}` : `fp:${contentFingerprint(keyId, log)}` +} diff --git a/apps/server/src/sync/index.ts b/apps/server/src/sync/index.ts index a8eb408..9713320 100644 --- a/apps/server/src/sync/index.ts +++ b/apps/server/src/sync/index.ts @@ -1,24 +1,206 @@ /** - * 同步引擎（见 `docs/design/同步引擎设计.md`）——骨架占位。 + * 同步引擎（见 `docs/design/同步引擎设计.md`）：调度 + 单轮执行。 * - * 已定骨架（§3.2 / §3.3 / §3.4）： - * - 后端进程内定时器（不引 worker、不拆进程） - * - 周期 5 分钟，固定 tick + 防重入 - * - 启动立即跑一轮 - * - 单轮串行遍历全部 active key - * - 每轮：日志（整批单事务入库 → log_synced_at 取请求发起时刻）· 额度（上游字段全组 + usage_synced_at）· QPU 采样（每轮一次、值变化才追加） - * - 失败：401 计数（N=3 摘除）· 403/500/网络不计数 · 不单轮重试 + * 结构（§3.2 / §3.3 / §3.4）： + * - 后端进程内定时器（不引 worker、不拆进程）；周期 5 分钟，固定 tick + 防重入；启动立即跑一轮 + * - 单轮串行遍历全部 `active` key；每 key 依次：日志（整批单事务入库 → `log_synced_at` 取请求发起时刻） + * → 额度（上游字段全组 + `usage_synced_at`，与日志各自独立 try） + * - 每轮一次 QPU 采样（独立于 key 循环；`qpu_history` 为空时插基线） + * - 失败：401 计数（N=3 摘除，计数只认 401）· 403/500/网络不计数 · 不单轮重试 + * - 可观测：`[sync]` 前缀一行式日志；每轮摘要；`log_synced_at` 落后 > 24h 每轮 warning */ +import { asc, desc, eq } from 'drizzle-orm' -const TICK_MS = 5 * 60 * 1000 +import { getDb } from '../db/client' +import { apiKeys, qpuHistory, usageLogs } from '../db/schema' +import { getKeySecrets } from '../lib/keys' +import { upstreamColumns } from '../lib/upstream-columns' +import { getUpstreamClient } from '../upstream/client' +import { errorMessage, UpstreamError } from '../upstream/errors' +import type { UpstreamLogRecord } from '../upstream/types' +import { toLogRow } from './log-row' +import { + combineRoundResult, + nextKeyCounter, + STALE_WARN_MS, + TICK_MS, + type CallResult, +} from './policy' + +type KeyRow = typeof apiKeys.$inferSelect + +interface RoundSummary { + keyCount: number + okCount: number + failCount: number + inserted: number +} let running = false -/** 一轮：遍历全部 active key（串行）→ 日志 / 额度；每轮一次 QPU 采样 */ -async function runOnce(): Promise { - // TODO(#5): 日志同步 —— GET /api/log/token → 整批单事务入库（ON CONFLICT (dedupe_key) DO NOTHING）→ 写 log_synced_at - // TODO(#6): 额度采集 —— GET /api/usage/token → 上游字段全组 + usage_synced_at（与日志各自独立） - // TODO(#7): QPU 采样 —— GET /api/status → 值变化才追加 qpu_history（每轮一次，独立于 key 循环） +/** 上游错误 → 本轮结果：只有 401 是「key 失效」信号（403 = 账号封禁，不计数） */ +function classify(err: unknown): CallResult { + if (err instanceof UpstreamError && err.kind === 'key_rejected' && err.status === 401) { + return 'unauthorized' + } + return 'failed' +} + +/** + * 日志整批单事务入库（`ON CONFLICT (dedupe_key) DO NOTHING`）+ 写 `log_synced_at` + * ——全成或全不写：失败则「数据截止」不前移，下一轮幂等重来（§3.3）。 + */ +async function persistLogs( + keyId: number, + logs: UpstreamLogRecord[], + syncedAt: Date, +): Promise { + const db = getDb() + const rows = logs.map((log) => toLogRow(keyId, log)) + return await db.transaction(async (tx) => { + let inserted = 0 + if (rows.length > 0) { + const res = await tx + .insert(usageLogs) + .values(rows) + .onConflictDoNothing({ target: usageLogs.dedupeKey }) + .returning({ id: usageLogs.id }) + inserted = res.length + } + await tx.update(apiKeys).set({ logSyncedAt: syncedAt }).where(eq(apiKeys.id, keyId)) + return inserted + }) +} + +/** 单 key 一轮：日志 → 额度（各自独立 try）→ 状态机（每轮一次） */ +async function syncOneKey(key: KeyRow): Promise<{ round: CallResult; inserted: number }> { + const db = getDb() + let plainKey: string + try { + plainKey = getKeySecrets().decryptKey(Buffer.from(key.keyCiphertext)) + } catch (err) { + // 解密失败：记日志、跳过、不影响其他 key；不计数、不摘除（§3.4） + console.error( + `[sync] key_id=${key.id} 凭证解密失败，跳过本轮（需人工重新登记）：${errorMessage(err)}`, + ) + return { round: 'failed', inserted: 0 } + } + + const results: CallResult[] = [] + let inserted = 0 + + // —— 日志同步（独立 try）—— + const logRequestedAt = new Date() + const logStartedMs = Date.now() + try { + const logs = await getUpstreamClient().fetchLogs(plainKey) + inserted = await persistLogs(key.id, logs, logRequestedAt) + results.push('ok') + console.log( + `[sync] key_id=${key.id} 日志同步：上游 ${logs.length} 条，新增 ${inserted} 条（${Date.now() - logStartedMs}ms）`, + ) + } catch (err) { + results.push(classify(err)) + console.error( + `[sync] key_id=${key.id} 日志同步失败（${Date.now() - logStartedMs}ms）：${errorMessage(err)}`, + ) + } + + // —— 额度采集（独立 try；与日志互不影响、各自写各自的时间戳）—— + const usageRequestedAt = new Date() + const usageStartedMs = Date.now() + try { + const usage = await getUpstreamClient().fetchTokenUsage(plainKey) + await db + .update(apiKeys) + .set({ usageSyncedAt: usageRequestedAt, ...upstreamColumns(usage) }) + .where(eq(apiKeys.id, key.id)) + results.push('ok') + } catch (err) { + results.push(classify(err)) + console.error( + `[sync] key_id=${key.id} 额度采集失败（${Date.now() - usageStartedMs}ms）：${errorMessage(err)}`, + ) + } + + // —— 401 状态机（每轮一次；401 优先于成功——见实施计划「实现假设」#15）—— + const round = combineRoundResult(results) + const next = nextKeyCounter( + { + consecutive401s: key.consecutive401s, + status: key.status as 'active' | 'disabled', + disabledSource: key.disabledSource as 'manual' | 'auto_401' | null, + }, + round, + ) + const patch: Partial = {} + if (next.consecutive401s !== key.consecutive401s) patch.consecutive401s = next.consecutive401s + if (next.status !== key.status) patch.status = next.status + if (next.disabledSource !== key.disabledSource) patch.disabledSource = next.disabledSource + if (Object.keys(patch).length > 0) { + await db.update(apiKeys).set(patch).where(eq(apiKeys.id, key.id)) + } + if (next.removed) { + console.warn( + `[sync] key_id=${key.id} 连续 ${next.consecutive401s} 次 401 → 自动摘除（disabled_source=auto_401）`, + ) + } + + return { round, inserted } +} + +/** QPU 采样：每轮一次；`qpu_history` 为空时插基线，值变化才追加（§3.3 / §3.6） */ +async function sampleQpu(): Promise { + const db = getDb() + try { + const status = await getUpstreamClient().fetchStatus() + const qpu = status.quota_per_unit + const [last] = await db + .select() + .from(qpuHistory) + .orderBy(desc(qpuHistory.effectiveFrom)) + .limit(1) + const effectiveFrom = new Date() + if (!last) { + await db.insert(qpuHistory).values({ qpu, effectiveFrom }) + console.log(`[sync] QPU 基线已插入：${qpu}`) + } else if (Number(last.qpu) !== qpu) { + await db.insert(qpuHistory).values({ qpu, effectiveFrom }) + console.log(`[sync] QPU 变化：${last.qpu} → ${qpu}`) + } + } catch (err) { + console.error(`[sync] QPU 采样失败：${errorMessage(err)}`) + } +} + +/** 一轮：遍历全部 active key（串行）→ 日志 / 额度；随后 QPU 采样（无 key 也跑） */ +async function runOnce(): Promise { + const db = getDb() + const keys = await db + .select() + .from(apiKeys) + .where(eq(apiKeys.status, 'active')) + .orderBy(asc(apiKeys.id)) + + const summary: RoundSummary = { keyCount: keys.length, okCount: 0, failCount: 0, inserted: 0 } + const nowMs = Date.now() + + for (const key of keys) { + // 「3 天窗口」警示：落后 > 24h 每轮 warning（§3.5） + if (key.logSyncedAt && nowMs - key.logSyncedAt.getTime() > STALE_WARN_MS) { + const hours = Math.floor((nowMs - key.logSyncedAt.getTime()) / 3_600_000) + console.warn( + `[sync] key_id=${key.id} 数据截止已落后 ${hours} 小时（>24h；窗口上限约 3 天，请尽快恢复同步）`, + ) + } + const { round, inserted } = await syncOneKey(key) + summary.inserted += inserted + if (round === 'ok') summary.okCount++ + else summary.failCount++ + } + + await sampleQpu() + return summary } async function tick(): Promise { @@ -27,11 +209,14 @@ async function tick(): Promise { return } running = true + const startedMs = Date.now() try { - await runOnce() - console.log('[sync] 轮次完成（骨架占位：暂无实际采集动作）') + const s = await runOnce() + console.log( + `[sync] 轮次完成：${s.keyCount} key，成功 ${s.okCount}，失败 ${s.failCount}，新增日志 ${s.inserted} 条，耗时 ${((Date.now() - startedMs) / 1000).toFixed(1)} 秒`, + ) } catch (err) { - console.error('[sync] 轮次异常', err) + console.error(`[sync] 轮次异常：${errorMessage(err)}`) } finally { running = false } diff --git a/apps/server/src/sync/log-row.ts b/apps/server/src/sync/log-row.ts new file mode 100644 index 0000000..a80c69d --- /dev/null +++ b/apps/server/src/sync/log-row.ts @@ -0,0 +1,45 @@ +/** + * 上游日志行 → `usage_logs` 列值（同步引擎与 seed 脚本共用，避免两份映射漂移）。 + */ +import { fromUnixSeconds } from '../lib/time' +import type { usageLogs } from '../db/schema' +import type { UpstreamLogRecord } from '../upstream/types' +import { dedupeKeyFor } from './dedupe' + +/** 上游 `other`（JSON 字符串）→ jsonb；解析失败退化为空对象（保真优先，不丢整条日志） */ +export function parseOther(raw: string | undefined | null): Record { + if (!raw) return {} + try { + const v = JSON.parse(raw) as unknown + return v && typeof v === 'object' && !Array.isArray(v) ? (v as Record) : {} + } catch { + return {} + } +} + +export function toLogRow(keyId: number, log: UpstreamLogRecord): typeof usageLogs.$inferInsert { + return { + keyId, + dedupeKey: dedupeKeyFor(keyId, log), + userId: log.user_id, + createdAt: fromUnixSeconds(log.created_at), + type: log.type, + content: log.content ?? '', + username: log.username ?? '', + tokenName: log.token_name ?? '', + modelName: log.model_name ?? '', + quota: log.quota, + promptTokens: log.prompt_tokens, + completionTokens: log.completion_tokens, + useTime: log.use_time, + isStream: log.is_stream, + channel: log.channel, + channelName: log.channel_name ?? '', + tokenId: log.token_id, + groupName: log.group ?? 'default', + ip: log.ip ?? '', + requestId: log.request_id ?? '', + upstreamRequestId: log.upstream_request_id ?? null, + other: parseOther(log.other), + } +} diff --git a/apps/server/src/sync/policy.test.ts b/apps/server/src/sync/policy.test.ts new file mode 100644 index 0000000..a381938 --- /dev/null +++ b/apps/server/src/sync/policy.test.ts @@ -0,0 +1,79 @@ +/** + * 401 状态机与轮次结果合并的单测（同步引擎设计 §3.4 + 实施计划「实现假设」#15）。 + */ +import { describe, expect, test } from 'bun:test' + +import { + combineRoundResult, + MAX_CONSECUTIVE_401S, + nextKeyCounter, + type KeyCounterNext, +} from './policy' + +const active = (consecutive401s = 0): KeyCounterNext => ({ + consecutive401s, + status: 'active', + disabledSource: null, + removed: false, +}) + +describe('combineRoundResult（两个调用合并为「本轮结果」）', () => { + test('任一 401 → unauthorized（401 优先于成功）', () => { + expect(combineRoundResult(['ok', 'unauthorized'])).toBe('unauthorized') + expect(combineRoundResult(['unauthorized', 'failed'])).toBe('unauthorized') + }) + + test('无 401 且任一成功 → ok', () => { + expect(combineRoundResult(['ok', 'failed'])).toBe('ok') + expect(combineRoundResult(['ok', 'ok'])).toBe('ok') + }) + + test('全失败 → failed（不动计数）', () => { + expect(combineRoundResult(['failed', 'failed'])).toBe('failed') + }) +}) + +describe('nextKeyCounter（401 计数 / 摘除 / 清零）', () => { + test('成功 → 计数清零（不摘除）', () => { + const next = nextKeyCounter(active(2), 'ok') + expect(next.consecutive401s).toBe(0) + expect(next.status).toBe('active') + expect(next.removed).toBe(false) + }) + + test('其他失败（403/500/网络）→ 计数不动', () => { + const next = nextKeyCounter(active(1), 'failed') + expect(next.consecutive401s).toBe(1) + expect(next.removed).toBe(false) + }) + + test('连续 401 累加；未达阈值不摘除', () => { + let state = active(0) + for (let i = 1; i < MAX_CONSECUTIVE_401S; i++) { + const next = nextKeyCounter(state, 'unauthorized') + expect(next.consecutive401s).toBe(i) + expect(next.removed).toBe(false) + expect(next.status).toBe('active') + state = next + } + }) + + test('第 N 次（N=3）连续 401 → 摘除且 disabled_source = auto_401', () => { + let state: KeyCounterNext = active(0) + for (let i = 1; i <= MAX_CONSECUTIVE_401S; i++) { + state = nextKeyCounter(state, 'unauthorized') + } + expect(state.consecutive401s).toBe(MAX_CONSECUTIVE_401S) + expect(state.status).toBe('disabled') + expect(state.disabledSource).toBe('auto_401') + expect(state.removed).toBe(true) + }) + + test('中途成功会打断连续性（计数回零）', () => { + let state: KeyCounterNext = active(2) + state = nextKeyCounter(state, 'ok') + state = nextKeyCounter(state, 'unauthorized') + expect(state.consecutive401s).toBe(1) + expect(state.removed).toBe(false) + }) +}) diff --git a/apps/server/src/sync/policy.ts b/apps/server/src/sync/policy.ts new file mode 100644 index 0000000..a9b145e --- /dev/null +++ b/apps/server/src/sync/policy.ts @@ -0,0 +1,60 @@ +/** + * 同步策略常量与纯决策函数（同步引擎设计 §3.2/§3.4/§3.5）。 + * 抽成纯函数是为了把「401 计数 / 摘除」与「本轮结果合并」的规则钉在单测里。 + */ + +/** 一轮的固定周期：5 分钟（§3.2） */ +export const TICK_MS = 5 * 60 * 1000 + +/** 连续 401 摘除阈值 N = 3（§3.4） */ +export const MAX_CONSECUTIVE_401S = 3 + +/** 落后警示阈值：`log_synced_at` 落后超过 24 小时 → 每轮 warning（§3.5，距 3 天窗口留 2 天余量） */ +export const STALE_WARN_MS = 24 * 60 * 60 * 1000 + +/** 单个上游调用在本轮的结果 */ +export type CallResult = 'ok' | 'unauthorized' | 'failed' + +/** + * 一轮内两个调用（日志 / 额度）的合并判定（实施计划「实现假设」#15）： + * 任一 401 → `unauthorized`（key 失效信号优先）；否则任一成功 → `ok`；全失败 → `failed`（不动计数）。 + */ +export function combineRoundResult(results: CallResult[]): CallResult { + if (results.includes('unauthorized')) return 'unauthorized' + if (results.includes('ok')) return 'ok' + return 'failed' +} + +export interface KeyCounterState { + consecutive401s: number + status: 'active' | 'disabled' + disabledSource: 'manual' | 'auto_401' | null +} + +export interface KeyCounterNext extends KeyCounterState { + /** 本轮是否触发自动摘除 */ + removed: boolean +} + +/** + * 401 状态机（§3.4）：401 → 计数 +1（达 N 摘除，`disabled_source = auto_401`）； + * 成功 → 计数清零；其他失败（403 / 500 / 网络 / 解密失败）→ 计数不动。 + */ +export function nextKeyCounter(state: KeyCounterState, round: CallResult): KeyCounterNext { + if (round === 'ok') { + return { ...state, consecutive401s: 0, removed: false } + } + if (round === 'failed') { + return { ...state, removed: false } + } + const consecutive401s = state.consecutive401s + 1 + if (consecutive401s >= MAX_CONSECUTIVE_401S) { + return { + consecutive401s, + status: 'disabled', + disabledSource: 'auto_401', + removed: true, + } + } + return { ...state, consecutive401s, removed: false } +} diff --git a/docs/design/实施计划.md b/docs/design/实施计划.md index b0b80a1..b441e07 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–3 已完成；下一批：**阶段 4**（同步引擎：tick / 日志入库 / 额度 / QPU / 401 状态机 / seed） | +| 当前进度 | 阶段 0–4 已完成；下一批：**阶段 5**（查看面数据面：`logs/query` + `summary` + 折算链） | --- @@ -40,6 +40,8 @@ | 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`（提示需重新登记） | 与「解密失败不摘除、需人工重新登记」一致；可调 | +| 15 | 一轮内「日志 + 额度」两个调用的合并判定：**任一 401 → 本轮计 1 次 401**（401 优先）；否则任一成功 → 计数清零；全失败 → 计数不动 | 设计（§3.4）未覆盖「一成功一 401」的冲突情形；401 优先 + N=3 阈值使单次误报无害 | +| 16 | seed 脚本用假 key 常量 `sk-seed-demo-key-0001`（`SEED_KEY` 可覆盖）；`apps/server/scripts/seed.ts` 幂等可重跑 | 实施计划 §6 已定「假 key 常量」；仅开发期 | ## 2. 前提（2026-09-14 定） @@ -55,7 +57,7 @@ | 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`） | ✅ 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` 判定正确） | ✅ 2026-09-15（33 项单测绿；curl 16 步——登记 201 / 缺必填与非法值与期间顺序 400 / 部分更新与显式清空 / 排序 / 404 / 物理删除 / 观测聚合 `matched_purchase_id` 判定正确） | -| 4 | **同步引擎**：tick（5 分钟 + 防重入 + 启动即跑）· 日志整批事务入库 · 额度采集 · QPU 采样 · 401 状态机（N=3）· **seed 脚本**（§6） | 库中出现数据（seed 或真实采集）；`[sync]` 日志符合引擎设计；401 状态机单测（连续 401 → 摘除且 `disabled_source = auto_401`） | ☐ | +| 4 | **同步引擎**：tick（5 分钟 + 防重入 + 启动即跑）· 日志整批事务入库 · 额度采集 · QPU 采样 · 401 状态机（N=3）· **seed 脚本**（§6） | 库中出现数据（seed 或真实采集）；`[sync]` 日志符合引擎设计；401 状态机单测（连续 401 → 摘除且 `disabled_source = auto_401`） | ✅ 2026-09-15（45 项单测绿；seed 灌入 984 条日志 + QPU 基线 + 示例购买；引擎实测——幂等重跑新增 0 条、摘要日志符合设计、**401 端到端：1→2→3 次摘除且摘除后不再轮询**） | | 5 | **查看面数据面**：`logs/query` + `summary` + 折算链（按购买期间分段取尺） | curl 实测：字段齐全、`meta.data_until` 取最慢、金额按购买分段累加；**无归属购买的行 `amount_rmb = null` 且计入 `unpriced_count`**（2026-09-14 定） | ☐ | | 6 | **前端 `/usage-log`**：筛选区（key 标签 + datetimerange + 快捷项）· 卡片 · 表格 + 展开行 · 分页器 · 六种状态模式 | 浏览器实测：seed 数据走一遍；六态逐一可见（加载 / 初始 / 未收录 / 无记录 / 同步中 / 失败）；金额为 `null` 显示「—」 | ☐ | | 7 | **前端管理面**：口令门帘 + 侧边栏 + 三页（key 管理 / 购买记录含核对提示与补录 / 换算参数） | 浏览器实测：登记、启停、CRUD、补录、时间线全操作 | ☐ | @@ -69,6 +71,7 @@ | 2 | 充值（`wallet`）路径的折算近似（设计文档标「后议」） | 先按「未折算」处理（`null` / 计入 `unpriced_count`）；池子加权平均待后议 | | 3 | `dedupe_key` 内容指纹的拼接细节 | **已决**：见「实现假设」#5（表设计 §3.1 已给字段清单） | | 4 | 上游对同账号请求的态度（限流 / 并发被拒时的表现） | 遇错即按「其他故障」处理（不计数、不摘除），如实记录日志 | +| 5 | **`request_id` 在错误行上不唯一**（2026-09-15 实测样板：8 个 `request_id` 各对应 3 条 `type=5` 错误记录，共 24 行）——当前按表设计原样实现 `dedupe_key = key_id:request_id`，同 `request_id` 的多条错误行**合并为 1 条** | **待作者裁决**：影响 = 失败请求次数（`error_count`）少算；候选方案 = 错误行去重键后缀内容指纹 / 序号（`usage_logs` 表设计 §3.1 需相应修订） | ## 5. 风险 diff --git a/docs/upstream/NewAPI-sk-key接口文档.md b/docs/upstream/NewAPI-sk-key接口文档.md index dfcee67..dafbeb5 100644 --- a/docs/upstream/NewAPI-sk-key接口文档.md +++ b/docs/upstream/NewAPI-sk-key接口文档.md @@ -196,7 +196,7 @@ Authorization: Bearer sk-xxxx | `token_id` | int | 上游 token ID（对账用） | | `group` | string | 分组名（默认 `default`） | | `ip` | string | 请求 IP（普通用户视角为空串） | -| `request_id` | string | 请求 ID（格式：`yyyyMMddHHmmss` 前缀 + 随机串）；JSON tag 为 `omitempty`，但源码写入时有兜底保证非空——**去重唯一依据** | +| `request_id` | string | 请求 ID（格式：`yyyyMMddHHmmss` 前缀 + 随机串）；JSON tag 为 `omitempty`，但源码写入时有兜底保证非空——**去重唯一依据**。**补充实测（2026-09-15）**：错误场景下 `request_id` **不保证行级唯一**——1000 条实测样本中有 8 个 `request_id` 各对应 3 条 `type=5` 错误记录（同一请求的多次失败 / 重试，`content` 的 `status_code` 可不同）；按 `request_id` 去重会把它们合并为 1 条（消费类 `type=2` 样本中未观察到重复） | | `upstream_request_id` | string | 上游请求 ID；JSON tag 为 `omitempty`——**空值时不出现、有值时才出现**（实测样本中缺席）；消费方需按「可选字段」处理 | | `other` | string | JSON 字符串，扩展信息（结构见下） | diff --git a/package.json b/package.json index cd2e24d..3e6ac58 100644 --- a/package.json +++ b/package.json @@ -17,7 +17,8 @@ "type-check": "bun run --filter '@naku/web' type-check && bunx tsc --noEmit -p apps/server", "db:generate": "bun run --filter '@naku/server' db:generate", "db:migrate": "bun run --filter '@naku/server' db:migrate", - "db:push": "bun run --filter '@naku/server' db:push" + "db:push": "bun run --filter '@naku/server' db:push", + "seed": "bun run --filter '@naku/server' seed" }, "devDependencies": { "typescript": "^6.0.0"