From 8252b25396b63c2a4e0147f942584cce2f266989 Mon Sep 17 00:00:00 2001 From: skycurtain Date: Tue, 15 Sep 2026 00:51:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20=E6=9F=A5=E7=9C=8B=E9=9D=A2?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E9=9D=A2=EF=BC=88=E9=98=B6=E6=AE=B5=205?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - logs/query：明细 1:1（除 dedupe_key）+ 当前 token_name（join）+ amount_rmb；分页（limit/offset，默认 50 / 上限 200）；排序 created_at DESC；type 过滤 - summary：卡片汇总（含 cache_tokens 的 jsonb 聚合）+ 折算累加 + unpriced_count - meta（约定 22/23）：data_until 取参与 key 中非 null 的最慢值、per-key 状态、未收录 key_hint 即时生成；全未收录仍 200 空结果 - 折算链（pricing.ts）：按归属购买取尺（quota × paid_amount ÷ quota_granted = 按 quota 占比分摊实付）；wallet 路径按未折算（null）；假设 #6 的多条匹配规则 - 修复 jsonb 双重编码（实测定位）：drizzle + bun-sql 直接传 JS 对象会落成 jsonb string，SQL 侧 ->> / ? 全失效 → 统一 toJsonb()（sql`${obj}::jsonb`），影响 other 与 model_limits - seed 调整为「为行数最多的订阅登记购买」，保留未折算展示样本 - 实测：jsonb_typeof=object、SQL 聚合恢复、金额与 unpriced_count 计数正确、data_until 取最慢 --- apps/server/scripts/seed.ts | 57 +++- apps/server/src/lib/pricing.test.ts | 79 ++++++ apps/server/src/lib/pricing.ts | 72 +++++ apps/server/src/lib/upstream-columns.ts | 4 +- apps/server/src/routes/usage.ts | 353 ++++++++++++++++++++++-- apps/server/src/sync/log-row.ts | 15 +- docs/design/实施计划.md | 6 +- 7 files changed, 546 insertions(+), 40 deletions(-) create mode 100644 apps/server/src/lib/pricing.test.ts create mode 100644 apps/server/src/lib/pricing.ts diff --git a/apps/server/scripts/seed.ts b/apps/server/scripts/seed.ts index ff8641c..2c57aad 100644 --- a/apps/server/scripts/seed.ts +++ b/apps/server/scripts/seed.ts @@ -8,7 +8,8 @@ * 查询面可用该常量 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`（订阅路径；**金额为占位值**，用于验证折算链） + * 4. 一条示例 `purchases`：为样例中**日志行数最多**的订阅登记（**金额为占位值**）； + * 其余订阅保持未登记——保留「未折算（`amount_rmb = null`）」的展示样本 * * 注：快照文件未入库（`.gitignore` 的 `/api*.json`），本脚本只在本地开发环境使用。 */ @@ -20,8 +21,9 @@ 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 { readSubscriptionId } from '../src/lib/pricing' import { fromUnixSeconds } from '../src/lib/time' -import { parseOther, toLogRow } from '../src/sync/log-row' +import { parseOther, toJsonb, toLogRow } from '../src/sync/log-row' import type { UpstreamLogRecord } from '../src/upstream/types' /** 假 key 常量：查看面 / 管理面用它做本地验证（可用 env SEED_KEY 覆盖） */ @@ -73,7 +75,7 @@ if (!keyRow) { totalUsed, totalAvailable: totalGranted - totalUsed, unlimitedQuota: false, - modelLimits: {}, + modelLimits: toJsonb({}), modelLimitsEnabled: false, expiresAt: null, }) @@ -103,12 +105,38 @@ if (!qpuRow) { } // ---------- 4. 示例 purchases（金额为占位值） ---------- +// 样例快照里可能有多个订阅（实测 89 / 90 两组）：只为行数最多的那笔登记购买记录， +// 其余保持未登记——既保留「未折算」样本，也贴近真实的「多订阅并存」形状 +interface SubInfo { + count: number + earliest: number + total: number + plan: string | null +} +const bySubscription = new Map() +for (const log of logs) { + const other = parseOther(log.other) + const sid = readSubscriptionId(other) + if (sid === null) continue + const cur = + bySubscription.get(sid) ?? + ({ count: 0, earliest: Number.POSITIVE_INFINITY, total: 0, plan: null } satisfies SubInfo) + cur.count += 1 + cur.earliest = Math.min(cur.earliest, log.created_at) + cur.total = Math.max(cur.total, asNumber(other.subscription_total, 0)) + cur.plan = typeof other.subscription_plan_title === 'string' ? other.subscription_plan_title : cur.plan + bySubscription.set(sid, cur) +} +const ranked = [...bySubscription.entries()].sort((a, b) => b[1].count - a[1].count) + 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) +if (purchaseRow) { + console.log(`[seed] purchases：已有数据（id=${purchaseRow.id}），跳过`) +} else if (ranked.length === 0) { + console.log('[seed] purchases：样例日志中无 subscription_id，跳过') +} else { + const [mainSid, info] = ranked[0]! + const earliest = fromUnixSeconds(info.earliest) await db.insert(purchases).values({ kind: 'subscription', paidAt: earliest, @@ -116,12 +144,15 @@ if (!purchaseRow) { note: 'seed 示例（金额为占位值，非真实）', periodStart: earliest, periodEnd: null, - subscriptionId: asNumber(other.subscription_id, 90), - quotaGranted: asNumber(other.subscription_total, 800_000_000), + subscriptionId: mainSid, + quotaGranted: info.total > 0 ? info.total : 800_000_000, }) - console.log('[seed] purchases：已插入示例一条（折算链验证用）') -} else { - console.log(`[seed] purchases：已有数据（id=${purchaseRow.id}），跳过`) + console.log( + `[seed] purchases：为订阅 #${mainSid}（${info.plan ?? '未知计划'}，${info.count} 行日志）插入示例一条`, + ) + for (const [sid, extra] of ranked.slice(1)) { + console.log(`[seed] 订阅 #${sid}（${extra.count} 行）保持未登记——用于演示「未折算」展示`) + } } console.log(`[seed] 完成。查询面可用假 key「${SEED_KEY}」命中已登记 key。`) diff --git a/apps/server/src/lib/pricing.test.ts b/apps/server/src/lib/pricing.test.ts new file mode 100644 index 0000000..4a93425 --- /dev/null +++ b/apps/server/src/lib/pricing.test.ts @@ -0,0 +1,79 @@ +/** + * 折算链单测（purchases 表设计 §1 · 实施计划「实现假设」#6）。 + */ +import { describe, expect, test } from 'bun:test' + +import { amountRmbFor, resolvePurchaseForLog, type PurchaseLite } from './pricing' + +const d = (s: string) => new Date(s) + +const P = (over: Partial = {}): PurchaseLite => ({ + id: 1, + paidAmount: 200, + quotaGranted: 1_000_000_000, + subscriptionId: 90, + periodStart: d('2026-09-01T00:00:00+08:00'), + periodEnd: d('2026-10-01T00:00:00+08:00'), + ...over, +}) + +describe('resolvePurchaseForLog（日志 → 归属购买）', () => { + test('同 ID 唯一 → 该条（期间内）', () => { + const p = P() + expect(resolvePurchaseForLog(d('2026-09-10T12:00:00+08:00'), 90, 'subscription', [p])).toBe(p) + }) + + test('充值（wallet）→ null（池子近似后议，先按未折算）', () => { + expect(resolvePurchaseForLog(d('2026-09-10T12:00:00+08:00'), 90, 'wallet', [P()])).toBeNull() + }) + + test('无 subscription_id → null', () => { + expect(resolvePurchaseForLog(d('2026-09-10T12:00:00+08:00'), null, 'subscription', [P()])).toBeNull() + }) + + test('同 ID 多条：取「期间覆盖日志时刻」的那条（跨期连续两笔）', () => { + const p1 = P({ id: 1, periodStart: d('2026-08-01T00:00:00+08:00'), periodEnd: d('2026-09-01T00:00:00+08:00') }) + const p2 = P({ id: 2, periodStart: d('2026-09-01T00:00:00+08:00'), periodEnd: d('2026-10-01T00:00:00+08:00') }) + expect(resolvePurchaseForLog(d('2026-09-10T00:00:00+08:00'), 90, 'subscription', [p1, p2])?.id).toBe(2) + expect(resolvePurchaseForLog(d('2026-08-10T00:00:00+08:00'), 90, 'subscription', [p1, p2])?.id).toBe(1) + }) + + test('同 ID 多条且都不覆盖 → null（不猜）', () => { + const p1 = P({ id: 1, periodStart: d('2026-07-01T00:00:00+08:00'), periodEnd: d('2026-08-01T00:00:00+08:00') }) + const p2 = P({ id: 2, periodStart: d('2026-08-01T00:00:00+08:00'), periodEnd: d('2026-09-01T00:00:00+08:00') }) + expect(resolvePurchaseForLog(d('2026-09-10T00:00:00+08:00'), 90, 'subscription', [p1, p2])).toBeNull() + }) + + test('期间边界：左闭右开（period_end 时刻本身不属于该笔）', () => { + const p = P({ periodEnd: d('2026-09-01T00:00:00+08:00') }) + expect(resolvePurchaseForLog(d('2026-08-31T23:59:59+08:00'), 90, 'subscription', [p])).toBe(p) + // 同 ID 唯一时不做期间判定（唯一即兜底），故用多条场景验证边界 + const p2 = P({ id: 2, periodStart: d('2026-09-01T00:00:00+08:00'), periodEnd: null }) + expect(resolvePurchaseForLog(d('2026-09-01T00:00:00+08:00'), 90, 'subscription', [p, p2])?.id).toBe(2) + }) +}) + +describe('amountRmbFor（quota 占比 × 该笔实付）', () => { + test('正常折算：quota × paid_amount ÷ quota_granted', () => { + // 1e6 quota / 1e9 总额 × 200 元 = 0.2 元 + expect(amountRmbFor(1_000_000, P())).toBeCloseTo(0.2, 10) + }) + + test('错误行 quota = 0 → 0（有归属时）', () => { + expect(amountRmbFor(0, P())).toBe(0) + }) + + test('无归属购买 → null（未登记 / 充值来源）', () => { + expect(amountRmbFor(1234, null)).toBeNull() + }) + + test('quota_granted 缺失或非正 → null（分母不成立）', () => { + expect(amountRmbFor(1234, P({ quotaGranted: null }))).toBeNull() + expect(amountRmbFor(1234, P({ quotaGranted: 0 }))).toBeNull() + }) + + test('全精度：不做舍入（约定 14）', () => { + const v = amountRmbFor(702, P({ paidAmount: 168.5, quotaGranted: 800_000_000 })) + expect(v).toBe((702 * 168.5) / 800_000_000) + }) +}) diff --git a/apps/server/src/lib/pricing.ts b/apps/server/src/lib/pricing.ts new file mode 100644 index 0000000..c66ab86 --- /dev/null +++ b/apps/server/src/lib/pricing.ts @@ -0,0 +1,72 @@ +/** + * 折算链（`purchases` 表设计 §1 · `qpu_history` 表设计 §3.3 · API 契约 §1.4/§2.1）： + * + * 日志.quota ÷ 购买时刻的 qpu = 额度美元；× 购买汇率 = 实付人民币 + * 购买汇率 = paid_amount ÷ (quota_granted ÷ 购买时刻的 qpu) + * + * 两处 qpu 同为「该日志所属购买的购买时刻」取值，代数上抵消，落到逐行计算即： + * + * amount_rmb = quota × paid_amount ÷ quota_granted （按 quota 占比分摊该笔实付） + * + * 这正是设计所称「跨期按购买期间边界分段，每段 quota 占比 × 该笔实付」的逐行等价形式 + * ——每一行都按自己的归属购买取尺，跨期自然分段；`qpu_history` 不参与金额计算。 + */ +export interface PurchaseLite { + id: number + paidAmount: number + quotaGranted: number | null + subscriptionId: number | null + periodStart: Date | null + periodEnd: Date | null +} + +/** 期间覆盖判定：`[period_start, period_end)` 左闭右开；两端都可为空（未填 = 不设限） */ +function covers(p: PurchaseLite, t: Date): boolean { + return ( + (!p.periodStart || t.getTime() >= p.periodStart.getTime()) && + (!p.periodEnd || t.getTime() < p.periodEnd.getTime()) + ) +} + +/** + * 日志 → 归属购买（实施计划「实现假设」#6：`subscription_id` 优先；同 ID 多条按 + * `[period_start, period_end)`；无期间记录仅在「同 ID 唯一」时兜底）。 + * 充值（`billing_source = "wallet"`）路径的池子近似后议——先按「未折算」处理（返回 null）。 + */ +export function resolvePurchaseForLog( + logTime: Date, + subscriptionId: number | null, + billingSource: unknown, + purchases: PurchaseLite[], +): PurchaseLite | null { + if (billingSource === 'wallet') return null + if (subscriptionId === null) return null + + const same = purchases.filter((p) => p.subscriptionId === subscriptionId) + if (same.length === 0) return null + if (same.length === 1) return same[0]! + + // 同 ID 多条（续费 / 跨期）：只认「期间覆盖该日志时刻」者；都不覆盖则不猜（null） + const inPeriod = same.filter((p) => covers(p, logTime)) + if (inPeriod.length === 0) return null + return [...inPeriod].sort( + (a, b) => (b.periodStart?.getTime() ?? 0) - (a.periodStart?.getTime() ?? 0), + )[0]! +} + +/** + * 单行折算：无归属购买（未登记 / 充值来源 / 无法判定）→ `null`； + * 有归属但 `quota_granted` 缺失 / 非正 → 折算分母不成立 → `null`（展示「—」，计入 `unpriced_count`）。 + */ +export function amountRmbFor(quota: number, purchase: PurchaseLite | null): number | null { + if (!purchase) return null + if (purchase.quotaGranted === null || purchase.quotaGranted <= 0) return null + return (quota * purchase.paidAmount) / purchase.quotaGranted +} + +/** `other.subscription_id` 的宽容读取（jsonb 里可能是数字或字符串） */ +export function readSubscriptionId(other: Record): number | null { + const raw = other.subscription_id + const n = typeof raw === 'number' ? raw : typeof raw === 'string' ? Number(raw) : NaN + return Number.isFinite(n) ? n : null +} diff --git a/apps/server/src/lib/upstream-columns.ts b/apps/server/src/lib/upstream-columns.ts index fef9b6d..18a1ab0 100644 --- a/apps/server/src/lib/upstream-columns.ts +++ b/apps/server/src/lib/upstream-columns.ts @@ -2,6 +2,7 @@ * 上游字段全组（`api_keys` 表 8 个上游列）——登记 / 启用 / 每轮额度采集共用一处， * 避免三处各写一份导致漏列（api_keys 表设计 §3.4/§3.5：整组一起写）。 */ +import { toJsonb } from '../sync/log-row' import type { UpstreamTokenUsage } from '../upstream/types' import { expiresAtFromUnix } from './time' @@ -12,7 +13,8 @@ export function upstreamColumns(usage: UpstreamTokenUsage) { totalUsed: usage.total_used, totalAvailable: usage.total_available, unlimitedQuota: usage.unlimited_quota, - modelLimits: usage.model_limits, + // jsonb 列需显式 ::jsonb（见 log-row.ts 文件头注释的写入陷阱） + modelLimits: toJsonb(usage.model_limits), modelLimitsEnabled: usage.model_limits_enabled, expiresAt: expiresAtFromUnix(usage.expires_at), } diff --git a/apps/server/src/routes/usage.ts b/apps/server/src/routes/usage.ts index aa80e02..65c2df5 100644 --- a/apps/server/src/routes/usage.ts +++ b/apps/server/src/routes/usage.ts @@ -1,4 +1,25 @@ import { Hono } from 'hono' +import { and, desc, eq, gte, inArray, lt, sql } from 'drizzle-orm' +import type { Context } from 'hono' +import { + KEYS_MAX, + LOGS_DEFAULT_LIMIT, + LOGS_MAX_LIMIT, + type ApiKeyRow as ApiKeyRowContract, +} from '@naku/shared' +import type { LogOther, LogType, QueryMeta, QueryMetaKey, SummaryResponse, UsageLogRow } from '@naku/shared' + +import { apiError } from '../lib/api-error' +import { getKeySecrets } from '../lib/keys' +import { + amountRmbFor, + readSubscriptionId, + resolvePurchaseForLog, + type PurchaseLite, +} from '../lib/pricing' +import { parseInputTime, toBeijingIso } from '../lib/time' +import { getDb } from '../db/client' +import { apiKeys, purchases, usageLogs } from '../db/schema' /** * 查看面（契约 §2.1）：无会话、无服务端状态——key 集合与过滤参数全在请求体。 @@ -6,26 +27,314 @@ import { Hono } from 'hono' */ export const usageRoutes = new Hono() -usageRoutes.post('/logs/query', (c) => - c.json( - { - error: { - code: 'internal', - message: '未实现：明细查询（见 docs/design/API契约.md §2.1）', - }, - }, - 501, - ), -) +type ApiKeyRow = typeof apiKeys.$inferSelect +type UsageLogDbRow = typeof usageLogs.$inferSelect -usageRoutes.post('/summary', (c) => - c.json( - { - error: { - code: 'internal', - message: '未实现：卡片汇总（见 docs/design/API契约.md §2.1）', - }, - }, - 501, - ), -) +// ---------- 请求解析 ---------- + +async function readJsonObject( + c: Context, +): Promise<{ error: string } | { value: Record }> { + let raw: unknown + try { + raw = await c.req.json() + } catch { + return { error: '请求体必须是 JSON' } + } + if (!raw || typeof raw !== 'object' || Array.isArray(raw)) { + return { error: '请求体必须是 JSON 对象' } + } + return { value: raw as Record } +} + +interface QueryInput { + keys: string[] + start: Date + end: Date +} + +/** 公共参数（keys / start / end）——两个端点同一套（约定 17/19 + §2.1 参数清单） */ +function parseBase(body: Record): { error: string } | { input: QueryInput } { + const rawKeys = body.keys + if (!Array.isArray(rawKeys) || rawKeys.length === 0) { + return { error: 'keys 必填（非空数组）' } + } + const keys = [ + ...new Set( + rawKeys + .filter((k): k is string => typeof k === 'string') + .map((k) => k.trim()) + .filter((k) => k.length > 0), + ), + ] + if (keys.length === 0) return { error: 'keys 必填（非空数组）' } + if (keys.length > KEYS_MAX) { + return { error: `keys 数量超过上限（去重后最多 ${KEYS_MAX} 个）` } + } + + const start = typeof body.start === 'string' ? parseInputTime(body.start) : null + if (!start) return { error: 'start 缺失或无法解析（ISO 8601）' } + const end = typeof body.end === 'string' ? parseInputTime(body.end) : null + if (!end) return { error: 'end 缺失或无法解析（ISO 8601）' } + if (start.getTime() >= end.getTime()) { + return { error: 'start 必须早于 end（左闭右开）' } + } + return { input: { keys, start, end } } +} + +// ---------- key 命中与 meta（约定 5 / 22 / 23） ---------- + +interface ResolvedKey { + plain: string + row: ApiKeyRow | null +} + +/** 未收录 key 的即时提示：`sk-…a1b2`（即时生成、不落库——api_keys 表设计 §5） */ +function keyHint(plainKey: string): string { + const tail = plainKey.slice(-4) + return plainKey.length <= 8 ? `…${tail}` : `${plainKey.slice(0, 3)}…${tail}` +} + +async function resolveKeys(keys: string[]): Promise { + const secrets = getKeySecrets() + const hashes = keys.map((k) => secrets.keyHash(k)) + const rows = await getDb().select().from(apiKeys).where(inArray(apiKeys.keyHash, hashes)) + const byHash = new Map(rows.map((r) => [r.keyHash, r])) + return keys.map((plain, i) => ({ plain, row: byHash.get(hashes[i]!) ?? null })) +} + +/** meta：`data_until` 取参与 key 中「非 null 的最慢 `log_synced_at`」；全部未同步时为 null */ +function buildMeta(resolved: ResolvedKey[]): QueryMeta { + const keys: QueryMetaKey[] = resolved.map(({ plain, row }) => + row + ? ({ + key_id: row.id, + token_name: row.tokenName, + registered: true, + status: row.status as ApiKeyRowContract['status'], + log_synced_at: row.logSyncedAt ? toBeijingIso(row.logSyncedAt) : null, + } satisfies QueryMetaKey) + : ({ registered: false, key_hint: keyHint(plain) } satisfies QueryMetaKey), + ) + const syncedMs = resolved + .map((r) => r.row?.logSyncedAt) + .filter((d): d is Date => d instanceof Date) + .map((d) => d.getTime()) + return { + data_until: syncedMs.length > 0 ? toBeijingIso(new Date(Math.max(...syncedMs))) : null, + served_at: toBeijingIso(new Date()), + keys, + } +} + +// ---------- 行映射与折算 ---------- + +async function loadPurchases(): Promise { + return await getDb() + .select({ + id: purchases.id, + paidAmount: purchases.paidAmount, + quotaGranted: purchases.quotaGranted, + subscriptionId: purchases.subscriptionId, + periodStart: purchases.periodStart, + periodEnd: purchases.periodEnd, + }) + .from(purchases) +} + +/** 明细行 → 契约行：表 1:1（除 `dedupe_key`）+ `token_name` 取当前名 + `amount_rmb` 折算 */ +function toUsageLogRow( + log: UsageLogDbRow, + currentTokenName: string, + allPurchases: PurchaseLite[], +): UsageLogRow { + const other = (log.other ?? {}) as LogOther + const purchase = resolvePurchaseForLog( + log.createdAt, + readSubscriptionId(other), + other.billing_source, + allPurchases, + ) + return { + id: log.id, + key_id: log.keyId, + user_id: log.userId, + username: log.username, + created_at: toBeijingIso(log.createdAt), + type: log.type as LogType, + content: log.content, + token_name: currentTokenName, + model_name: log.modelName, + quota: log.quota, + prompt_tokens: log.promptTokens, + completion_tokens: log.completionTokens, + use_time: log.useTime, + is_stream: log.isStream, + channel: log.channel, + channel_name: log.channelName, + token_id: log.tokenId, + group_name: log.groupName, + ip: log.ip, + request_id: log.requestId, + upstream_request_id: log.upstreamRequestId, + amount_rmb: amountRmbFor(log.quota, purchase), + other, + } +} + +// ---------- 端点 ---------- + +/** 明细（分页；排序固定 created_at DESC——约定 16） */ +usageRoutes.post('/logs/query', async (c) => { + const body = await readJsonObject(c) + if ('error' in body) return apiError(c, 400, 'invalid_params', body.error) + const parsed = parseBase(body.value) + if ('error' in parsed) return apiError(c, 400, 'invalid_params', parsed.error) + const { keys, start, end } = parsed.input + + let types: number[] | null = null + if (body.value.type !== undefined) { + const raw = body.value.type + if ( + !Array.isArray(raw) || + raw.length === 0 || + raw.some((t) => !Number.isInteger(t) || (t as number) < 0 || (t as number) > 7) + ) { + return apiError(c, 400, 'invalid_params', 'type 必须是非空的 0–7 整数数组') + } + types = [...new Set(raw as number[])] + } + const limit = body.value.limit === undefined ? LOGS_DEFAULT_LIMIT : Number(body.value.limit) + if (!Number.isInteger(limit) || limit <= 0) { + return apiError(c, 400, 'invalid_params', 'limit 必须是正整数') + } + if (limit > LOGS_MAX_LIMIT) { + return apiError(c, 400, 'invalid_params', `limit 超过上限（最多 ${LOGS_MAX_LIMIT}）`) + } + const offset = body.value.offset === undefined ? 0 : Number(body.value.offset) + if (!Number.isInteger(offset) || offset < 0) { + return apiError(c, 400, 'invalid_params', 'offset 必须是非负整数') + } + + const db = getDb() + const resolved = await resolveKeys(keys) + const keyIds = resolved.filter((r) => r.row !== null).map((r) => r.row!.id) + const meta = buildMeta(resolved) + + // 全部未收录：200 + 空结果（约定 23——未收录是结果、不是错误） + if (keyIds.length === 0) { + return c.json({ items: [], total: 0, meta }) + } + + const conditions = [ + inArray(usageLogs.keyId, keyIds), + gte(usageLogs.createdAt, start), + lt(usageLogs.createdAt, end), + ] + if (types) conditions.push(inArray(usageLogs.type, types)) + const where = and(...conditions) + + const totalRows = await db + .select({ total: sql`count(*)::int` }) + .from(usageLogs) + .where(where) + const rows = await db + .select({ log: usageLogs, currentTokenName: apiKeys.tokenName }) + .from(usageLogs) + .innerJoin(apiKeys, eq(usageLogs.keyId, apiKeys.id)) + .where(where) + .orderBy(desc(usageLogs.createdAt)) + .limit(limit) + .offset(offset) + + const allPurchases = await loadPurchases() + const items = rows.map(({ log, currentTokenName }) => + toUsageLogRow(log, currentTokenName, allPurchases), + ) + return c.json({ items, total: Number(totalRows[0]?.total ?? 0), meta }) +}) + +/** 汇总（卡片；无 type / limit / offset） */ +usageRoutes.post('/summary', async (c) => { + const body = await readJsonObject(c) + if ('error' in body) return apiError(c, 400, 'invalid_params', body.error) + const parsed = parseBase(body.value) + if ('error' in parsed) return apiError(c, 400, 'invalid_params', parsed.error) + const { keys, start, end } = parsed.input + + const db = getDb() + const resolved = await resolveKeys(keys) + const keyIds = resolved.filter((r) => r.row !== null).map((r) => r.row!.id) + const meta = buildMeta(resolved) + + const empty: SummaryResponse = { + request_count: 0, + consume_count: 0, + error_count: 0, + prompt_tokens: 0, + completion_tokens: 0, + cache_tokens: 0, + quota: 0, + amount_rmb: 0, + unpriced_count: 0, + meta, + } + if (keyIds.length === 0) return c.json(empty) + + const where = and( + inArray(usageLogs.keyId, keyIds), + gte(usageLogs.createdAt, start), + lt(usageLogs.createdAt, end), + ) + + const aggRows = await db + .select({ + request_count: sql`count(*)::int`, + consume_count: sql`count(*) filter (where ${usageLogs.type} = 2)::int`, + error_count: sql`count(*) filter (where ${usageLogs.type} = 5)::int`, + prompt_tokens: sql`coalesce(sum(${usageLogs.promptTokens}), 0)::bigint`, + completion_tokens: sql`coalesce(sum(${usageLogs.completionTokens}), 0)::bigint`, + cache_tokens: sql`coalesce(sum((${usageLogs.other} ->> 'cache_tokens')::bigint), 0)::bigint`, + quota: sql`coalesce(sum(${usageLogs.quota}), 0)::bigint`, + }) + .from(usageLogs) + .where(where) + const agg = aggRows[0] + + // 折算：按行取归属购买（跨期自然分段）；`unpriced_count` = 无法折算的行数 + const pricingRows = await db + .select({ quota: usageLogs.quota, createdAt: usageLogs.createdAt, other: usageLogs.other }) + .from(usageLogs) + .where(where) + const allPurchases = await loadPurchases() + let amountRmb = 0 + let unpricedCount = 0 + for (const r of pricingRows) { + const other = (r.other ?? {}) as LogOther + const amount = amountRmbFor( + r.quota, + resolvePurchaseForLog( + r.createdAt, + readSubscriptionId(other), + other.billing_source, + allPurchases, + ), + ) + if (amount === null) unpricedCount++ + else amountRmb += amount + } + + const response: SummaryResponse = { + request_count: Number(agg?.request_count ?? 0), + consume_count: Number(agg?.consume_count ?? 0), + error_count: Number(agg?.error_count ?? 0), + prompt_tokens: Number(agg?.prompt_tokens ?? 0), + completion_tokens: Number(agg?.completion_tokens ?? 0), + cache_tokens: Number(agg?.cache_tokens ?? 0), + quota: Number(agg?.quota ?? 0), + amount_rmb: amountRmb, + unpriced_count: unpricedCount, + meta, + } + return c.json(response) +}) diff --git a/apps/server/src/sync/log-row.ts b/apps/server/src/sync/log-row.ts index a80c69d..a52499a 100644 --- a/apps/server/src/sync/log-row.ts +++ b/apps/server/src/sync/log-row.ts @@ -1,12 +1,18 @@ /** * 上游日志行 → `usage_logs` 列值（同步引擎与 seed 脚本共用，避免两份映射漂移）。 + * + * ⚠️ jsonb 写入陷阱（2026-09-15 实测）：drizzle + `bun-sql` 组合下直接传 JS 对象会 + * **双重编码**（落库成 `jsonb` 的 *string* 值，`other->>'k'` / `other ? 'k'` 全部失效）。 + * 正确写法是 `sql\`${obj}::jsonb\``——由驱动序列化对象一次并由 PG 解析为 object。 */ +import { sql } from 'drizzle-orm' + import { fromUnixSeconds } from '../lib/time' import type { usageLogs } from '../db/schema' import type { UpstreamLogRecord } from '../upstream/types' import { dedupeKeyFor } from './dedupe' -/** 上游 `other`（JSON 字符串）→ jsonb；解析失败退化为空对象（保真优先，不丢整条日志） */ +/** 上游 `other`（JSON 字符串）→ jsonb 值；解析失败退化为空对象（保真优先，不丢整条日志） */ export function parseOther(raw: string | undefined | null): Record { if (!raw) return {} try { @@ -17,6 +23,11 @@ export function parseOther(raw: string | undefined | null): Record>'k'` / `? 'k'` 全部失效（2026-09-15 实测定位） | 已在 `other` / `model_limits` 落地；新增 jsonb 列时照此办理 | +| 18 | seed 为「日志行数最多的订阅」登记示例购买，其余订阅保持未登记（保留 `amount_rmb = null` 的展示样本） | 样例实测含两个订阅（89 / 90）；可改 | ## 2. 前提（2026-09-14 定） @@ -58,7 +60,7 @@ | 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`） | ✅ 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 定） | ☐ | +| 5 | **查看面数据面**：`logs/query` + `summary` + 折算链（按购买期间分段取尺） | curl 实测：字段齐全、`meta.data_until` 取最慢、金额按购买分段累加；**无归属购买的行 `amount_rmb = null` 且计入 `unpriced_count`**（2026-09-14 定） | ✅ 2026-09-15（56 项单测绿；curl 17 步——明细字段齐全 / 分页 / type 过滤 / 参数校验 6 类 400 / 未收录 200 空结果 / `data_until` 取最慢 / 金额按归属购买折算、未折算行 `null` + `unpriced_count` 计数正确） | | 6 | **前端 `/usage-log`**：筛选区（key 标签 + datetimerange + 快捷项）· 卡片 · 表格 + 展开行 · 分页器 · 六种状态模式 | 浏览器实测：seed 数据走一遍；六态逐一可见（加载 / 初始 / 未收录 / 无记录 / 同步中 / 失败）；金额为 `null` 显示「—」 | ☐ | | 7 | **前端管理面**：口令门帘 + 侧边栏 + 三页（key 管理 / 购买记录含核对提示与补录 / 换算参数） | 浏览器实测：登记、启停、CRUD、补录、时间线全操作 | ☐ | | 8 | **收尾**：部署文档（`docs/delivery/`）· 端到端走查 · 文档状态更新（设计文档 → 已实现） | 一条命令链可部署（前后端独立 + 后端单实例注记）；走查无阻 | ☐ |