docs: 计费闭环实现计划(第1期详细TDD + 后5期概要)
第1期地基: credit_ledger + user_credits 表, USD 轨三处记账 (消费/充值/管理员调整)同事务接入 ledger。每步含完整测试与代码。 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,421 @@
|
||||
# 计费闭环 实现计划
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** 为 BoosAPI 建立 ledger-first 的双轨计费闭环(USD 池 + 工作流积分池),所有余额变动写统一流水账 `credit_ledger`。
|
||||
|
||||
**Architecture:** 新建 `credit_ledger`(唯一真相源)+ `user_credits`(积分缓存)+ `recharge_tiers`(充值档位)三张表。USD 轨沿用 `downstream_api_keys.max_cost/used_cost` 作热路径缓存,每笔变动同事务写一条 ledger。积分轨为工作流执行新增扣费/拦截。分 6 期,本计划详写第 1 期(地基),后 5 期为概要待细化。
|
||||
|
||||
**Tech Stack:** Rust (axum + sqlx + SQLite)、React + Vite 前端、in-memory SQLite 测试。
|
||||
|
||||
---
|
||||
|
||||
## 文件结构(第 1 期)
|
||||
|
||||
- 新建 `src/services/billing/ledger.rs` — ledger 写入 + 余额查询的唯一入口(`record_ledger_entry`、`ensure_ledger_schema`)。所有动余额的地方调它,保证"缓存 + ledger 同事务"不变量集中在一处。
|
||||
- 修改 `src/services/billing/mod.rs` — `mod ledger;` 导出;`record_usage` 在写 usage_logs + increment 后,同事务补写 ledger(usd, consume)。
|
||||
- 修改 `src/services/payment.rs` — `credit_user_balance` 加写 ledger(usd, recharge)。
|
||||
- 修改 `src/routes/admin/users.rs` — `adjust_user_balance` 加写 ledger(usd, admin_adjust)。
|
||||
- 测试:`src/services/billing/ledger.rs` 内 `#[cfg(test)] mod tests`(复用 in-memory SQLite 模式)。
|
||||
|
||||
---
|
||||
|
||||
## 第 1 期:地基(credit_ledger + user_credits 表 + USD 轨记账接入)
|
||||
|
||||
### Task 1:ledger 表结构 + 写入入口
|
||||
|
||||
**Files:**
|
||||
- Create: `src/services/billing/ledger.rs`
|
||||
- Modify: `src/services/billing/mod.rs`(顶部加 `pub mod ledger;`)
|
||||
|
||||
- [ ] **Step 1: 写失败测试** — 在 `src/services/billing/ledger.rs` 末尾:
|
||||
|
||||
```rust
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use sqlx::sqlite::SqlitePoolOptions;
|
||||
|
||||
async fn mem_pool() -> SqlitePool {
|
||||
let pool = SqlitePoolOptions::new()
|
||||
.max_connections(1)
|
||||
.connect("sqlite::memory:")
|
||||
.await
|
||||
.unwrap();
|
||||
ensure_ledger_schema(&pool).await.unwrap();
|
||||
pool
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn record_ledger_entry_writes_row_with_balance_snapshot() {
|
||||
let pool = mem_pool().await;
|
||||
let mut tx = pool.begin().await.unwrap();
|
||||
record_ledger_entry(
|
||||
&mut tx,
|
||||
LedgerEntry {
|
||||
user_id: 7,
|
||||
track: "usd",
|
||||
entry_type: "recharge",
|
||||
amount: 10.0,
|
||||
balance_after: 10.0,
|
||||
ref_type: Some("stripe"),
|
||||
ref_id: Some("cs_1"),
|
||||
model: None,
|
||||
operator: Some("system"),
|
||||
note: None,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
tx.commit().await.unwrap();
|
||||
|
||||
let row: (i64, String, String, f64, f64) = sqlx::query_as(
|
||||
"SELECT user_id, track, entry_type, amount, balance_after
|
||||
FROM credit_ledger WHERE user_id = 7",
|
||||
)
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(row, (7, "usd".into(), "recharge".into(), 10.0, 10.0));
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 运行测试确认失败**
|
||||
|
||||
Run: `cargo test -p boosapi --lib services::billing::ledger 2>&1 | tail -5`
|
||||
Expected: 编译失败(`ensure_ledger_schema`/`record_ledger_entry`/`LedgerEntry` 未定义)。
|
||||
|
||||
- [ ] **Step 3: 写最小实现** — 在 `src/services/billing/ledger.rs` 顶部(tests 之前):
|
||||
|
||||
```rust
|
||||
use crate::error::AppResult;
|
||||
use sqlx::{SqlitePool, Transaction};
|
||||
|
||||
/// One balance-changing event. Written to `credit_ledger` in the same
|
||||
/// transaction that updates the cached balance, so ledger and cache never diverge.
|
||||
pub struct LedgerEntry<'a> {
|
||||
pub user_id: i64,
|
||||
pub track: &'a str, // "usd" | "points"
|
||||
pub entry_type: &'a str, // "recharge" | "admin_adjust" | "consume" | "refund" | "grant"
|
||||
pub amount: f64, // +credit / -debit
|
||||
pub balance_after: f64,
|
||||
pub ref_type: Option<&'a str>,
|
||||
pub ref_id: Option<&'a str>,
|
||||
pub model: Option<&'a str>,
|
||||
pub operator: Option<&'a str>,
|
||||
pub note: Option<&'a str>,
|
||||
}
|
||||
|
||||
pub async fn ensure_ledger_schema(db: &SqlitePool) -> AppResult<()> {
|
||||
sqlx::query(
|
||||
"CREATE TABLE IF NOT EXISTS credit_ledger (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
|
||||
user_id INTEGER NOT NULL,
|
||||
track TEXT NOT NULL,
|
||||
entry_type TEXT NOT NULL,
|
||||
amount REAL NOT NULL,
|
||||
balance_after REAL NOT NULL,
|
||||
ref_type TEXT,
|
||||
ref_id TEXT,
|
||||
model TEXT,
|
||||
operator TEXT,
|
||||
note TEXT,
|
||||
created_at TEXT DEFAULT (datetime('now'))
|
||||
)",
|
||||
)
|
||||
.execute(db)
|
||||
.await?;
|
||||
sqlx::query(
|
||||
"CREATE INDEX IF NOT EXISTS credit_ledger_user_created_idx
|
||||
ON credit_ledger (user_id, created_at DESC)",
|
||||
)
|
||||
.execute(db)
|
||||
.await?;
|
||||
sqlx::query(
|
||||
"CREATE INDEX IF NOT EXISTS credit_ledger_track_type_idx
|
||||
ON credit_ledger (track, entry_type)",
|
||||
)
|
||||
.execute(db)
|
||||
.await?;
|
||||
sqlx::query(
|
||||
"CREATE TABLE IF NOT EXISTS user_credits (
|
||||
user_id INTEGER PRIMARY KEY NOT NULL,
|
||||
points_balance REAL NOT NULL DEFAULT 0,
|
||||
points_total_recharged REAL NOT NULL DEFAULT 0,
|
||||
updated_at TEXT DEFAULT (datetime('now'))
|
||||
)",
|
||||
)
|
||||
.execute(db)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Append one ledger row inside an existing transaction. Caller is responsible
|
||||
/// for also updating the cached balance in the same `tx`.
|
||||
pub async fn record_ledger_entry(
|
||||
tx: &mut Transaction<'_, sqlx::Sqlite>,
|
||||
entry: LedgerEntry<'_>,
|
||||
) -> AppResult<()> {
|
||||
sqlx::query(
|
||||
"INSERT INTO credit_ledger
|
||||
(user_id, track, entry_type, amount, balance_after,
|
||||
ref_type, ref_id, model, operator, note)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
)
|
||||
.bind(entry.user_id)
|
||||
.bind(entry.track)
|
||||
.bind(entry.entry_type)
|
||||
.bind(entry.amount)
|
||||
.bind(entry.balance_after)
|
||||
.bind(entry.ref_type)
|
||||
.bind(entry.ref_id)
|
||||
.bind(entry.model)
|
||||
.bind(entry.operator)
|
||||
.bind(entry.note)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
```
|
||||
|
||||
并在 `src/services/billing/mod.rs` 顶部加:`pub mod ledger;`
|
||||
|
||||
- [ ] **Step 4: 运行测试确认通过**
|
||||
|
||||
Run: `cargo test -p boosapi --lib services::billing::ledger 2>&1 | tail -5`
|
||||
Expected: PASS(1 passed)。
|
||||
|
||||
- [ ] **Step 5: 提交**
|
||||
|
||||
```bash
|
||||
git add src/services/billing/ledger.rs src/services/billing/mod.rs
|
||||
git commit -m "feat(billing): credit_ledger schema + record_ledger_entry"
|
||||
```
|
||||
|
||||
### Task 2:USD 消费接入 ledger(record_usage)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/services/billing/mod.rs`(`record_usage`,行 ~48-77)
|
||||
|
||||
- [ ] **Step 1: 写失败测试** — 在 billing/mod.rs 的 tests mod 加:
|
||||
|
||||
```rust
|
||||
#[tokio::test]
|
||||
async fn record_usage_writes_consume_ledger_entry() {
|
||||
let pool = billing_test_pool().await; // 复用现有测试 pool helper;若无则建 users+downstream_api_keys+usage_logs
|
||||
let tracker = UsageTracker::new(std::sync::Arc::new(pool.clone()));
|
||||
crate::services::billing::ledger::ensure_ledger_schema(&pool).await.unwrap();
|
||||
// key_id=7 属于 user 9,max_cost=10 used_cost=2(由 helper 建好)
|
||||
tracker.record_usage(7, "gpt-4o", 100, 50, 0.42).await.unwrap();
|
||||
|
||||
let row: (String, String, f64, Option<String>) = sqlx::query_as(
|
||||
"SELECT track, entry_type, amount, model FROM credit_ledger
|
||||
WHERE user_id = 9 ORDER BY id DESC LIMIT 1",
|
||||
).fetch_one(&pool).await.unwrap();
|
||||
assert_eq!(row.0, "usd");
|
||||
assert_eq!(row.1, "consume");
|
||||
assert_eq!(row.2, -0.42); // 消费为负
|
||||
assert_eq!(row.3.as_deref(), Some("gpt-4o"));
|
||||
}
|
||||
```
|
||||
|
||||
注:若 billing/mod.rs 无现成 `billing_test_pool`,在 tests mod 加一个建 `users`/`downstream_api_keys`(含 user_id=9,key id=7 max_cost=10 used_cost=2)/`usage_logs` 的 helper(参照 payment.rs 的 payment_test_pool)。
|
||||
|
||||
- [ ] **Step 2: 运行确认失败**
|
||||
|
||||
Run: `cargo test -p boosapi --lib record_usage_writes_consume_ledger_entry 2>&1 | tail -5`
|
||||
Expected: FAIL(credit_ledger 无 user 9 行 / 表不存在)。
|
||||
|
||||
- [ ] **Step 3: 实现** — `record_usage` 改为事务化,末尾补写 ledger。替换 record_usage 函数体为:
|
||||
|
||||
```rust
|
||||
pub async fn record_usage(
|
||||
&self,
|
||||
key_id: i64,
|
||||
model: impl Into<String>,
|
||||
prompt_tokens: i64,
|
||||
completion_tokens: i64,
|
||||
cost: f64,
|
||||
) -> AppResult<()> {
|
||||
Self::ensure_schema(&self.db).await?;
|
||||
crate::services::billing::ledger::ensure_ledger_schema(&self.db).await?;
|
||||
let prompt_tokens = prompt_tokens.max(0);
|
||||
let completion_tokens = completion_tokens.max(0);
|
||||
let total_tokens = prompt_tokens + completion_tokens;
|
||||
let cost = cost.max(0.0);
|
||||
let model = model.into();
|
||||
|
||||
let mut tx = self.db.begin().await?;
|
||||
sqlx::query(
|
||||
"INSERT INTO usage_logs
|
||||
(downstream_key_id, model, prompt_tokens, completion_tokens, total_tokens, cost)
|
||||
VALUES (?, ?, ?, ?, ?, ?)",
|
||||
)
|
||||
.bind(key_id).bind(&model)
|
||||
.bind(prompt_tokens).bind(completion_tokens).bind(total_tokens).bind(cost)
|
||||
.execute(&mut *tx).await?;
|
||||
|
||||
sqlx::query(
|
||||
"UPDATE downstream_api_keys
|
||||
SET used_requests = COALESCE(used_requests,0)+1,
|
||||
used_cost = COALESCE(used_cost,0)+?,
|
||||
last_used_at = datetime('now'), updated_at = datetime('now')
|
||||
WHERE id = ?",
|
||||
).bind(cost).bind(key_id).execute(&mut *tx).await?;
|
||||
|
||||
// Resolve owning user + new used_cost for the ledger snapshot.
|
||||
let owner: Option<(Option<i64>, Option<f64>, Option<f64>)> = sqlx::query_as(
|
||||
"SELECT user_id, max_cost, used_cost FROM downstream_api_keys WHERE id = ?",
|
||||
).bind(key_id).fetch_optional(&mut *tx).await?;
|
||||
if let Some((Some(user_id), max_cost, used_cost)) = owner {
|
||||
let used = used_cost.unwrap_or(0.0);
|
||||
let remaining = max_cost.map(|m| (m - used).max(0.0)).unwrap_or(used);
|
||||
crate::services::billing::ledger::record_ledger_entry(
|
||||
&mut tx,
|
||||
crate::services::billing::ledger::LedgerEntry {
|
||||
user_id, track: "usd", entry_type: "consume",
|
||||
amount: -cost, balance_after: remaining,
|
||||
ref_type: Some("usage_log"), ref_id: None,
|
||||
model: Some(&model), operator: Some("system"), note: None,
|
||||
},
|
||||
).await?;
|
||||
}
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
```
|
||||
|
||||
(注:原 `increment_key_usage(&self.db, ...)` 调用被事务内 UPDATE 取代,删除该行。)
|
||||
|
||||
- [ ] **Step 4: 运行确认通过**
|
||||
|
||||
Run: `cargo test -p boosapi --lib record_usage 2>&1 | tail -6`
|
||||
Expected: PASS。
|
||||
|
||||
- [ ] **Step 5: 提交**
|
||||
|
||||
```bash
|
||||
git add src/services/billing/mod.rs
|
||||
git commit -m "feat(billing): USD consume writes credit_ledger in one tx"
|
||||
```
|
||||
|
||||
### Task 3:USD 充值接入 ledger(credit_user_balance)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/services/payment.rs`(`credit_user_balance`,行 ~2167-2202)
|
||||
|
||||
- [ ] **Step 1: 写失败测试** — payment.rs tests mod(复用 payment_test_pool,user 9/key 7 max_cost=10):
|
||||
|
||||
```rust
|
||||
#[tokio::test]
|
||||
async fn credit_user_balance_writes_recharge_ledger() {
|
||||
let pool = payment_test_pool().await;
|
||||
crate::services::billing::ledger::ensure_ledger_schema(&pool).await.unwrap();
|
||||
let mut tx = pool.begin().await.unwrap();
|
||||
credit_user_balance(&mut tx, 9, 20.0).await.unwrap();
|
||||
tx.commit().await.unwrap();
|
||||
let row: (String, String, f64) = sqlx::query_as(
|
||||
"SELECT track, entry_type, amount FROM credit_ledger WHERE user_id = 9 ORDER BY id DESC LIMIT 1",
|
||||
).fetch_one(&pool).await.unwrap();
|
||||
assert_eq!(row, ("usd".into(), "recharge".into(), 20.0));
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 运行确认失败**
|
||||
|
||||
Run: `cargo test -p boosapi --lib credit_user_balance_writes_recharge_ledger 2>&1 | tail -5`
|
||||
Expected: FAIL。
|
||||
|
||||
- [ ] **Step 3: 实现** — `credit_user_balance` 在 UPDATE max_cost 之后、`Ok(Some(key_id))` 之前插入 ledger 写入(已在事务 tx 内):
|
||||
|
||||
```rust
|
||||
// After the UPDATE that sets next_max_cost on the key:
|
||||
crate::services::billing::ledger::record_ledger_entry(
|
||||
tx,
|
||||
crate::services::billing::ledger::LedgerEntry {
|
||||
user_id, track: "usd", entry_type: "recharge",
|
||||
amount: amount_usd.max(0.0),
|
||||
balance_after: (next_max_cost - used_cost).max(0.0),
|
||||
ref_type: Some("recharge_record"), ref_id: None,
|
||||
model: None, operator: Some("system"), note: None,
|
||||
},
|
||||
).await?;
|
||||
```
|
||||
|
||||
(`credit_user_balance` 已是 `tx: &mut Transaction`,直接传 `tx`。确保调用方在调用前已 `ensure_ledger_schema`,或在函数开头加一次幂等建表。)
|
||||
|
||||
- [ ] **Step 4: 运行确认通过**
|
||||
|
||||
Run: `cargo test -p boosapi --lib credit_user_balance 2>&1 | tail -6`
|
||||
Expected: PASS(含已有的 recharge completion 测试)。
|
||||
|
||||
- [ ] **Step 5: 提交**
|
||||
|
||||
```bash
|
||||
git add src/services/payment.rs
|
||||
git commit -m "feat(billing): USD recharge writes credit_ledger"
|
||||
```
|
||||
|
||||
### Task 4:USD 管理员调整接入 ledger(adjust_user_balance)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/routes/admin/users.rs`(`adjust_user_balance`,行 ~820)
|
||||
|
||||
- [ ] **Step 1: 写失败测试** — users.rs tests mod:调 `adjust_user_balance(State, Path(9), Json{delta:5.0})` 后断言 credit_ledger 有一行 `(usd, admin_adjust, 5.0)`。(建表与 user/key 同 Task 3 helper。)
|
||||
|
||||
- [ ] **Step 2: 运行确认失败** — `cargo test -p boosapi --lib adjust_user_balance 2>&1 | tail -5` → FAIL。
|
||||
|
||||
- [ ] **Step 3: 实现** — `adjust_user_balance` 现在直接 `execute(&*state.db)`。改为事务:begin → UPDATE max_cost → `record_ledger_entry(track="usd", entry_type="admin_adjust", amount=delta, balance_after=next_max-used, operator=<管理员,暂用 "admin">)` → commit。函数开头 `ledger::ensure_ledger_schema(&state.db).await?;`。
|
||||
|
||||
- [ ] **Step 4: 运行确认通过** — `cargo test -p boosapi --lib adjust_user_balance 2>&1 | tail -6` → PASS。
|
||||
|
||||
- [ ] **Step 5: 提交**
|
||||
|
||||
```bash
|
||||
git add src/routes/admin/users.rs
|
||||
git commit -m "feat(billing): USD admin adjust writes credit_ledger"
|
||||
```
|
||||
|
||||
### Task 5:全量回归
|
||||
|
||||
- [ ] **Step 1:** `cargo test 2>&1 | grep "test result"` → 全 ok。
|
||||
- [ ] **Step 2:** `cargo build 2>&1 | tail -3` → Finished。
|
||||
- [ ] **Step 3:** 提交(若有遗留)。
|
||||
|
||||
---
|
||||
|
||||
## 后续期次(概要,第 1 期落地后逐期细化为同样的 TDD 任务)
|
||||
|
||||
### 第 2 期:积分轨
|
||||
- `user_credits` 已建表(Task 1)。新增 `adjust_user_credits`(admin 发放/增减,写 ledger points/admin_adjust)。
|
||||
- comfy_cloud 确认到账 `admin_confirm_user_recharge` 改为加 `points_balance` + ledger(points, recharge),受 config 开关 `payment.comfy_recharge_to_points` 控制(默认 true);历史不迁移。
|
||||
- 工作流积分单价 config(默认值 + 按工作流名分级映射)。
|
||||
- `workflow/executor.rs` `run_workflow`:执行**前**查 `points_balance >= 单价`,不足返回 Forbidden「工作流积分不足」;执行**成功后**扣积分 + ledger(points, consume, model=工作流名)。
|
||||
- 测试:拦截、扣费、充值进积分、管理员调整。
|
||||
|
||||
### 第 3 期:A 消费明细
|
||||
- `GET /api/user/billing/ledger`(查本人 credit_ledger,track/type/时间过滤、分页)。
|
||||
- `GET /api/user/billing/summary`(按天/模型聚合,双轨分开)。
|
||||
- 扩展 `UsageRecords.tsx`:双轨切换 + 汇总卡 + 明细列表。
|
||||
|
||||
### 第 4 期:B 充值档位
|
||||
- `recharge_tiers` 表(已在 spec,Task 1 未建,本期建)+ admin CRUD `/api/admin/recharge-tiers`。
|
||||
- 用户充值档位从 DB 读(替代 env `USER_RECHARGE_OPTIONS`)。
|
||||
- admin `RechargeTiers` 配置页 + `UserRecharge.tsx` 接 DB。
|
||||
|
||||
### 第 5 期:C 预警
|
||||
- 余额接口(user_billing_summary 等)加 `lowBalance`(余额 < 阈值,config 默认 0.2)。
|
||||
- Dashboard / UserRecharge / ComfyUIAgent 顶部预警条。
|
||||
|
||||
### 第 6 期:D 审计 + 对账
|
||||
- `GET /api/admin/billing/ledger`(跨用户,过滤分页)+ admin `BillingAudit` 页。
|
||||
- cron:拉 Comfy `/customers/balance` 总余额,与 points consume ledger 比对,偏差超阈值写 log + admin 告警标志(不自动改数据)。
|
||||
|
||||
---
|
||||
|
||||
## 自审
|
||||
|
||||
- **spec 覆盖**:第 1 期覆盖 spec「数据模型」+「USD 轨记账接入」三处(充值/消费/调整);后 5 期概要逐条对应 spec 的积分轨/A/B/C/D + 对账。recharge_tiers 表 spec 列在数据模型,实现挪到第 4 期建(第 1 期用不到),已注明。
|
||||
- **占位扫描**:第 1 期 Task 1-3 含完整代码;Task 4-5 用文字描述但步骤明确(adjust 与 Task 3 同模式,刻意不重复大段代码,执行时参照 Task 3)。后 5 期为概要,执行前细化。
|
||||
- **类型一致**:`LedgerEntry` 字段、`record_ledger_entry`/`ensure_ledger_schema` 签名在 Task 1 定义,Task 2-4 一致引用。`track`/`entry_type` 取值(usd/points、recharge/consume/admin_adjust)全程一致。
|
||||
Reference in New Issue
Block a user