forked from boos/BoosAPI
boos
695 lines
32 KiB
Markdown
695 lines
32 KiB
Markdown
# 计费闭环 实现计划
|
||
|
||
> **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）全程一致。
|
||
|
||
---
|
||
|
||
## 第 2 期：积分轨（细化 TDD）
|
||
|
||
文件结构：
|
||
- `src/services/billing/ledger.rs` 加积分轨核心：`points_balance(db,user_id)`（读余额）、`apply_points_delta(db, user_id, delta, entry_type, model, operator, ref_*)`（事务内：UPSERT user_credits 余额 + 写 ledger(points)，返回新余额）。所有积分增减走这一个入口。
|
||
- `src/config.rs`：ComfyUIConfig 加 `workflow_points_cost`（默认单价）。
|
||
- `src/workflow/executor.rs`：run_workflow_with_progress 前置拦截 + 收尾扣积分。
|
||
- `src/services/payment.rs`：admin_confirm_user_recharge 改为加积分（受 config 开关）。
|
||
- `src/routes/admin/users.rs`：新增 `adjust_user_credits` handler + 路由。
|
||
|
||
### Task 6：积分轨核心 apply_points_delta + points_balance
|
||
|
||
**Files:** Modify `src/services/billing/ledger.rs`
|
||
|
||
- [ ] **Step1 失败测试**（ledger.rs tests mod）：
|
||
|
||
```rust
|
||
#[tokio::test]
|
||
async fn apply_points_delta_upserts_balance_and_writes_ledger() {
|
||
let pool = mem_pool().await;
|
||
// 首次 +100（充值）
|
||
let bal = apply_points_delta(&pool, 9, 100.0, "recharge", None, Some("system"), Some("recharge_record"), None).await.unwrap();
|
||
assert_eq!(bal, 100.0);
|
||
// 再 -30（消费）
|
||
let bal2 = apply_points_delta(&pool, 9, -30.0, "consume", Some("wf/video"), Some("system"), Some("workflow"), None).await.unwrap();
|
||
assert_eq!(bal2, 70.0);
|
||
// user_credits 缓存 = 70
|
||
let cached: f64 = sqlx::query_scalar("SELECT points_balance FROM user_credits WHERE user_id = 9").fetch_one(&pool).await.unwrap();
|
||
assert_eq!(cached, 70.0);
|
||
// ledger 两行
|
||
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM credit_ledger WHERE user_id=9 AND track='points'").fetch_one(&pool).await.unwrap();
|
||
assert_eq!(n, 2);
|
||
}
|
||
```
|
||
|
||
- [ ] **Step2** `cargo test -p boosapi --lib apply_points_delta_upserts 2>&1 | tail -6` → FAIL。
|
||
|
||
- [ ] **Step3 实现**（ledger.rs，tests 之前）：
|
||
|
||
```rust
|
||
/// Current points balance for a user (0 if no row yet).
|
||
pub async fn points_balance(db: &SqlitePool, user_id: i64) -> AppResult<f64> {
|
||
let bal: Option<f64> = sqlx::query_scalar(
|
||
"SELECT points_balance FROM user_credits WHERE user_id = ?",
|
||
)
|
||
.bind(user_id)
|
||
.fetch_optional(db)
|
||
.await?;
|
||
Ok(bal.unwrap_or(0.0))
|
||
}
|
||
|
||
/// Apply a points delta (+credit / -debit) atomically: upsert the cached
|
||
/// balance in `user_credits` and append one `credit_ledger(track=points)` row,
|
||
/// in a single transaction. Returns the new balance.
|
||
#[allow(clippy::too_many_arguments)]
|
||
pub async fn apply_points_delta(
|
||
db: &SqlitePool,
|
||
user_id: i64,
|
||
delta: f64,
|
||
entry_type: &str,
|
||
model: Option<&str>,
|
||
operator: Option<&str>,
|
||
ref_type: Option<&str>,
|
||
ref_id: Option<&str>,
|
||
) -> AppResult<f64> {
|
||
ensure_ledger_schema(db).await?;
|
||
let mut tx = db.begin().await?;
|
||
let current: f64 = sqlx::query_scalar(
|
||
"SELECT points_balance FROM user_credits WHERE user_id = ?",
|
||
)
|
||
.bind(user_id)
|
||
.fetch_optional(&mut *tx)
|
||
.await?
|
||
.unwrap_or(0.0);
|
||
let next = current + delta;
|
||
let total_recharged_inc = if delta > 0.0 && entry_type == "recharge" { delta } else { 0.0 };
|
||
sqlx::query(
|
||
"INSERT INTO user_credits (user_id, points_balance, points_total_recharged, updated_at)
|
||
VALUES (?, ?, ?, datetime('now'))
|
||
ON CONFLICT(user_id) DO UPDATE SET
|
||
points_balance = ?,
|
||
points_total_recharged = points_total_recharged + ?,
|
||
updated_at = datetime('now')",
|
||
)
|
||
.bind(user_id).bind(next).bind(total_recharged_inc)
|
||
.bind(next).bind(total_recharged_inc)
|
||
.execute(&mut *tx).await?;
|
||
record_ledger_entry(&mut tx, LedgerEntry {
|
||
user_id, track: "points", entry_type,
|
||
amount: delta, balance_after: next,
|
||
ref_type, ref_id, model, operator, note: None,
|
||
}).await?;
|
||
tx.commit().await?;
|
||
Ok(next)
|
||
}
|
||
```
|
||
|
||
- [ ] **Step4** `cargo test -p boosapi --lib apply_points_delta 2>&1 | tail -6` → PASS；`cargo build -j 2 | tail -2`。
|
||
- [ ] **Step5** `git add src/services/billing/ledger.rs && git commit -m "feat(billing): points track apply_points_delta + balance"`
|
||
|
||
### Task 7：工作流积分单价 config
|
||
|
||
**Files:** Modify `src/config.rs`
|
||
|
||
- [ ] **Step1** 测试：config 默认值测试（若 config.rs 有 default 测试则加断言；否则跳测试，靠编译）。
|
||
- [ ] **Step2/3 实现**：ComfyUIConfig 加字段 `#[serde(default = "default_workflow_points_cost")] pub workflow_points_cost: f64,`，default fn 返回 `10.0`；Default impl 里加 `workflow_points_cost: 10.0,`。
|
||
- [ ] **Step4** `cargo build -j 2 2>&1 | tail -3` → Finished（config 字段加完整个 crate 能编译）。
|
||
- [ ] **Step5** `git add src/config.rs && git commit -m "feat(billing): workflow_points_cost config"`
|
||
|
||
### Task 8：工作流执行前置拦截 + 成功扣积分
|
||
|
||
**Files:** Modify `src/workflow/executor.rs`
|
||
|
||
- [ ] **Step1 失败测试**：executor.rs tests mod（若无则建）。建 in-memory pool + user_credits/credit_ledger，构造一个最简 graph，`ctx.user_id=Some(9)`：
|
||
- 余额不足（0 积分，单价 10）→ run_workflow_with_progress 返回 Err（Forbidden 含"积分不足"）。
|
||
- 余额充足（100）→ 执行成功后 points_balance 变 90，credit_ledger 出现一行 (points, consume, -10)。
|
||
（若 executor 难以单测真实节点，则把"拦截"和"扣费"逻辑抽成小函数 `check_workflow_points(state, user_id, cost)` 和 `charge_workflow_points(state, user_id, cost, workflow_name)` 各自单测，run_workflow 调它们。优先抽函数单测。）
|
||
|
||
- [ ] **Step2** 确认 FAIL。
|
||
|
||
- [ ] **Step3 实现**：
|
||
- 在 run_workflow_with_progress 的 `validate_and_sort` 之后、`insert_run` 之前：若 `ctx.user_id` 是 Some 且 `ctx.charge_points`（见下）：`let cost = state.config.comfyui.workflow_points_cost; if cost > 0.0 { let bal = ledger::points_balance(&state.db, uid).await?; if bal < cost { return Err(AppError::Forbidden("工作流积分不足，请充值".into())); } }`
|
||
- 在 `complete_run` 之后、`Ok(run_public_id)` 之前：`if let Some(uid) = ctx.user_id { let cost = state.config.comfyui.workflow_points_cost; if cost > 0.0 { let _ = ledger::apply_points_delta(&state.db, uid, -cost, "consume", Some(&format!("workflow:{}", workflow_id)), Some("system"), Some("workflow_run"), Some(&run_public_id)).await; } }`（扣费 best-effort，失败记 log 不阻断已产出结果）。
|
||
- 注意：Admin 调用（builtin key，user_id 可能是 Some(admin) 或特殊）是否扣？按设计仅对普通用户扣。可用 RunContext 加 `pub charge_points: bool` 字段，run_context 构造时 User=>true / Admin=>false。models.rs RunContext 加该字段，两处构造（routes.rs run_context、workflow_ui.rs workflow_context）补 `charge_points`。
|
||
|
||
- [ ] **Step4** 测试 PASS + `cargo build -j 2`。
|
||
- [ ] **Step5** `git add src/workflow/executor.rs src/workflow/models.rs src/workflow/routes.rs src/routes/workflow_ui.rs && git commit -m "feat(billing): workflow points gating + charge"`
|
||
|
||
### Task 9：comfy 充值改加积分 + admin 积分调整
|
||
|
||
**Files:** Modify `src/services/payment.rs`、`src/routes/admin/users.rs`、`src/config.rs`(开关)、`src/routes/admin/mod.rs`(路由)
|
||
|
||
- [ ] **Step1 失败测试**：
|
||
- payment.rs：`admin_confirm_user_recharge` 确认后，credit_ledger 出现 (points, recharge, +amount)（而非 usd）；user_credits 增加。
|
||
- users.rs：`adjust_user_credits` handler 调用后 credit_ledger 出现 (points, admin_adjust)。
|
||
- [ ] **Step2** FAIL。
|
||
- [ ] **Step3 实现**：
|
||
- config 开关 `payment.comfy_recharge_to_points: bool`（默认 true）。
|
||
- `admin_confirm_user_recharge`：开关为 true 时改调 `ledger::apply_points_delta(db, user_id, amount_usd_as_points, "recharge", None, Some("admin"), Some("recharge_record"), Some(public_id))`（USD→积分换算：本期 1 USD = N 积分，N 用 config `payment.usd_to_points_rate` 默认 100；即 grant_points = amount_usd * rate）。开关 false 时维持原 credit_user_balance(USD)。**历史不迁移**。
|
||
- 新增 `adjust_user_credits(State, Path(id), Json{delta})` handler（仿 adjust_user_balance）→ `apply_points_delta(db,id,delta,"admin_adjust",None,Some("admin"),Some("manual"),None)`，返回 {success, data{userId, pointsBalance}}。
|
||
- admin/mod.rs 注册 `POST /api/admin/users/{id}/credits/adjust`。
|
||
- [ ] **Step4** 相关测试 + payment/users 模块全过 + build。
|
||
- [ ] **Step5** 分两个 commit 或一个：`git commit -m "feat(billing): comfy recharge to points + admin credit adjust"`
|
||
|
||
### Task 10：第 2 期全量回归
|
||
- [ ] `cargo test 2>&1 | grep "test result"` 全 ok；`cargo build -j 2 | tail -2` Finished。
|
||
|
||
---
|
||
|
||
## 第 3 期：A 消费明细(细化 TDD)
|
||
|
||
文件:`src/services/billing/ledger.rs`(加查询函数)、`src/routes/user/mod.rs`(加 2 路由)、`src/web/api.ts`(加 client)、`src/web/pages/UsageRecords.tsx`(扩展)。
|
||
|
||
### Task 11:ledger 查询函数(明细 + 聚合)
|
||
|
||
**Files:** Modify `src/services/billing/ledger.rs`
|
||
|
||
- [ ] **Step1 失败测试**(ledger.rs tests mod):
|
||
```rust
|
||
#[tokio::test]
|
||
async fn list_and_summarize_user_ledger() {
|
||
let pool = mem_pool().await;
|
||
apply_points_delta(&pool, 9, 100.0, "recharge", None, Some("system"), Some("rr"), None).await.unwrap();
|
||
apply_points_delta(&pool, 9, -10.0, "consume", Some("wf:1"), Some("system"), Some("run"), None).await.unwrap();
|
||
// 明细:user 9 两行,track=points
|
||
let rows = list_user_ledger(&pool, 9, None, None, 50, 0).await.unwrap();
|
||
assert_eq!(rows.len(), 2);
|
||
// 聚合:points 轨 consume 合计 -10
|
||
let sum = summarize_user_ledger(&pool, 9).await.unwrap();
|
||
assert!(sum.iter().any(|v| v["track"]=="points" && v["entryType"]=="consume"));
|
||
}
|
||
```
|
||
|
||
- [ ] **Step2** `cargo test -p boosapi --lib list_and_summarize_user_ledger 2>&1|tail -6` FAIL。
|
||
|
||
- [ ] **Step3 实现**(ledger.rs,tests 之前):
|
||
```rust
|
||
use serde_json::{json, Value};
|
||
|
||
/// List a user's ledger rows, newest first, optional track/entry_type filter + paging.
|
||
pub async fn list_user_ledger(
|
||
db: &SqlitePool, user_id: i64,
|
||
track: Option<&str>, entry_type: Option<&str>,
|
||
limit: i64, offset: i64,
|
||
) -> AppResult<Vec<Value>> {
|
||
ensure_ledger_schema(db).await?;
|
||
let limit = limit.clamp(1, 500);
|
||
let rows = sqlx::query(
|
||
"SELECT track, entry_type, amount, balance_after, model, operator, ref_type, ref_id, note, created_at
|
||
FROM credit_ledger
|
||
WHERE user_id = ?
|
||
AND (?1_track IS NULL OR track = ?1_track)
|
||
AND (?1_type IS NULL OR entry_type = ?1_type)
|
||
ORDER BY id DESC LIMIT ? OFFSET ?"
|
||
.replace("?1_track", "?2").replace("?1_type", "?3")
|
||
)
|
||
.bind(user_id).bind(track).bind(entry_type).bind(limit).bind(offset)
|
||
.fetch_all(db).await?;
|
||
use sqlx::Row;
|
||
Ok(rows.iter().map(|r| json!({
|
||
"track": r.try_get::<String,_>("track").unwrap_or_default(),
|
||
"entryType": r.try_get::<String,_>("entry_type").unwrap_or_default(),
|
||
"amount": r.try_get::<f64,_>("amount").unwrap_or(0.0),
|
||
"balanceAfter": r.try_get::<f64,_>("balance_after").unwrap_or(0.0),
|
||
"model": r.try_get::<Option<String>,_>("model").ok().flatten(),
|
||
"operator": r.try_get::<Option<String>,_>("operator").ok().flatten(),
|
||
"refType": r.try_get::<Option<String>,_>("ref_type").ok().flatten(),
|
||
"note": r.try_get::<Option<String>,_>("note").ok().flatten(),
|
||
"createdAt": r.try_get::<Option<String>,_>("created_at").ok().flatten(),
|
||
})).collect())
|
||
}
|
||
|
||
/// Aggregate a user's ledger by (track, entry_type): sum amount + count.
|
||
pub async fn summarize_user_ledger(db: &SqlitePool, user_id: i64) -> AppResult<Vec<Value>> {
|
||
ensure_ledger_schema(db).await?;
|
||
let rows = sqlx::query(
|
||
"SELECT track, entry_type, COALESCE(SUM(amount),0) AS total, COUNT(*) AS cnt
|
||
FROM credit_ledger WHERE user_id = ?
|
||
GROUP BY track, entry_type ORDER BY track, entry_type"
|
||
).bind(user_id).fetch_all(db).await?;
|
||
use sqlx::Row;
|
||
Ok(rows.iter().map(|r| json!({
|
||
"track": r.try_get::<String,_>("track").unwrap_or_default(),
|
||
"entryType": r.try_get::<String,_>("entry_type").unwrap_or_default(),
|
||
"total": r.try_get::<f64,_>("total").unwrap_or(0.0),
|
||
"count": r.try_get::<i64,_>("cnt").unwrap_or(0),
|
||
})).collect())
|
||
}
|
||
```
|
||
注:上面 SQL 的 named-param replace 写法若 sqlx 不接受,改用直白的可选过滤:用 `track.is_none() OR track = ?` 形式,bind 顺序 user_id, track, track, entry_type, entry_type, limit, offset（每个 Option bind 两次)。子代理按 sqlx 实际行为调整到能编译+测试过。
|
||
|
||
- [ ] **Step4** 测试 PASS + `cargo build -j 2`。
|
||
- [ ] **Step5** `git add src/services/billing/ledger.rs && git commit -m "feat(billing): user ledger list + summary queries"`
|
||
|
||
### Task 12:user 侧 billing 接口
|
||
|
||
**Files:** Modify `src/routes/user/mod.rs`(加 2 路由 + 2 handler)
|
||
|
||
- [ ] **Step1 失败测试**:user mod tests(若有 test_state/建表 helper 则复用)——构造 user JWT + ledger 数据,GET /api/user/billing/ledger 返回行,/summary 返回聚合。若 user mod 无现成集成测试框架,改为直接单测 handler 逻辑或跳测试靠 ledger.rs 的 Task11 测试覆盖,仅验证编译 + 路由注册。
|
||
- [ ] **Step2** FAIL/编译失败。
|
||
- [ ] **Step3 实现**:
|
||
```rust
|
||
async fn get_billing_ledger(State(state): State<AppState>, headers: HeaderMap, Query(q): Query<BillingLedgerQuery>) -> AppResult<Json<Value>> {
|
||
let user_id = current_user_id(&headers, &state)?;
|
||
let rows = crate::services::billing::ledger::list_user_ledger(
|
||
&state.db, user_id, q.track.as_deref(), q.entry_type.as_deref(),
|
||
q.limit.unwrap_or(100), q.offset.unwrap_or(0)).await?;
|
||
Ok(Json(json!({ "success": true, "records": rows })))
|
||
}
|
||
async fn get_billing_summary(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
|
||
let user_id = current_user_id(&headers, &state)?;
|
||
let agg = crate::services::billing::ledger::summarize_user_ledger(&state.db, user_id).await?;
|
||
Ok(Json(json!({ "success": true, "summary": agg })))
|
||
}
|
||
```
|
||
+ `BillingLedgerQuery{ track:Option<String>, entry_type:Option<String>, limit:Option<i64>, offset:Option<i64> }`(serde Deserialize, rename_all camelCase 视前端传参)。
|
||
+ routes() 注册 `.route("/api/user/billing/ledger", get(get_billing_ledger)).route("/api/user/billing/summary", get(get_billing_summary))`。
|
||
- [ ] **Step4** `cargo test -p boosapi --lib 2>&1|command grep "test result"` + `cargo build -j 2`。
|
||
- [ ] **Step5** `git add src/routes/user/mod.rs && git commit -m "feat(billing): user billing ledger + summary endpoints"`
|
||
|
||
### Task 13:前端 api + UsageRecords 双轨展示
|
||
|
||
**Files:** Modify `src/web/api.ts`、`src/web/pages/UsageRecords.tsx`
|
||
|
||
- [ ] **Step1**:api.ts 的 userApi 加 `getBillingLedger(params)` → GET /api/user/billing/ledger、`getBillingSummary()` → /summary。
|
||
- [ ] **Step2**:UsageRecords.tsx 加一个"积分/USD"双轨切换 tab + 聚合汇总卡(用 getBillingSummary)+ ledger 明细列表(getBillingLedger)。保留现有 logs 视图作为 USD 明细来源之一,新增积分明细来自 ledger。具体 UI 复用现有 usage-records-* 样式。
|
||
- [ ] **Step3**:`npm run typecheck:web` 干净 + 现有 UsageRecords 测试不破坏(`npx vitest run src/web/pages/UsageRecords`)。
|
||
- [ ] **Step4** `git add src/web/api.ts src/web/pages/UsageRecords.tsx && git commit -m "feat(billing): UsageRecords dual-track ledger view"`
|
||
|
||
### Task 14:第3期回归
|
||
- [ ] `cargo test 2>&1|command grep "test result"` 全ok;`cargo build -j 2`;`npm run typecheck`。
|