feat(billing): USD admin adjust writes credit_ledger

Co-Authored-By: Claude <noreply@anthropic.com>

src/routes/admin/users.rs

@@ -4,6 +4,7 @@ use crate::{
     auth::password::hash_password,
     db::models::{downstream_api_key::DownstreamApiKey, user::User},
     error::{AppError, AppResult},
+    routes::auth::DEFAULT_NEW_USER_MAX_COST_USD,
     AppState,
 };
 use axum::{
@@ -238,17 +239,18 @@ pub async fn create_user(
 
     let user_id = result.last_insert_rowid();
 
-    // Auto-generate downstream API key
+    // Auto-generate downstream API key with the default starter credit.
     let key = format!("sk-user-{}", uuid::Uuid::new_v4());
     let key_name = format!("{} 的默认密钥", payload.username.trim());
 
     sqlx::query(
-        "INSERT INTO downstream_api_keys (name, key, user_id, enabled) VALUES (?, ?, ?, ?)",
+        "INSERT INTO downstream_api_keys (name, key, user_id, enabled, max_cost) VALUES (?, ?, ?, ?, ?)",
     )
     .bind(&key_name)
     .bind(&key)
     .bind(user_id)
     .bind(true)
+    .bind(DEFAULT_NEW_USER_MAX_COST_USD)
     .execute(&mut *tx)
     .await?;
 
@@ -708,6 +710,195 @@ pub async fn update_user_key(
     Ok(Json(json!({ "data": updated })))
 }
 
+// ── User balance management (admin) ────────────────────────────────────
+
+#[derive(Debug, Deserialize)]
+pub struct ListBalancesQuery {
+    pub page: Option<i64>,
+    pub page_size: Option<i64>,
+    pub q: Option<String>,
+}
+
+/// GET /api/admin/user-balances — list users with their credit pool
+/// (summed max_cost / used_cost across their downstream keys).
+pub async fn list_user_balances(
+    State(state): State<AppState>,
+    Query(query): Query<ListBalancesQuery>,
+) -> AppResult<Json<Value>> {
+    let page = query.page.unwrap_or(1).max(1);
+    let page_size = query.page_size.unwrap_or(50).clamp(1, 200);
+    let offset = (page - 1) * page_size;
+    let q = query.q.unwrap_or_default().trim().to_string();
+    let q_like = format!("%{}%", q);
+
+    let rows = sqlx::query(
+        r#"
+        SELECT
+            u.id AS user_id,
+            u.username AS username,
+            u.email AS email,
+            u.role AS role,
+            u.status AS status,
+            COALESCE(SUM(COALESCE(k.used_cost, 0.0)), 0.0) AS used_cost,
+            COALESCE(SUM(CASE WHEN k.max_cost IS NULL THEN 0.0 ELSE k.max_cost END), 0.0) AS limited_max_cost,
+            SUM(CASE WHEN k.id IS NULL THEN 0 WHEN k.max_cost IS NULL THEN 1 ELSE 0 END) AS unlimited_key_count,
+            COUNT(k.id) AS key_count
+        FROM users u
+        LEFT JOIN downstream_api_keys k ON k.user_id = u.id
+        WHERE (? = '' OR u.username LIKE ? OR u.email LIKE ? OR CAST(u.id AS TEXT) = ?)
+        GROUP BY u.id
+        ORDER BY u.id DESC
+        LIMIT ? OFFSET ?
+        "#,
+    )
+    .bind(&q)
+    .bind(&q_like)
+    .bind(&q_like)
+    .bind(&q)
+    .bind(page_size)
+    .bind(offset)
+    .fetch_all(&*state.db)
+    .await?;
+
+    let total: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM users
+         WHERE (? = '' OR username LIKE ? OR email LIKE ? OR CAST(id AS TEXT) = ?)",
+    )
+    .bind(&q)
+    .bind(&q_like)
+    .bind(&q_like)
+    .bind(&q)
+    .fetch_one(&*state.db)
+    .await?;
+
+    let items: Vec<Value> = rows
+        .iter()
+        .map(|row| {
+            use sqlx::Row;
+            let used: f64 = row.try_get("used_cost").unwrap_or(0.0);
+            let limited_max: f64 = row.try_get("limited_max_cost").unwrap_or(0.0);
+            let unlimited: i64 = row.try_get("unlimited_key_count").unwrap_or(0);
+            let remaining = if unlimited > 0 {
+                Value::Null
+            } else {
+                Value::from((limited_max - used).max(0.0))
+            };
+            json!({
+                "userId": row.try_get::<i64, _>("user_id").unwrap_or(0),
+                "username": row.try_get::<Option<String>, _>("username").ok().flatten(),
+                "email": row.try_get::<Option<String>, _>("email").ok().flatten(),
+                "role": row.try_get::<Option<String>, _>("role").ok().flatten(),
+                "status": row.try_get::<Option<String>, _>("status").ok().flatten(),
+                "keyCount": row.try_get::<i64, _>("key_count").unwrap_or(0),
+                "usedCost": used,
+                "maxCost": if unlimited > 0 { Value::Null } else { Value::from(limited_max) },
+                "remainingCost": remaining,
+                "unlimitedKeyCount": unlimited,
+            })
+        })
+        .collect();
+
+    Ok(Json(json!({
+        "data": items,
+        "meta": {
+            "page": page,
+            "pageSize": page_size,
+            "total": total,
+            "totalPages": ((total + page_size - 1) / page_size).max(1),
+        },
+    })))
+}
+
+#[derive(Debug, Deserialize)]
+pub struct AdjustBalanceRequest {
+    /// Amount to add (positive) or remove (negative) from the user's credit pool.
+    pub delta: f64,
+}
+
+/// POST /api/admin/users/{id}/balance/adjust — add or remove credit on the
+/// user's primary (lowest-id) downstream key by bumping its max_cost.
+pub async fn adjust_user_balance(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+    Json(payload): Json<AdjustBalanceRequest>,
+) -> AppResult<Json<Value>> {
+    let delta = payload.delta;
+    if !delta.is_finite() || delta == 0.0 {
+        return Err(AppError::BadRequest("delta must be a non-zero finite number".into()));
+    }
+
+    crate::services::billing::ledger::ensure_ledger_schema(&state.db).await?;
+
+    let key: Option<(i64, Option<f64>, Option<f64>)> = sqlx::query_as(
+        "SELECT id, max_cost, used_cost
+         FROM downstream_api_keys
+         WHERE user_id = ?
+         ORDER BY id ASC
+         LIMIT 1",
+    )
+    .bind(id)
+    .fetch_optional(&*state.db)
+    .await?;
+
+    let Some((key_id, max_cost, used_cost)) = key else {
+        return Err(AppError::BadRequest(
+            "该用户没有可调整额度的 API Key（请先为其创建 Key）".into(),
+        ));
+    };
+
+    let used = used_cost.unwrap_or(0.0).max(0.0);
+    let current = max_cost.unwrap_or(used).max(used);
+    // Never let the limit drop below what's already been spent.
+    let next_max = (current + delta).max(used);
+    let remaining = (next_max - used).max(0.0);
+
+    let mut tx = state.db.begin().await?;
+    sqlx::query(
+        "UPDATE downstream_api_keys
+         SET max_cost = ?, updated_at = datetime('now')
+         WHERE id = ?",
+    )
+    .bind(next_max)
+    .bind(key_id)
+    .execute(&mut *tx)
+    .await?;
+
+    crate::services::billing::ledger::record_ledger_entry(
+        &mut tx,
+        crate::services::billing::ledger::LedgerEntry {
+            user_id: id,
+            track: "usd",
+            entry_type: "admin_adjust",
+            amount: delta,
+            balance_after: remaining,
+            ref_type: Some("manual"),
+            ref_id: None,
+            model: None,
+            operator: Some("admin"),
+            note: None,
+        },
+    )
+    .await?;
+    tx.commit().await?;
+
+    info!(
+        user_id = id,
+        key_id, delta, next_max, "admin adjusted user balance"
+    );
+
+    Ok(Json(json!({
+        "success": true,
+        "data": {
+            "userId": id,
+            "keyId": key_id,
+            "maxCost": next_max,
+            "usedCost": used,
+            "remainingCost": remaining,
+        }
+    })))
+}
+
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -800,6 +991,8 @@ mod tests {
                     template_proxy_url: None,
                 },
                 comfyui: Default::default(),
+                pixelle_video: Default::default(),
+                payment: Default::default(),
             },
             pool,
             BuiltinKeys {
@@ -847,4 +1040,53 @@ mod tests {
             .expect("read role");
         assert_eq!(role, "admin");
     }
+
+    #[tokio::test]
+    async fn adjust_user_balance_writes_admin_adjust_ledger() {
+        let state = test_state().await;
+        crate::services::billing::ledger::ensure_ledger_schema(&state.db)
+            .await
+            .expect("ledger schema");
+
+        let user_id = sqlx::query(
+            "INSERT INTO users (username, email, password_hash, role, status)
+             VALUES (?, ?, ?, 'user', 'active')",
+        )
+        .bind("Spender")
+        .bind("spender@example.test")
+        .bind("hash")
+        .execute(&*state.db)
+        .await
+        .expect("insert user")
+        .last_insert_rowid();
+
+        sqlx::query(
+            "INSERT INTO downstream_api_keys (name, key, user_id, max_cost, used_cost)
+             VALUES (?, ?, ?, 10.0, 0.0)",
+        )
+        .bind("primary")
+        .bind("sk-spender")
+        .bind(user_id)
+        .execute(&*state.db)
+        .await
+        .expect("insert key");
+
+        adjust_user_balance(
+            State(state.clone()),
+            Path(user_id),
+            Json(AdjustBalanceRequest { delta: 5.0 }),
+        )
+        .await
+        .expect("adjust balance");
+
+        let row: (String, String, f64) = sqlx::query_as(
+            "SELECT track, entry_type, amount FROM credit_ledger
+             WHERE user_id = ? ORDER BY id DESC LIMIT 1",
+        )
+        .bind(user_id)
+        .fetch_one(&*state.db)
+        .await
+        .expect("ledger row");
+        assert_eq!(row, ("usd".into(), "admin_adjust".into(), 5.0));
+    }
 }