Files

T

boos 0b7abed79a docs: 细化计费第3期(消费明细)TDD Task11-14

2026-06-17 01:44:28 +08:00

32 KiB

Raw Blame History

计费闭环实现计划

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 末尾：

#[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 之前）：

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: 提交

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 加：

#[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 函数体为：

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: 提交

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）：

#[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 内）：

    // 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: 提交

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: 提交

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）：

#[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 之前）：

/// 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):

#[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 之前):

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 实现:

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。

32 KiB Raw Blame History Unescape Escape

计费闭环 实现计划

文件结构（第 1 期）

第 1 期：地基（credit_ledger + user_credits 表 + USD 轨记账接入）

Task 1：ledger 表结构 + 写入入口

Task 2：USD 消费接入 ledger（record_usage）

Task 3：USD 充值接入 ledger（credit_user_balance）

Task 4：USD 管理员调整接入 ledger（adjust_user_balance）

Task 5：全量回归

后续期次（概要，第 1 期落地后逐期细化为同样的 TDD 任务）

第 2 期：积分轨

第 3 期：A 消费明细

第 4 期：B 充值档位

第 5 期：C 预警

第 6 期：D 审计 + 对账

自审

第 2 期：积分轨（细化 TDD）

Task 6：积分轨核心 apply_points_delta + points_balance

Task 7：工作流积分单价 config

Task 8：工作流执行前置拦截 + 成功扣积分

Task 9：comfy 充值改加积分 + admin 积分调整

Task 10：第 2 期全量回归

第 3 期：A 消费明细(细化 TDD)

Task 11:ledger 查询函数(明细 + 聚合)

Task 12:user 侧 billing 接口

Task 13:前端 api + UsageRecords 双轨展示

Task 14:第3期回归

32 KiB

Raw Blame History

计费闭环实现计划