diff --git a/docs/architecture.md b/docs/architecture.md index 5af36d9..05c8d37 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -16,7 +16,7 @@ | 层 | 选型 | |---|---| | 后端 | **Rust**(`rust-version 1.86`)+ axum + tokio + rusqlite | -| 数据库 | **SQLite**(单文件,`data/aitokenpool.db`,幂等迁移至 v12,见 `src/db.rs` 的 `SCHEMA_VERSION`) | +| 数据库 | **SQLite**(单文件,`data/aitokenpool.db`,幂等迁移,版本见 `src/db.rs` 的 `SCHEMA_VERSION`) | | 加密 | AES-256-GCM(上游 key,`src/crypto.rs`)、argon2(密码哈希) | | 上游调用 | reqwest(非流式)+ SSE 流式转发(`src/sse.rs` 跨协议转换) | | 前端 | **原生 JS** 静态页(`ui/`,无构建步骤;i18n 中英双语) | @@ -61,7 +61,7 @@ | `usage_records` | 调用明细(tokens 拆 input / cached / output) | | `departments` / `raise_requests` | 部门 + 成员加额申请(企业版) | | `email_verifications` | 注册邮箱验证码(与 `users.verified` 配套,v6 起) | -| `schema_version` | 迁移记录(当前 v12,见 `src/db.rs` 的 `SCHEMA_VERSION`) | +| `schema_version` | 迁移记录(当前版本见 `src/db.rs` 的 `SCHEMA_VERSION`) | ## 6. API 一览 diff --git a/src/billing.rs b/src/billing.rs index 9f77f6c..fb384ba 100644 --- a/src/billing.rs +++ b/src/billing.rs @@ -6,7 +6,7 @@ //! - 点数 = 锚定货币成本 × points_per_unit //! - 消费者扣 balance(上游调用前预检,余额 ≤ 0 → 402) //! - 分享者(key 属主)得 90%(平台抽成 10%),写 transactions(consume / earn) -//! - 写 usage_records;更新 keys.used += tokens +//! - 写 usage_records;更新 keys.used += pts(该 key 已消耗的**点数**,与 keys.quota 同单位) //! - 调用+记账事务性处理:上游失败不入账(settle 只在成功响应后调用) //! //! P1(rant 2026-08-18T11:03:02): @@ -144,7 +144,8 @@ pub struct SettleParams { } /// 事务性入账:扣消费者(先赠送后永久)→ 加分享者 90% → 两条 transactions → -/// usage_records → keys.used。任一步失败整体回滚(调用方只在成功响应后调用, +/// usage_records → keys.used(按 `p.pts` 累计**点数**,与 `keys.quota` 同单位)。任一步失败整体回滚 +/// (调用方只在成功响应后调用, /// 天然满足「失败不入账」) pub fn settle(conn: &mut Connection, p: &SettleParams) -> Result<()> { let tx = conn.transaction()?; @@ -216,10 +217,13 @@ pub fn settle(conn: &mut Connection, p: &SettleParams) -> Result<()> { ], )?; - // keys.used 累计(key 不存在 → 报错回滚:账本不允许记到幽灵 key) + // keys.used 累计:记该 key 已消耗的**点数**(与 keys.quota 同单位 —— 共享页把两者相除画 + // 进度条并按「点」渲染,见 ui/js/app.js);token 计数另存 transactions.tokens / + // usage_records.tokens,不再进这一列。 + // key 不存在 → 报错回滚:账本不允许记到幽灵 key let n = tx.execute( "UPDATE keys SET used = used + ?1 WHERE id = ?2", - rusqlite::params![p.tokens, p.key_id], + rusqlite::params![p.pts, p.key_id], )?; if n != 1 { return Err(anyhow::anyhow!("key {} not found", p.key_id)); @@ -517,11 +521,117 @@ mod tests { .unwrap(); assert!((cached_u - 30.0).abs() < 1e-9, "usage cached={cached_u}"); assert!((output_u - 20.0).abs() < 1e-9, "usage output={output_u}"); - // keys.used 更新 + // keys.used 记录的是**消耗的点数**(= 本笔 consume 的 pts),不是 token 数(150.0)—— + // 该列与 keys.quota 同单位,共享页据此渲染「已用 / 额度」。 let used: f64 = conn .query_row("SELECT used FROM keys WHERE id = 9", [], |r| r.get(0)) .unwrap(); - assert!((used - 150.0).abs() < 1e-9); + assert!( + (used - 2.0).abs() < 1e-9, + "keys.used 应为点数 2.0(而非 token 150.0),实际 {used}" + ); + let consume_pts: f64 = conn + .query_row( + "SELECT COALESCE(SUM(pts), 0) FROM transactions WHERE key_id = 9 AND type = 'consume'", + [], + |r| r.get(0), + ) + .unwrap(); + assert!( + (used - consume_pts).abs() < 1e-9, + "keys.used 必须等于账本里该 key 的 consume 点数合计(used={used} ledger={consume_pts})" + ); + + drop(conn); + let _ = std::fs::remove_file(p); + } + + /// 共享页把 `keys.used` / `keys.quota` 当成同一单位(点数)相除画进度条, + /// 所以 settle 必须往这一列写**消耗的点数**。本测试让 tokens 与 pts 差 5 个数量级, + /// 并覆盖「只有 consume 行计数」这一边界(同一笔的 earn 行、以及别处写入的 topup 行都不许进来)。 + #[test] + fn keys_used_records_consumed_points_not_tokens() { + let (mut conn, p) = tmp_db("used_points"); + conn.execute( + "INSERT INTO users (id, email, password_hash, name, role) VALUES (100, 'owner2@t.local', 'x', '分享者', 'user')", + [], + ) + .unwrap(); + conn.execute("INSERT INTO quotas (user_id, balance) VALUES (100, 0)", []) + .unwrap(); + conn.execute( + "INSERT INTO keys (id, provider, plan, model, status, owner_id, encrypted_key, quota, used) \ + VALUES (9, 'test', 'test-plan', 'test-model', 'on', 2, 'sk-test', 5000, 0)", + [], + ) + .unwrap(); + + let params = SettleParams { + consumer_id: 1, + api_key_id: Some(3), + key_id: 9, + owner_id: 100, + model: "test-model".into(), + tokens: 1_000_000.0, + cached_tokens: 0.0, + output_tokens: 0.0, + pts: 4.5, + cost: 0.0045, + }; + settle(&mut conn, ¶ms).unwrap(); + + let used: f64 = conn + .query_row("SELECT used FROM keys WHERE id = 9", [], |r| r.get(0)) + .unwrap(); + // ① 记的是点数,不是 token 数 + assert!((used - 4.5).abs() < 1e-9, "keys.used={used},应为 4.5 点"); + assert!( + (used - params.tokens).abs() > 1.0, + "keys.used 不能是 token 数({})", + params.tokens + ); + // ② 与账本一致:该 key 的 consume 行合计 + let consume_pts: f64 = conn + .query_row( + "SELECT COALESCE(SUM(pts), 0) FROM transactions WHERE key_id = 9 AND type = 'consume'", + [], + |r| r.get(0), + ) + .unwrap(); + assert!( + (used - consume_pts).abs() < 1e-9, + "keys.used={used} 应等于 consume 点数合计 {consume_pts}" + ); + // ③ 阴性对照:同一笔的 earn 行(同 key_id、pts=4.05)不参与 + let all_types: f64 = conn + .query_row( + "SELECT COALESCE(SUM(pts), 0) FROM transactions WHERE key_id = 9", + [], + |r| r.get(0), + ) + .unwrap(); + assert!( + (all_types - 8.55).abs() < 1e-9, + "该 key 的 consume+earn 合计应为 8.55,实际 {all_types}" + ); + assert!( + (used - all_types).abs() > 0.1, + "keys.used 不得把 earn 行算进来(used={used} all={all_types})" + ); + // ④ 阴性对照:别处写入的其他类型(加额 topup)也不许改动这一列 + conn.execute( + "INSERT INTO transactions (user_id, key_id, model, tokens, pts, type, status) \ + VALUES (100, 9, 'recharge', 0, 777.0, 'topup', '成功')", + [], + ) + .unwrap(); + let used_after: f64 = conn + .query_row("SELECT used FROM keys WHERE id = 9", [], |r| r.get(0)) + .unwrap(); + assert!( + (used_after - 4.5).abs() < 1e-9, + "非 consume 行不得改变 keys.used,实际 {used_after}" + ); drop(conn); let _ = std::fs::remove_file(p); diff --git a/src/db.rs b/src/db.rs index e4f2b43..8c6c200 100644 --- a/src/db.rs +++ b/src/db.rs @@ -10,7 +10,7 @@ use anyhow::{Context, Result}; use rusqlite::Connection; -pub const SCHEMA_VERSION: i64 = 12; +pub const SCHEMA_VERSION: i64 = 13; /// 打开(或创建)数据库并执行幂等迁移(生产标准:空库只建表,不种任何假数据) pub fn open(path: &str) -> Result { @@ -314,10 +314,25 @@ pub fn migrate(conn: &Connection) -> Result<()> { CREATE INDEX IF NOT EXISTS idx_transactions_user_id_time ON transactions(user_id, time); CREATE INDEX IF NOT EXISTS idx_transactions_user_id_type ON transactions(user_id, type);", )?; - // schema_version:INSERT OR REPLACE 保证幂等 + // schema_version 记录**最高的**已迁移版本。⚠️ 必须用 MAX 读:该表没有唯一约束, + // 而 `INSERT OR REPLACE` 在无冲突时就是普通 INSERT —— 只读第一行的话 + // `v < SCHEMA_VERSION` 会永远为真,每次启动都追加一行(`atp-data/aitokenpool.db` + // 实测已累积 22 行:7,8,8,8,9×13,10,10),任何版本门控也随之失效。 let v: i64 = conn - .query_row("SELECT version FROM schema_version", [], |r| r.get(0)) + .query_row( + "SELECT COALESCE(MAX(version), 0) FROM schema_version", + [], + |r| r.get(0), + ) .unwrap_or(0); + // v13(共享页「已用 / 额度」单位修复):`keys.used` 的语义从「token 数」改为「点数」 + // (与 `keys.quota` 同单位)。老库只在这一步重算一次;重算值由账本唯一确定 ⇒ 幂等。 + if v < 13 { + let healed = keys_used_from_ledger(conn)?; + if healed > 0 { + log::info!("keys.used 单位迁移:按账本重算 {healed} 条 key 的已用点数"); + } + } if v < SCHEMA_VERSION { conn.execute( "INSERT OR REPLACE INTO schema_version (version) VALUES (?1)", @@ -327,6 +342,23 @@ pub fn migrate(conn: &Connection) -> Result<()> { Ok(()) } +/// v13 数据修复:把 `keys.used` 从「token 累计」重算为「该 key 消耗的**点数**」。 +/// +/// 真源是账本:`used = SUM(transactions.pts WHERE key_id = keys.id AND type = 'consume')`。 +/// 这正是修复后的 `billing::settle` 所维护的不变量(每次 settle 加一笔记为 `p.pts` 的 consume 行), +/// 所以对已修复的库重复执行是**空操作**;没有任何 consume 行的 key 归 0。 +/// 返回被更新的 key 行数。 +fn keys_used_from_ledger(conn: &Connection) -> Result { + let n = conn + .execute( + "UPDATE keys SET used = COALESCE((SELECT SUM(t.pts) FROM transactions t \ + WHERE t.key_id = keys.id AND t.type = 'consume'), 0)", + [], + ) + .with_context(|| "按账本重算 keys.used 失败".to_string())?; + Ok(n) +} + /// 幂等补列:列不存在才 ALTER TABLE ADD COLUMN fn ensure_column(conn: &Connection, table: &str, column: &str, ddl: &str) -> Result<()> { let sql = format!("SELECT 1 FROM pragma_table_info('{table}') WHERE name = ?1"); @@ -577,6 +609,131 @@ mod tests { let _ = std::fs::remove_file(p); } + #[test] + fn keys_used_is_healed_from_the_ledger_on_upgrade() { + let (conn, p) = tmp_db("heal_used"); + conn.execute( + "INSERT INTO users (id, email, password_hash, name, role) VALUES (1, 'heal@t.local', 'x', 'u', 'user')", + [], + ) + .unwrap(); + // ① key 1:列里存着旧的 token 量级(319883396),但账本里该 key 的 consume 点数合计只有 12.5 + // ② key 2:从未被调用(没有 consume 行),列里残留一个旧值 —— 应归 0 + conn.execute( + "INSERT INTO keys (id, provider, plan, model, status, owner_id, encrypted_key, quota, used) \ + VALUES (1, 'test', 'p', 'm', 'on', 1, 'sk-a', 5000, 319883396)", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO keys (id, provider, plan, model, status, owner_id, encrypted_key, quota, used) \ + VALUES (2, 'test', 'p', 'm', 'on', 1, 'sk-b', 1000, 42)", + [], + ) + .unwrap(); + for pts in [10.0, 2.5] { + conn.execute( + "INSERT INTO transactions (user_id, counterpart, key_id, model, tokens, pts, type, status) \ + VALUES (1, '2', 1, 'm', 1000000, ?1, 'consume', '成功')", + rusqlite::params![pts], + ) + .unwrap(); + } + // 同一 key 的 earn / topup 行不得计入 + conn.execute( + "INSERT INTO transactions (user_id, counterpart, key_id, model, tokens, pts, type, status) \ + VALUES (1, '1', 1, 'm', 0, 11.25, 'earn', '成功')", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO transactions (user_id, key_id, model, tokens, pts, type, status) \ + VALUES (1, 1, 'recharge', 0, 777.0, 'topup', '成功')", + [], + ) + .unwrap(); + // 伪造成修复前的库版本(迁移前是 12);该表无唯一约束,必须先清掉本库已有的 13 行 + conn.execute("DELETE FROM schema_version", []).unwrap(); + conn.execute("INSERT INTO schema_version (version) VALUES (12)", []) + .unwrap(); + + migrate(&conn).expect("v12 -> v13 迁移应成功"); + + let used1: f64 = conn + .query_row("SELECT used FROM keys WHERE id = 1", [], |r| r.get(0)) + .unwrap(); + assert!( + (used1 - 12.5).abs() < 1e-9, + "key 1 的 used 应为 consume 点数合计 12.5,实际 {used1}" + ); + let used2: f64 = conn + .query_row("SELECT used FROM keys WHERE id = 2", [], |r| r.get(0)) + .unwrap(); + assert!( + (used2 - 0.0).abs() < 1e-9, + "无 consume 账本的 key 应归 0,实际 {used2}" + ); + let v: i64 = conn + .query_row( + "SELECT COALESCE(MAX(version), 0) FROM schema_version", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(v, SCHEMA_VERSION); + + // 二次迁移不得**重跑**修复:往账本追加一笔 3 点 consume 但**不改** keys.used。 + // 若 v<13 的门控失效(重跑),used 会被账本改写成 15.5;正常运行应保持 12.5。 + conn.execute( + "INSERT INTO transactions (user_id, counterpart, key_id, model, tokens, pts, type, status) \ + VALUES (1, '2', 1, 'm', 0, 3.0, 'consume', '成功')", + [], + ) + .unwrap(); + migrate(&conn).expect("第二次迁移应成功"); + let used1b: f64 = conn + .query_row("SELECT used FROM keys WHERE id = 1", [], |r| r.get(0)) + .unwrap(); + assert!( + (used1b - 12.5).abs() < 1e-9, + "v13 之后迁移不得重跑修复(若重跑会变成账本值 15.5),实际 {used1b}" + ); + + drop(conn); + let _ = std::fs::remove_file(p); + } + + /// `schema_version` 没有唯一约束,历史启动会把同一版本追加成多行(真库里实测 22 行)。 + /// 版本门控必须按**最高**行判断,否则每次启动都会再追加一行、且所有 `v < N` 的迁移永远重跑。 + #[test] + fn schema_version_gate_uses_the_highest_recorded_version() { + let (conn, p) = tmp_db("schemaver"); + conn.execute("DELETE FROM schema_version", []).unwrap(); + for v in [7i64, 9, 9, 10] { + conn.execute("INSERT INTO schema_version (version) VALUES (?1)", [v]) + .unwrap(); + } + + migrate(&conn).expect("迁移应成功"); + let max: i64 = conn + .query_row("SELECT MAX(version) FROM schema_version", [], |r| r.get(0)) + .unwrap(); + assert_eq!(max, SCHEMA_VERSION); + let n1: i64 = conn + .query_row("SELECT COUNT(*) FROM schema_version", [], |r| r.get(0)) + .unwrap(); + + // 第二次迁移:起点已是最高版本 ⇒ 不得再追加行(按最旧行读则会每轮 +1) + migrate(&conn).expect("第二次迁移应成功"); + let n2: i64 = conn + .query_row("SELECT COUNT(*) FROM schema_version", [], |r| r.get(0)) + .unwrap(); + assert_eq!(n1, n2, "已是最新版本时重复迁移不得再追加 schema_version 行"); + + drop(conn); + let _ = std::fs::remove_file(p); + } + #[test] fn transactions_perf_indexes_created_on_migrate() { // v12(rant 2026-08-25T12:02:13):transactions 性能索引在迁移时建好 diff --git a/src/routes/sharing.rs b/src/routes/sharing.rs index cd2f4aa..8ebea0c 100644 --- a/src/routes/sharing.rs +++ b/src/routes/sharing.rs @@ -169,6 +169,8 @@ fn sharing_row( let status: String = r.get(4)?; let encrypted_key: String = r.get(5)?; let quota: f64 = r.get(6)?; + // keys.used = 该 key 已消耗的**点数**,与 `quota` 同单位(进度条 `used / quota`、卡片「已用 N 点」 + // 都按点数渲染)。由 `billing::settle` 按 `p.pts` 累加;历史库由迁移按账本 `SUM(consume pts)` 重算。 let used: f64 = r.get(7)?; let days: String = r.get(8)?; let start: String = r.get(9)?;