From d47cb5543c3d9cbd9813862eadbd9599d90e57da Mon Sep 17 00:00:00 2001 From: skycurtain Date: Wed, 7 Oct 2026 12:53:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(engine):=20=E5=AE=9E=E7=8E=B0=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E5=BC=95=E6=93=8E=E4=B8=8E=20engine=5Fstatus=20?= =?UTF-8?q?=E5=BF=AB=E7=85=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit apps/engine 独立进程（syncRound + 防重叠调度）；上游 log/token；QPU 采样； server 读 engine_status；时刻出参 toUnixSec；中文日志；zod-openapi 扩展顺序修复。 --- AGENTS.md | 1 + apps/engine/package.json | 13 ++ apps/engine/src/env.ts | 43 ++++++ apps/engine/src/index.ts | 29 ++++ apps/engine/src/loop.ts | 75 ++++++++++ apps/engine/src/sync-round.ts | 174 +++++++++++++++++++++++ apps/engine/tsconfig.json | 8 ++ apps/server/src/api-schemas.ts | 45 +++--- apps/server/src/app.ts | 6 +- apps/server/src/index.ts | 4 +- apps/server/src/routes/admin-audit.ts | 4 +- apps/server/src/routes/admin-engine.ts | 34 +++-- apps/server/src/routes/admin-keys.ts | 17 ++- apps/server/src/routes/admin-members.ts | 6 +- apps/server/src/routes/admin-payments.ts | 8 +- apps/server/src/routes/admin-usage.ts | 22 +-- apps/server/src/routes/member-view.ts | 12 +- apps/server/src/zod-openapi-setup.ts | 7 + bun.lock | 10 ++ package.json | 1 + packages/core/src/log.ts | 16 ++- packages/core/src/schemas.ts | 27 ++-- packages/core/src/time.ts | 29 +++- packages/core/src/types.ts | 30 ++-- packages/core/src/upstream.ts | 33 +++++ packages/core/src/usage.ts | 20 +++ packages/db/src/repos.ts | 90 ++++++++++++ packages/db/src/schema.ts | 14 ++ 28 files changed, 676 insertions(+), 102 deletions(-) create mode 100644 apps/engine/package.json create mode 100644 apps/engine/src/env.ts create mode 100644 apps/engine/src/index.ts create mode 100644 apps/engine/src/loop.ts create mode 100644 apps/engine/src/sync-round.ts create mode 100644 apps/engine/tsconfig.json create mode 100644 apps/server/src/zod-openapi-setup.ts diff --git a/AGENTS.md b/AGENTS.md index 6beffd6..f4999b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,6 +29,7 @@ Bun 工具链（选型见 `docs/design/技术选型.md`）。**Windows 上 `bun` | `bun --env-file=.env scripts/seed-from-samples.ts` | 从根目录 `api*.json` 实测快照灌开发库（幂等）；验收接口口径用 | | `bun run typecheck` | 根 tsconfig 类型检查（不含 web `.vue`） | | `bun run dev:server` | 开发跑母体后端（hot reload） | +| `bun run dev:engine` | 开发跑同步引擎（独立进程，读 `SYNC_*` env） | | `bun run dev:web` | 开发跑前端 Vite | | `docker compose up -d` | 起开发用 PostgreSQL | | `bun run --filter @new-api-key-usage/db db:generate` | 生成 SQL 迁移（`packages/db/drizzle/`） | diff --git a/apps/engine/package.json b/apps/engine/package.json new file mode 100644 index 0000000..0d024e5 --- /dev/null +++ b/apps/engine/package.json @@ -0,0 +1,13 @@ +{ + "name": "@new-api-key-usage/engine", + "private": true, + "version": "0.0.0", + "type": "module", + "scripts": { + "dev": "bun run src/index.ts" + }, + "dependencies": { + "@new-api-key-usage/core": "workspace:*", + "@new-api-key-usage/db": "workspace:*" + } +} diff --git a/apps/engine/src/env.ts b/apps/engine/src/env.ts new file mode 100644 index 0000000..e4d2255 --- /dev/null +++ b/apps/engine/src/env.ts @@ -0,0 +1,43 @@ +export interface EngineEnv { + DATABASE_URL: string; + SK_KEY_ENCRYPTION_SECRET: string; + NEWAPI_BASE: string; + SYNC_INTERVAL_MS: number; + LOG_LEVEL: "debug" | "info" | "warning" | "error" | "fatal"; +} + +function parseLogLevel( + value: string | undefined, +): "debug" | "info" | "warning" | "error" | "fatal" { + const v = (value ?? "info").trim().toLowerCase(); + if ( + v === "debug" || + v === "info" || + v === "warning" || + v === "error" || + v === "fatal" + ) { + return v; + } + throw new Error("LOG_LEVEL must be debug|info|warning|error|fatal"); +} + +export function loadEnv(source: Record): EngineEnv { + const databaseUrl = source.DATABASE_URL; + const secret = source.SK_KEY_ENCRYPTION_SECRET; + if (!databaseUrl) throw new Error("DATABASE_URL is required"); + if (!secret) throw new Error("SK_KEY_ENCRYPTION_SECRET is required"); + + const interval = Number(source.SYNC_INTERVAL_MS ?? 300_000); + if (!Number.isInteger(interval) || interval < 1000) { + throw new Error("SYNC_INTERVAL_MS must be an integer >= 1000"); + } + + return { + DATABASE_URL: databaseUrl, + SK_KEY_ENCRYPTION_SECRET: secret, + NEWAPI_BASE: source.NEWAPI_BASE ?? "http://127.0.0.1:8080", + SYNC_INTERVAL_MS: interval, + LOG_LEVEL: parseLogLevel(source.LOG_LEVEL), + }; +} diff --git a/apps/engine/src/index.ts b/apps/engine/src/index.ts new file mode 100644 index 0000000..db6f47f --- /dev/null +++ b/apps/engine/src/index.ts @@ -0,0 +1,29 @@ +import { appLogger, setupLogging } from "@new-api-key-usage/core"; +import { createDb, createRepos } from "@new-api-key-usage/db"; +import { loadEnv } from "./env.ts"; +import { createSyncLoop } from "./loop.ts"; + +const env = loadEnv(process.env); +setupLogging(env.LOG_LEVEL); +const log = appLogger("engine"); + +const db = createDb(env.DATABASE_URL); +const repos = createRepos(db); + +const loop = createSyncLoop({ + repos, + newapiBase: env.NEWAPI_BASE, + skKeyEncryptionSecret: env.SK_KEY_ENCRYPTION_SECRET, + intervalMs: env.SYNC_INTERVAL_MS, +}); + +loop.start(); + +log.info`apps/engine 已就绪`; + +process.on("SIGINT", () => { + void loop.stop().then(() => process.exit(0)); +}); +process.on("SIGTERM", () => { + void loop.stop().then(() => process.exit(0)); +}); diff --git a/apps/engine/src/loop.ts b/apps/engine/src/loop.ts new file mode 100644 index 0000000..78b71ef --- /dev/null +++ b/apps/engine/src/loop.ts @@ -0,0 +1,75 @@ +/** + * 调度循环（方案 A）：一轮结束 → setTimeout 下一枪；inFlight 防重叠。 + * 启停不得抹掉已有 last_round。 + */ +import { appLogger } from "@new-api-key-usage/core"; +import { syncRound, type SyncRoundDeps } from "./sync-round.ts"; + +const log = appLogger("engine"); + +export interface LoopOptions extends SyncRoundDeps { + intervalMs: number; +} + +export function createSyncLoop(opts: LoopOptions) { + let inFlight = false; + let stopped = false; + let timer: ReturnType | null = null; + let last_round: unknown = null; + + async function persist(running: boolean) { + await opts.repos.engine.upsertSnapshot({ + running, + sync_interval_ms: opts.intervalMs, + last_round, + }); + } + + async function runOnce(): Promise { + if (stopped) return; + if (inFlight) { + log.warn("上一轮尚未结束，跳过本枪"); + schedule(); + return; + } + inFlight = true; + try { + last_round = await syncRound(opts); + await persist(!stopped); + } catch (e) { + log.error(`同步轮次异常：${e}`); + await persist(!stopped).catch(() => {}); + } finally { + inFlight = false; + schedule(); + } + } + + function schedule() { + if (stopped) return; + timer = setTimeout(() => { + void runOnce(); + }, opts.intervalMs); + } + + return { + start() { + log.info(`同步引擎启动：周期=${opts.intervalMs}ms`); + void (async () => { + const snap = await opts.repos.engine.getSnapshot(); + last_round = snap?.last_round ?? null; + await persist(true).catch(() => {}); + void runOnce(); + })(); + }, + async stop() { + stopped = true; + if (timer) clearTimeout(timer); + // 等在跑的一轮结束，避免 stop 与 upsert 交错 + for (let i = 0; i < 50 && inFlight; i++) { + await new Promise((r) => setTimeout(r, 100)); + } + await persist(false).catch(() => {}); + }, + }; +} diff --git a/apps/engine/src/sync-round.ts b/apps/engine/src/sync-round.ts new file mode 100644 index 0000000..83fa00c --- /dev/null +++ b/apps/engine/src/sync-round.ts @@ -0,0 +1,174 @@ +/** + * 同步轮次：按《同步引擎设计》§3 编排。 + * 只经 packages/core（上游 + 去重）与 packages/db（仓储）写库。 + */ +import { + appLogger, + buildDedupeKey, + createUpstreamClient, + KeyCrypto, + parseOtherJson, + toUnixSec, +} from "@new-api-key-usage/core"; +import type { Repos } from "@new-api-key-usage/db"; + +const log = appLogger("engine"); + +export interface SyncRoundDeps { + repos: Repos; + newapiBase: string; + skKeyEncryptionSecret: string; +} + +export interface LastRound { + started_at: number; + keys_total: number; + succeeded: number; + failed: number; + failed_api_key_ids: number[]; +} + +function parseOther(other: string | undefined): { + billing_source: string | null; + subscription_id: number | null; + raw: Record | null; +} { + return parseOtherJson(other); +} + +export async function syncRound(deps: SyncRoundDeps): Promise { + const started = Date.now(); + const started_at = toUnixSec(new Date())!; + const client = createUpstreamClient({ baseUrl: deps.newapiBase }); + const crypto = await KeyCrypto.fromEncryptionSecret( + deps.skKeyEncryptionSecret, + ); + + const keys = await deps.repos.keys.listActive(); + const failed_api_key_ids: number[] = []; + let succeeded = 0; + let pulled = 0; + let inserted = 0; + let skipped = 0; + + for (const key of keys) { + try { + const sk = await crypto.openKey(key.key_ciphertext); + + const usage = await client.fetchTokenUsage(sk); + if (!usage.ok) { + failed_api_key_ids.push(key.id); + log.warn(`同步失败 key_id=${key.id} 原因=usage/${usage.kind}`); + continue; + } + await deps.repos.keys.update(key.id, { + token_name: usage.data.name ?? key.token_name, + total_granted: usage.data.total_granted, + total_used: usage.data.total_used, + total_available: usage.data.total_available, + unlimited_quota: usage.data.unlimited_quota, + model_limits: usage.data.model_limits, + model_limits_enabled: usage.data.model_limits_enabled, + expires_at: normalizeExpires(usage.data.expires_at), + usage_synced_at: new Date(), + }); + + const logs = await client.fetchTokenLogs(sk); + if (!logs.ok) { + failed_api_key_ids.push(key.id); + log.warn(`同步失败 key_id=${key.id} 原因=logs/${logs.kind}`); + continue; + } + + const rows = [...(logs.data ?? [])].sort((a, b) => { + const ta = a.created_at ?? 0; + const tb = b.created_at ?? 0; + if (ta !== tb) return ta - tb; + return String(a.request_id ?? "").localeCompare(String(b.request_id ?? "")); + }); + pulled += rows.length; + + for (const row of rows) { + const other = parseOther(row.other); + const created_at = new Date(row.created_at * 1000); + const dedupe_key = await buildDedupeKey({ + api_key_id: key.id, + request_id: row.request_id, + type: row.type ?? 2, + created_at: row.created_at, + model_name: row.model_name ?? "", + quota: row.quota ?? 0, + prompt_tokens: row.prompt_tokens ?? 0, + completion_tokens: row.completion_tokens ?? 0, + use_time: row.use_time ?? 0, + content: row.content ?? "", + }); + const ok = await deps.repos.keys.insertLog({ + api_key_id: key.id, + dedupe_key, + billing_source: other.billing_source, + subscription_id: other.subscription_id, + user_id: row.user_id ?? 0, + created_at, + type: row.type ?? 2, + content: row.content ?? "", + username: row.username ?? "", + token_name: row.token_name ?? "", + model_name: row.model_name ?? "", + quota: row.quota ?? 0, + prompt_tokens: row.prompt_tokens ?? 0, + completion_tokens: row.completion_tokens ?? 0, + use_time: row.use_time ?? 0, + is_stream: !!row.is_stream, + channel: row.channel ?? 0, + channel_name: row.channel_name ?? "", + token_id: row.token_id ?? 0, + group_name: row.group ?? "", + ip: row.ip ?? "", + request_id: row.request_id ?? "", + upstream_request_id: row.upstream_request_id ?? null, + other: other.raw, + }); + if (ok) inserted += 1; + else skipped += 1; + } + await deps.repos.keys.update(key.id, { log_synced_at: new Date() }); + succeeded += 1; + } catch (e) { + failed_api_key_ids.push(key.id); + log.error(`同步异常 key_id=${key.id} ${e}`); + } + } + + // QPU 采样：值变化才追加 + const qpuRes = await client.fetchQuotaPerUnit(); + if (qpuRes.ok) { + const latest = await deps.repos.qpu.latest(); + const next = qpuRes.data.quota_per_unit; + if (!latest || latest.qpu !== next) { + await deps.repos.qpu.insertChange(next, new Date()); + log.info(`QPU 变更 ${latest?.qpu ?? "-"} -> ${next}`); + } + } else { + log.warn(`QPU 采样失败 ${qpuRes.kind}`); + } + + const last_round: LastRound = { + started_at, + keys_total: keys.length, + succeeded, + failed: failed_api_key_ids.length, + failed_api_key_ids, + }; + + log.info( + `同步轮次完成 key=${keys.length} 成功=${succeeded} 失败=${failed_api_key_ids.length} 拉取=${pulled} 新写=${inserted} 去重跳过=${skipped} 耗时=${Date.now() - started}ms`, + ); + + return last_round; +} + +function normalizeExpires(unixSeconds: number | undefined | null): Date | null { + if (!unixSeconds) return null; + return new Date(unixSeconds * 1000); +} diff --git a/apps/engine/tsconfig.json b/apps/engine/tsconfig.json new file mode 100644 index 0000000..3cdc5e8 --- /dev/null +++ b/apps/engine/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "types": ["bun"], + "allowImportingTsExtensions": true + }, + "include": ["src/**/*.ts"] +} diff --git a/apps/server/src/api-schemas.ts b/apps/server/src/api-schemas.ts index 1f80587..160bb8a 100644 --- a/apps/server/src/api-schemas.ts +++ b/apps/server/src/api-schemas.ts @@ -1,5 +1,5 @@ -import { z } from "zod"; -import { extendZodWithOpenApi } from "@hono/zod-openapi"; +/** 必须先扩展 zod，再加载 core schema（见 zod-openapi-setup） */ +import "./zod-openapi-setup.ts"; import { changePasswordBody, createMemberBody, @@ -24,21 +24,34 @@ import { usernameBody, } from "@new-api-key-usage/core"; -// OpenAPI 组件命名只在 server 层装饰，core 保持 Web 标准中立 -extendZodWithOpenApi(z); +/** OpenAPI 组件命名只在 server 层装饰，core 保持 Web 标准中立 */ +function nameSchema unknown }>( + schema: T, + name: string, +): T { + return typeof schema.openapi === "function" + ? (schema.openapi(name) as T) + : schema; +} -export const errorSchema = errorEnvelope.openapi("ErrorEnvelope"); -export const setupStateSchema = setupState.openapi("SetupState"); -export const engineStatusSchema = engineStatus.openapi("EngineStatus"); -export const keyItemSchema = keyItem.openapi("KeyItem"); -export const memberItemSchema = memberItem.openapi("MemberItem"); -export const memberWithTokenSchema = memberWithToken.openapi("MemberWithToken"); -export const memberMeSchema = memberMe.openapi("MemberMe"); -export const memberOverviewSchema = memberOverview.openapi("MemberOverview"); -export const logItemSchema = logItem.openapi("LogItem"); -export const logPageSchema = logPage.openapi("LogPage"); -export const paymentItemSchema = paymentItem.openapi("PaymentItem"); -export const auditLogItemSchema = auditLogItem.openapi("AuditLogItem"); +export const errorSchema = nameSchema(errorEnvelope, "ErrorEnvelope"); +export const setupStateSchema = nameSchema(setupState, "SetupState"); +export const engineStatusSchema = nameSchema(engineStatus, "EngineStatus"); +export const keyItemSchema = nameSchema(keyItem, "KeyItem"); +export const memberItemSchema = nameSchema(memberItem, "MemberItem"); +export const memberWithTokenSchema = nameSchema( + memberWithToken, + "MemberWithToken", +); +export const memberMeSchema = nameSchema(memberMe, "MemberMe"); +export const memberOverviewSchema = nameSchema( + memberOverview, + "MemberOverview", +); +export const logItemSchema = nameSchema(logItem, "LogItem"); +export const logPageSchema = nameSchema(logPage, "LogPage"); +export const paymentItemSchema = nameSchema(paymentItem, "PaymentItem"); +export const auditLogItemSchema = nameSchema(auditLogItem, "AuditLogItem"); export { changePasswordBody as changePasswordBodySchema, diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index bfe9b10..4878119 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -43,7 +43,9 @@ export function createApp(opts: CreateAppOptions) { try { await next(); } finally { - appLogger("http").info`${c.req.method} ${c.req.path} → ${c.res.status} · ${Date.now() - start}ms`; + appLogger("http").info( + `${c.req.method} ${c.req.path} -> ${c.res.status} | ${Date.now() - start}ms`, + ); } }); app.use("*", async (c, next) => { @@ -67,7 +69,7 @@ export function createApp(opts: CreateAppOptions) { }); app.onError((err, c) => { - appLogger("server").error`未捕获错误 ${c.req.method} ${c.req.path}：${err}`; + appLogger("server").error(`未捕获错误 ${c.req.method} ${c.req.path}：${err}`); return jsonError(c, 500, ERR.internal()); }); diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 3e8e2d7..656441d 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -17,7 +17,9 @@ const app = createApp({ adminInitPassword: env.ADMIN_INIT_PASSWORD, }); -appLogger("server").info`服务已启动，端口 ${env.PORT}（cookie_secure=${env.COOKIE_SECURE}，日志级别=${env.LOG_LEVEL}`; +appLogger("server").info( + `服务已启动，端口 ${env.PORT}（cookie_secure=${env.COOKIE_SECURE}，日志级别=${env.LOG_LEVEL}）`, +); export default { port: env.PORT, diff --git a/apps/server/src/routes/admin-audit.ts b/apps/server/src/routes/admin-audit.ts index 02f51dc..a4efa2c 100644 --- a/apps/server/src/routes/admin-audit.ts +++ b/apps/server/src/routes/admin-audit.ts @@ -1,5 +1,5 @@ import { OpenAPIHono, createRoute } from "@hono/zod-openapi"; -import { unixToIso, type AuditLogItem } from "@new-api-key-usage/core"; +import { toUnixSec, type AuditLogItem } from "@new-api-key-usage/core"; import { requireAdminSession, type AppVars, @@ -56,7 +56,7 @@ export function adminAuditRoutes() { }); const items: AuditLogItem[] = rows.map((row) => ({ id: row.id, - created_at: unixToIso(row.created_at)!, + created_at: toUnixSec(row.created_at)!, operator: row.operator, domain: row.domain as AuditLogItem["domain"], action: row.action, diff --git a/apps/server/src/routes/admin-engine.ts b/apps/server/src/routes/admin-engine.ts index 1d39996..92b04bd 100644 --- a/apps/server/src/routes/admin-engine.ts +++ b/apps/server/src/routes/admin-engine.ts @@ -3,11 +3,22 @@ import type { EngineStatus } from "@new-api-key-usage/core"; import { requireAdminSession, type AppVars } from "../middleware.ts"; import { engineStatusSchema, err401 } from "../api-schemas.ts"; -/** 状态存储归同步引擎设计；此处只暴露契约字段 */ -let engineStatus: EngineStatus = { - running: false, - last_round: null, -}; +export type EngineStatusView = EngineStatus & { sync_interval_ms: number }; + +export function toEngineStatusView(row: { + running: boolean; + sync_interval_ms: number; + last_round: unknown; +} | null): EngineStatusView { + if (!row) { + return { running: false, sync_interval_ms: 300_000, last_round: null }; + } + return { + running: row.running, + sync_interval_ms: row.sync_interval_ms, + last_round: (row.last_round as EngineStatus["last_round"]) ?? null, + }; +} export function adminEngineRoutes() { const r = new OpenAPIHono<{ Variables: AppVars }>(); @@ -19,7 +30,7 @@ export function adminEngineRoutes() { path: "/engine", summary: "引擎状态", description: - "最近轮次摘要；last_round 为 null 表示引擎尚未跑过。状态存储细节归同步引擎设计。", + "同步引擎单行快照（engine_status）；last_round 为 null 表示尚未跑过。sync_interval_ms 只读。", tags: ["引擎"], security: [{ adminSession: [] }], responses: { @@ -31,17 +42,10 @@ export function adminEngineRoutes() { }, }), async (c) => { - return c.json(engineStatus, 200); + const row = await c.get("repos").engine.getSnapshot(); + return c.json(toEngineStatusView(row), 200); }, ); return r; } - -export function setEngineStatus(next: EngineStatus) { - engineStatus = next; -} - -export function getEngineStatus(): EngineStatus { - return engineStatus; -} diff --git a/apps/server/src/routes/admin-keys.ts b/apps/server/src/routes/admin-keys.ts index f9c97b0..c8659d6 100644 --- a/apps/server/src/routes/admin-keys.ts +++ b/apps/server/src/routes/admin-keys.ts @@ -3,7 +3,7 @@ import { ERR, KeyCrypto, normalizeExpiresAt, - unixToIso, + toUnixSec, type KeyItem, } from "@new-api-key-usage/core"; import type { ApiKeyRow } from "@new-api-key-usage/db"; @@ -30,14 +30,14 @@ function toKeyItem(row: ApiKeyRow & { member_name?: string | null }): KeyItem { status: row.status as "active" | "disabled", member_id: row.member_id, member_name: row.member_name ?? null, - registered_at: unixToIso(row.registered_at)!, - log_synced_at: unixToIso(row.log_synced_at), - usage_synced_at: unixToIso(row.usage_synced_at), + registered_at: toUnixSec(row.registered_at)!, + log_synced_at: toUnixSec(row.log_synced_at), + usage_synced_at: toUnixSec(row.usage_synced_at), total_used: row.total_used, total_available: row.total_available, total_granted: row.total_granted, unlimited_quota: row.unlimited_quota, - expires_at: unixToIso(row.expires_at), + expires_at: toUnixSec(row.expires_at), }; } @@ -85,8 +85,11 @@ export function adminKeyRoutes() { } let rows = await c.get("repos").keys.listAll(filter); if (q.recent_failed === "true") { - const { getEngineStatus } = await import("./admin-engine.ts"); - const failed = new Set(getEngineStatus().last_round?.failed_api_key_ids ?? []); + const snap = await c.get("repos").engine.getSnapshot(); + const failed = new Set( + ((snap?.last_round as { failed_api_key_ids?: number[] } | null) + ?.failed_api_key_ids) ?? [], + ); rows = rows.filter((r) => failed.has(r.id)); } return c.json(rows.map(toKeyItem), 200); diff --git a/apps/server/src/routes/admin-members.ts b/apps/server/src/routes/admin-members.ts index 7d3d364..47e752a 100644 --- a/apps/server/src/routes/admin-members.ts +++ b/apps/server/src/routes/admin-members.ts @@ -1,5 +1,5 @@ import { OpenAPIHono, createRoute } from "@hono/zod-openapi"; -import { ERR, generateToken, unixToIso } from "@new-api-key-usage/core"; +import { ERR, generateToken, toUnixSec } from "@new-api-key-usage/core"; import type { MemberRow } from "@new-api-key-usage/db"; import { jsonError, @@ -25,8 +25,8 @@ function toMemberItem(row: MemberRow, key_count: number) { name: row.name, remark: row.remark, status: row.status as "active" | "disabled", - token_issued_at: unixToIso(row.token_issued_at), - created_at: unixToIso(row.created_at)!, + token_issued_at: toUnixSec(row.token_issued_at), + created_at: toUnixSec(row.created_at)!, key_count, }; } diff --git a/apps/server/src/routes/admin-payments.ts b/apps/server/src/routes/admin-payments.ts index a052de0..6550221 100644 --- a/apps/server/src/routes/admin-payments.ts +++ b/apps/server/src/routes/admin-payments.ts @@ -4,7 +4,7 @@ import { qpuAt, quotaToUsdStr, resolveQpuTimeline, - unixToIso, + toUnixSec, type PaymentItem, } from "@new-api-key-usage/core"; import type { PaymentRow } from "@new-api-key-usage/db"; @@ -31,11 +31,11 @@ function toPaymentItem(row: PaymentRow, quota_usd: string): PaymentItem { return { id: row.id, kind: row.kind as PaymentItem["kind"], - paid_at: unixToIso(row.paid_at)!, + paid_at: toUnixSec(row.paid_at)!, paid_amount: row.paid_amount, remark: row.remark, - period_start: unixToIso(row.period_start), - period_end: unixToIso(row.period_end), + period_start: toUnixSec(row.period_start), + period_end: toUnixSec(row.period_end), subscription_id: row.subscription_id, quota_granted: row.quota_granted, quota_usd, diff --git a/apps/server/src/routes/admin-usage.ts b/apps/server/src/routes/admin-usage.ts index 42c58d2..c007e15 100644 --- a/apps/server/src/routes/admin-usage.ts +++ b/apps/server/src/routes/admin-usage.ts @@ -3,7 +3,7 @@ import { ERR, resolveQpuTimeline, sumUsage, - unixToIso, + toUnixSec, type LogItem, } from "@new-api-key-usage/core"; import { @@ -25,7 +25,7 @@ const usageTotals = z.object({ }); const overviewSchema = z.object({ - data_cutoff: z.string().nullable(), + data_cutoff: z.number().int().nullable(), unassigned_subscription: usageTotals, by_source: z.array( z.object({ @@ -40,7 +40,7 @@ const overviewSchema = z.object({ member_id: z.number().int().nullable(), member_name: z.string().nullable(), key_count: z.number().int(), - data_cutoff: z.string().nullable(), + data_cutoff: z.number().int().nullable(), usage: usageTotals, }), ), @@ -50,7 +50,7 @@ const overviewSchema = z.object({ token_name: z.string(), member_id: z.number().int().nullable(), status: z.string(), - data_cutoff: z.string().nullable(), + data_cutoff: z.number().int().nullable(), usage: usageTotals, }), ), @@ -58,7 +58,7 @@ const overviewSchema = z.object({ const qpuLatestSchema = z.object({ qpu: z.number(), - effective_from: z.string(), + effective_from: z.number().int(), }); const timeRange = z.object({ @@ -106,14 +106,14 @@ export function adminUsageRoutes() { const latest = await c.get("repos").qpu.latest(); if (!latest) { return c.json( - { qpu: 500_000, effective_from: "1970-01-01T00:00:00Z" }, + { qpu: 500_000, effective_from: 0 }, 200, ); } return c.json( { qpu: latest.qpu, - effective_from: unixToIso(latest.effective_from)!, + effective_from: toUnixSec(latest.effective_from)!, }, 200, ); @@ -208,7 +208,7 @@ export function adminUsageRoutes() { token_name: v.token_name, member_id: v.member_id, status: v.status, - data_cutoff: unixToIso(v.log_synced_at), + data_cutoff: toUnixSec(v.log_synced_at), usage: sumUsage(v.rows, timeline), })) .sort((a, b) => b.usage.quota - a.usage.quota); @@ -248,13 +248,13 @@ export function adminUsageRoutes() { member_id: b.member_id, member_name: b.member_name, key_count: b.key_count, - data_cutoff: unixToIso(b.cutoff), + data_cutoff: toUnixSec(b.cutoff), usage: sumUsage(b.rows, timeline), })); return c.json( { - data_cutoff: unixToIso(data_cutoff), + data_cutoff: toUnixSec(data_cutoff), unassigned_subscription, by_source, by_member, @@ -326,7 +326,7 @@ export function adminUsageRoutes() { return { id: row.id, api_key_id: row.api_key_id, - created_at: unixToIso(row.created_at)!, + created_at: toUnixSec(row.created_at)!, token_name: row.token_name, model_name: row.model_name, type: row.type, diff --git a/apps/server/src/routes/member-view.ts b/apps/server/src/routes/member-view.ts index c57bbdd..e9ce286 100644 --- a/apps/server/src/routes/member-view.ts +++ b/apps/server/src/routes/member-view.ts @@ -3,7 +3,7 @@ import { ERR, resolveQpuTimeline, sumUsage, - unixToIso, + toUnixSec, type LogItem, type PaymentItem, } from "@new-api-key-usage/core"; @@ -156,7 +156,7 @@ export function memberViewRoutes() { const keys = [...keyMap.entries()].map(([key_id, v]) => ({ key_id, token_name: v.token_name, - log_synced_at: unixToIso(v.log_synced_at), + log_synced_at: toUnixSec(v.log_synced_at), usage: sumUsage(v.rows, timeline), })); @@ -185,11 +185,11 @@ export function memberViewRoutes() { payments.push({ id: p.id, kind: p.kind as PaymentItem["kind"], - paid_at: unixToIso(p.paid_at)!, + paid_at: toUnixSec(p.paid_at)!, paid_amount: p.paid_amount, remark: p.remark, - period_start: unixToIso(p.period_start), - period_end: unixToIso(p.period_end), + period_start: toUnixSec(p.period_start), + period_end: toUnixSec(p.period_end), subscription_id: p.subscription_id, quota_granted: p.quota_granted, quota_usd: @@ -272,7 +272,7 @@ export function memberViewRoutes() { return { id: row.id, api_key_id: row.api_key_id, - created_at: unixToIso(row.created_at)!, + created_at: toUnixSec(row.created_at)!, token_name: row.token_name ?? "", model_name: row.model_name, type: row.type, diff --git a/apps/server/src/zod-openapi-setup.ts b/apps/server/src/zod-openapi-setup.ts new file mode 100644 index 0000000..bf73bd9 --- /dev/null +++ b/apps/server/src/zod-openapi-setup.ts @@ -0,0 +1,7 @@ +/** 必须在 core 的 zod schema 被求值之前执行（api-schemas 首行 side-effect import） */ +import { z } from "zod"; +import { extendZodWithOpenApi } from "@hono/zod-openapi"; + +extendZodWithOpenApi(z); + +export {}; diff --git a/bun.lock b/bun.lock index 69f13a9..34239f1 100644 --- a/bun.lock +++ b/bun.lock @@ -9,6 +9,14 @@ "typescript": "^7.0.2", }, }, + "apps/engine": { + "name": "@new-api-key-usage/engine", + "version": "0.0.0", + "dependencies": { + "@new-api-key-usage/core": "workspace:*", + "@new-api-key-usage/db": "workspace:*", + }, + }, "apps/server": { "name": "@new-api-key-usage/server", "version": "0.0.0", @@ -291,6 +299,8 @@ "@new-api-key-usage/db": ["@new-api-key-usage/db@workspace:packages/db"], + "@new-api-key-usage/engine": ["@new-api-key-usage/engine@workspace:apps/engine"], + "@new-api-key-usage/server": ["@new-api-key-usage/server@workspace:apps/server"], "@new-api-key-usage/web": ["@new-api-key-usage/web@workspace:apps/web"], diff --git a/package.json b/package.json index 7347d78..9bc486c 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "test": "bun test", "typecheck": "tsc -p tsconfig.json --noEmit", "dev:server": "bun --hot --env-file=.env apps/server/src/index.ts", + "dev:engine": "bun --env-file=.env apps/engine/src/index.ts", "dev:web": "bun --filter @new-api-key-usage/web dev" }, "devDependencies": { diff --git a/packages/core/src/log.ts b/packages/core/src/log.ts index b612afd..e03270e 100644 --- a/packages/core/src/log.ts +++ b/packages/core/src/log.ts @@ -23,6 +23,20 @@ export function redactSecrets(message: string): string { ); } +/** 日志时间戳：本地时区、可读（终端运维友好） */ +export function formatLogTime(d: Date = new Date()): string { + const pad = (n: number, w = 2) => String(n).padStart(w, "0"); + const offsetMin = -d.getTimezoneOffset(); + const sign = offsetMin >= 0 ? "+" : "-"; + const abs = Math.abs(offsetMin); + const off = `${sign}${pad(Math.floor(abs / 60))}:${pad(abs % 60)}`; + return ( + `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}` + + ` ${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}` + + `.${pad(d.getMilliseconds(), 3)}${off}` + ); +} + let configured = false; /** @@ -37,7 +51,7 @@ export function setupLogging(level: AppLogLevel = "info"): void { sinks: { console: getConsoleSink({ formatter: (record) => { - const ts = new Date(record.timestamp).toISOString(); + const ts = formatLogTime(new Date(record.timestamp)); const cat = record.category.join("."); const msg = redactSecrets(String(record.message)); return `${ts} ${record.level.toUpperCase()} [${cat}] ${msg}`; diff --git a/packages/core/src/schemas.ts b/packages/core/src/schemas.ts index 5dd5897..a08ed22 100644 --- a/packages/core/src/schemas.ts +++ b/packages/core/src/schemas.ts @@ -52,7 +52,7 @@ export const usageTotals = z.object({ export const logItem = z.object({ id: z.number().int(), api_key_id: z.number().int(), - created_at: z.string(), + created_at: z.number().int(), token_name: z.string(), model_name: z.string(), type: z.number().int(), @@ -78,14 +78,14 @@ export const keyItem = z.object({ status: z.enum(["active", "disabled"]), member_id: z.number().int().nullable(), member_name: z.string().nullable(), - registered_at: z.string(), - log_synced_at: z.string().nullable(), - usage_synced_at: z.string().nullable(), + registered_at: z.number().int(), + log_synced_at: z.number().int().nullable(), + usage_synced_at: z.number().int().nullable(), total_used: z.number(), total_available: z.number(), total_granted: z.number(), unlimited_quota: z.boolean(), - expires_at: z.string().nullable(), + expires_at: z.number().int().nullable(), }); export const registerKeyBody = z.object({ @@ -103,8 +103,8 @@ export const memberItem = z.object({ name: z.string(), remark: z.string(), status: z.enum(["active", "disabled"]), - token_issued_at: z.string().nullable(), - created_at: z.string(), + token_issued_at: z.number().int().nullable(), + created_at: z.number().int(), key_count: z.number().int(), }); @@ -132,11 +132,11 @@ export const memberMe = z.object({ export const paymentItem = z.object({ id: z.number().int(), kind: z.enum(["subscription", "topup"]), - paid_at: z.string(), + paid_at: z.number().int(), paid_amount: z.number(), remark: z.string(), - period_start: z.string().nullable(), - period_end: z.string().nullable(), + period_start: z.number().int().nullable(), + period_end: z.number().int().nullable(), subscription_id: z.number().int().nullable(), quota_granted: z.number().int().nullable(), quota_usd: z.string(), @@ -144,9 +144,10 @@ export const paymentItem = z.object({ export const engineStatus = z.object({ running: z.boolean(), + sync_interval_ms: z.number().int().optional(), last_round: z .object({ - started_at: z.string(), + started_at: z.number().int(), keys_total: z.number().int(), succeeded: z.number().int(), failed: z.number().int(), @@ -157,7 +158,7 @@ export const engineStatus = z.object({ export const auditLogItem = z.object({ id: z.number().int(), - created_at: z.string(), + created_at: z.number().int(), operator: z.string().nullable(), domain: z.enum(["key", "payment", "member", "account"]), action: z.string(), @@ -182,7 +183,7 @@ export const memberOverview = z.object({ z.object({ key_id: z.number().int(), token_name: z.string(), - log_synced_at: z.string().nullable(), + log_synced_at: z.number().int().nullable(), usage: usageTotals, }), ), diff --git a/packages/core/src/time.ts b/packages/core/src/time.ts index f4b6458..0401ec8 100644 --- a/packages/core/src/time.ts +++ b/packages/core/src/time.ts @@ -1,18 +1,35 @@ -/** 存储层 UTC Unix 秒 / Date / ISO 字符串 ↔ DTO 层 ISO 8601 UTC */ -export function unixToIso( - value: number | string | Date | null | undefined, -): string | null { +/** + * 时间口径（API 响应统一 Unix 秒整数；请求 from/to 仍收 ISO 8601 便于试调） + * 库内 timestamptz ↔ JS Date ↔ Unix 秒 + */ +function toDate(value: number | string | Date | null | undefined): Date | null { if (value === null || value === undefined) return null; let d: Date; if (value instanceof Date) { d = value; } else if (typeof value === "number") { - // Unix 秒（或毫秒：量级区分） d = new Date(value < 1e12 ? value * 1000 : value); } else { d = new Date(value); } - if (Number.isNaN(d.getTime())) return null; + return Number.isNaN(d.getTime()) ? null : d; +} + +/** → Unix 秒（UTC）；无效则 null */ +export function toUnixSec( + value: number | string | Date | null | undefined, +): number | null { + const d = toDate(value); + if (!d) return null; + return Math.floor(d.getTime() / 1000); +} + +/** 兼容旧名：ISO 串（仅本地/文档用；API 响应改用 toUnixSec） */ +export function unixToIso( + value: number | string | Date | null | undefined, +): string | null { + const d = toDate(value); + if (!d) return null; return d.toISOString().replace(/\.\d{3}Z$/, "Z"); } diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 6abcfd1..39d6350 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -1,5 +1,5 @@ -/** ISO 8601 UTC 字符串，如 2026-10-02T14:22:05Z */ -export type Iso8601Utc = string; +/** Unix 秒（UTC 时刻）——API 响应时刻字段统一此类型 */ +export type UnixSec = number; /** 支付金额（币种无关数值） */ export type PaidAmount = number; @@ -24,7 +24,7 @@ export interface UsageTotals { export interface LogItem { id: number; api_key_id: number; - created_at: Iso8601Utc; + created_at: UnixSec; token_name: string; model_name: string; type: number; @@ -45,24 +45,24 @@ export interface KeyItem { status: KeyStatus; member_id: number | null; member_name: string | null; - registered_at: Iso8601Utc; - log_synced_at: Iso8601Utc | null; - usage_synced_at: Iso8601Utc | null; + registered_at: UnixSec; + log_synced_at: UnixSec | null; + usage_synced_at: UnixSec | null; total_used: number; total_available: number; total_granted: number; unlimited_quota: boolean; - expires_at: Iso8601Utc | null; + expires_at: UnixSec | null; } export interface PaymentItem { id: number; kind: PaymentKind; - paid_at: Iso8601Utc; + paid_at: UnixSec; paid_amount: PaidAmount; remark: string; - period_start: Iso8601Utc | null; - period_end: Iso8601Utc | null; + period_start: UnixSec | null; + period_end: UnixSec | null; subscription_id: number | null; quota_granted: number | null; quota_usd: QuotaUsd; @@ -73,14 +73,14 @@ export interface MemberItem { name: string; remark: string; status: MemberStatus; - token_issued_at: Iso8601Utc | null; - created_at: Iso8601Utc; + token_issued_at: UnixSec | null; + created_at: UnixSec; key_count: number; } export interface AuditLogItem { id: number; - created_at: Iso8601Utc; + created_at: UnixSec; operator: string | null; domain: "key" | "payment" | "member" | "account"; action: string; @@ -93,7 +93,7 @@ export interface Paginated { } export interface EngineLastRound { - started_at: Iso8601Utc; + started_at: UnixSec; keys_total: number; succeeded: number; failed: number; @@ -107,7 +107,7 @@ export interface EngineStatus { export interface QpuLatest { qpu: number; - effective_from: Iso8601Utc; + effective_from: UnixSec; } export type AuditAction = diff --git a/packages/core/src/upstream.ts b/packages/core/src/upstream.ts index ccf949d..7c99fe0 100644 --- a/packages/core/src/upstream.ts +++ b/packages/core/src/upstream.ts @@ -73,6 +73,30 @@ async function requestJson( } } +export interface TokenLogRow { + 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; +} + export function createUpstreamClient(opts: UpstreamClientOptions) { const timeoutMs = opts.timeoutMs ?? 8_000; @@ -87,6 +111,15 @@ export function createUpstreamClient(opts: UpstreamClientOptions) { return r; }, + /** 轮询日志：GET /api/log/token（最近最多 1000 条） */ + async fetchTokenLogs(skKey: string): Promise> { + return requestJson( + joinUrl(opts.baseUrl, "/api/log/token"), + skKey, + timeoutMs, + ); + }, + /** 换算率权威来源：GET /api/status（取 quota_per_unit） */ async fetchQuotaPerUnit(): Promise> { return requestJson<{ quota_per_unit: number }>( diff --git a/packages/core/src/usage.ts b/packages/core/src/usage.ts index 6039d6c..a7b606a 100644 --- a/packages/core/src/usage.ts +++ b/packages/core/src/usage.ts @@ -62,3 +62,23 @@ export function addUsage(a: UsageSum, b: UsageSum): UsageSum { quota_usd: a.quota_usd, }; } + +/** 解析日志行 other JSON（billing_source / subscription_id） */ +export function parseOtherJson(other: string | undefined | null): { + billing_source: string | null; + subscription_id: number | null; + raw: Record | null; +} { + if (!other) return { billing_source: null, subscription_id: null, raw: null }; + try { + const o = JSON.parse(other) as Record; + return { + billing_source: typeof o.billing_source === "string" ? o.billing_source : null, + subscription_id: + typeof o.subscription_id === "number" ? o.subscription_id : null, + raw: o, + }; + } catch { + return { billing_source: null, subscription_id: null, raw: null }; + } +} diff --git a/packages/db/src/repos.ts b/packages/db/src/repos.ts index 0cb60db..a7f1496 100644 --- a/packages/db/src/repos.ts +++ b/packages/db/src/repos.ts @@ -2,6 +2,7 @@ import type { Db } from "./index.ts"; import { apiKeys, auditLogs, + engineStatus, members, payments, qpuHistory, @@ -10,6 +11,7 @@ import { users, type ApiKeyRow, type AuditLogRow, + type EngineStatusRow, type MemberRow, type PaymentRow, type SessionRow, @@ -148,6 +150,54 @@ export function createRepos(db: Db) { } return base; }, + async listActive() { + return db + .select({ + id: apiKeys.id, + key_hash: apiKeys.key_hash, + key_ciphertext: apiKeys.key_ciphertext, + status: apiKeys.status, + member_id: apiKeys.member_id, + token_name: apiKeys.token_name, + }) + .from(apiKeys) + .where(eq(apiKeys.status, "active")) + .orderBy(asc(apiKeys.id)); + }, + async insertLog(values: { + api_key_id: number; + dedupe_key: string; + billing_source: string | null; + subscription_id: number | null; + user_id: number; + created_at: Date; + 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_name: string; + ip: string; + request_id: string; + upstream_request_id: string | null; + other: unknown; + }): Promise { + const before = await db + .select({ n: count() }) + .from(usageLogs) + .where(eq(usageLogs.dedupe_key, values.dedupe_key)); + if ((before[0]?.n ?? 0) > 0) return false; + await db.insert(usageLogs).values(values).onConflictDoNothing(); + return true; + }, async findById(id: number): Promise { const rows = await db .select() @@ -293,6 +343,43 @@ export function createRepos(db: Db) { }, }, + engine: { + async upsertSnapshot(values: { + running: boolean; + sync_interval_ms: number; + last_round: unknown; + }): Promise { + const [row] = await db + .insert(engineStatus) + .values({ + id: 1, + running: values.running, + sync_interval_ms: values.sync_interval_ms, + last_round: values.last_round, + updated_at: new Date(), + }) + .onConflictDoUpdate({ + target: engineStatus.id, + set: { + running: values.running, + sync_interval_ms: values.sync_interval_ms, + last_round: values.last_round, + updated_at: new Date(), + }, + }) + .returning(); + return row!; + }, + async getSnapshot(): Promise { + const rows = await db + .select() + .from(engineStatus) + .where(eq(engineStatus.id, 1)) + .limit(1); + return rows[0] ?? null; + }, + }, + payments: { async listAll(): Promise { return db.select().from(payments).orderBy(desc(payments.paid_at)); @@ -354,6 +441,9 @@ export function createRepos(db: Db) { .limit(1); return rows[0] ?? null; }, + async insertChange(qpu: number, effective_from: Date): Promise { + await db.insert(qpuHistory).values({ qpu, effective_from }); + }, }, logs: { diff --git a/packages/db/src/schema.ts b/packages/db/src/schema.ts index 9e4c2d8..31c0940 100644 --- a/packages/db/src/schema.ts +++ b/packages/db/src/schema.ts @@ -191,6 +191,19 @@ export const auditLogs = pgTable( (t) => [index("audit_logs_created_at_idx").on(t.created_at)], ); +/** 同步引擎状态单行快照（《同步引擎设计》§5） */ +export const engineStatus = pgTable("engine_status", { + id: integer("id").primaryKey().default(1), + running: boolean("running").notNull().default(false), + sync_interval_ms: bigint("sync_interval_ms", { mode: "number" }) + .notNull() + .default(300_000), + last_round: jsonb("last_round"), + updated_at: timestamp("updated_at", { withTimezone: true, mode: "date" }) + .notNull() + .defaultNow(), +}); + export type MemberRow = typeof members.$inferSelect; export type ApiKeyRow = typeof apiKeys.$inferSelect; export type UsageLogRow = typeof usageLogs.$inferSelect; @@ -199,3 +212,4 @@ export type QpuHistoryRow = typeof qpuHistory.$inferSelect; export type UserRow = typeof users.$inferSelect; export type SessionRow = typeof sessions.$inferSelect; export type AuditLogRow = typeof auditLogs.$inferSelect; +export type EngineStatusRow = typeof engineStatus.$inferSelect;