Skip to content

fix(sharing): count the points spent through a key, not its tokens, in keys.used #217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
argszero merged 1 commit into from
Sep 13, 2026
Merged

fix(sharing): count the points spent through a key, not its tokens, in keys.used #217

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions docs/architecture.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 中英双语） |
Expand Down Expand Up @@ -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 一览

Expand Down
122 changes: 116 additions & 6 deletions src/billing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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）：
Expand Down Expand Up @@ -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()?;
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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, &params).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);
Expand Down
163 changes: 160 additions & 3 deletions src/db.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Connection> {
Expand Down Expand Up @@ -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)",
Expand All @@ -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<usize> {
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");
Expand Down Expand Up @@ -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 性能索引在迁移时建好
Expand Down
2 changes: 2 additions & 0 deletions src/routes/sharing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
Expand Down
Loading