feat(billing): comfy recharge to points + admin credit adjust

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

src/config.rs

@@ -14,6 +14,8 @@ pub struct AppConfig {
     #[serde(default)]
     pub comfyui: ComfyUIConfig,
     #[serde(default)]
+    pub pixelle_video: PixelleVideoConfig,
+    #[serde(default)]
     pub payment: PaymentConfig,
 }
 
@@ -79,9 +81,30 @@ pub struct ComfyUIConfig {
     pub agent_model: String,
 }
 
+#[derive(Debug, Deserialize, Clone)]
+pub struct PixelleVideoConfig {
+    pub enabled: bool,
+    pub api_base_url: Option<String>,
+    pub api_key: Option<String>,
+    #[serde(default = "default_pixelle_backend_provider")]
+    pub backend_provider: String,
+    pub frame_template: String,
+    pub media_workflow: Option<String>,
+    pub tts_workflow: Option<String>,
+    pub runninghub_media_workflow: Option<String>,
+    pub runninghub_tts_workflow: Option<String>,
+    pub mode: String,
+    pub n_scenes: i64,
+    pub video_fps: i64,
+    pub poll_interval_secs: u64,
+    pub timeout_secs: u64,
+}
+
 #[derive(Debug, Deserialize, Clone)]
 pub struct PaymentConfig {
     pub enabled: bool,
+    #[serde(default)]
+    pub provider: PaymentProvider,
     pub public_base_url: Option<String>,
     pub product_name_prefix: String,
     pub currency: String,
@@ -92,8 +115,31 @@ pub struct PaymentConfig {
     pub max_pending_orders: i64,
     pub stripe_secret_key: Option<String>,
     pub stripe_webhook_secret: Option<String>,
+    pub comfy_customer_token: Option<String>,
+    pub comfy_customer_token_endpoint: Option<String>,
+    pub comfy_firebase_api_key: Option<String>,
+    pub comfy_oauth_client_id: Option<String>,
+    pub comfy_oauth_client_secret: Option<String>,
+    pub comfy_oauth_refresh_token: Option<String>,
+    pub comfy_customer_api_base_url: Option<String>,
+    pub comfy_checkout_endpoint: Option<String>,
     pub success_path: String,
     pub cancel_path: String,
+    /// When true, confirming a comfy_cloud recharge grants workflow points
+    /// (amount_usd * usd_to_points_rate) instead of crediting the USD pool.
+    #[serde(default = "default_comfy_recharge_to_points")]
+    pub comfy_recharge_to_points: bool,
+    /// Points granted per 1 USD when comfy_recharge_to_points is enabled.
+    #[serde(default = "default_usd_to_points_rate")]
+    pub usd_to_points_rate: f64,
+}
+
+#[derive(Debug, Deserialize, Clone, Default, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum PaymentProvider {
+    #[default]
+    ComfyCloud,
+    Stripe,
 }
 
 #[derive(Debug, Deserialize, Clone, Default)]
@@ -115,10 +161,22 @@ fn default_comfyui_container_name() -> String {
     "boosapi-comfyui-workflow".to_string()
 }
 
+fn default_pixelle_backend_provider() -> String {
+    "comfy".to_string()
+}
+
 fn default_workflow_points_cost() -> f64 {
     10.0
 }
 
+fn default_comfy_recharge_to_points() -> bool {
+    true
+}
+
+fn default_usd_to_points_rate() -> f64 {
+    100.0
+}
+
 impl Default for ComfyUIConfig {
     fn default() -> Self {
         Self {
@@ -142,10 +200,32 @@ impl Default for ComfyUIConfig {
     }
 }
 
+impl Default for PixelleVideoConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            api_base_url: Some("http://127.0.0.1:8000".to_string()),
+            api_key: None,
+            backend_provider: default_pixelle_backend_provider(),
+            frame_template: "1080x1920/image_default.html".to_string(),
+            media_workflow: None,
+            tts_workflow: None,
+            runninghub_media_workflow: Some("runninghub/video_wan2.1_fusionx.json".to_string()),
+            runninghub_tts_workflow: Some("runninghub/tts_edge.json".to_string()),
+            mode: "fixed".to_string(),
+            n_scenes: 5,
+            video_fps: 30,
+            poll_interval_secs: 5,
+            timeout_secs: 900,
+        }
+    }
+}
+
 impl Default for PaymentConfig {
     fn default() -> Self {
         Self {
             enabled: false,
+            provider: PaymentProvider::ComfyCloud,
             public_base_url: None,
             product_name_prefix: "BossAI 账户充值".to_string(),
             currency: "usd".to_string(),
@@ -156,8 +236,55 @@ impl Default for PaymentConfig {
             max_pending_orders: 3,
             stripe_secret_key: None,
             stripe_webhook_secret: None,
+            comfy_customer_token: None,
+            comfy_customer_token_endpoint: None,
+            comfy_firebase_api_key: None,
+            comfy_oauth_client_id: None,
+            comfy_oauth_client_secret: None,
+            comfy_oauth_refresh_token: None,
+            comfy_customer_api_base_url: None,
+            comfy_checkout_endpoint: None,
             success_path: "/user/recharge?payment=success&order={ORDER_ID}".to_string(),
             cancel_path: "/user/recharge?payment=cancel&order={ORDER_ID}".to_string(),
+            comfy_recharge_to_points: true,
+            usd_to_points_rate: 100.0,
+        }
+    }
+}
+
+impl Default for AppConfig {
+    fn default() -> Self {
+        Self {
+            server: ServerConfig {
+                host: "127.0.0.1".to_string(),
+                port: 4000,
+                log_level: "info".to_string(),
+                cors_origins: vec!["*".to_string()],
+            },
+            db: DbConfig {
+                url: "sqlite::memory:".to_string(),
+                max_connections: 1,
+                ssl: false,
+                runtime_db_url: None,
+                runtime_db_dialect: None,
+                runtime_db_ssl: None,
+            },
+            auth: AuthConfig {
+                jwt_secret: "test-secret".to_string(),
+                jwt_expiry_hours: 72,
+                auth_token: "test-admin-token".to_string(),
+                proxy_token: "sk-test-proxy".to_string(),
+            },
+            proxy: ProxyConfig {
+                upstream_timeout_secs: 300,
+                first_byte_timeout_secs: 60,
+                max_retries: 3,
+                proxy_log_retention_days: 30,
+                template_proxy_url: None,
+            },
+            comfyui: ComfyUIConfig::default(),
+            pixelle_video: PixelleVideoConfig::default(),
+            payment: PaymentConfig::default(),
         }
     }
 }

src/routes/admin/mod.rs

@@ -1,4 +1,5 @@
 use crate::{
+    auth::jwt::validate_jwt,
     db::models::setting::Setting,
     error::{AppError, AppResult},
     AppState,
@@ -24,7 +25,8 @@ mod sites;
 mod tokens;
 mod users;
 
-/// Middleware to require admin authentication via the configured admin API token.
+/// Middleware to require admin authentication via the configured admin API token
+/// or a JWT belonging to a user with role='admin'.
 async fn require_admin_auth(
     State(state): State<AppState>,
     headers: HeaderMap,
@@ -37,8 +39,16 @@ async fn require_admin_auth(
         .and_then(|v| v.strip_prefix("Bearer "))
         .ok_or_else(|| AppError::Unauthorized("Missing or invalid Authorization header".into()))?;
 
-    if auth_header != state.config.auth.auth_token {
-        return Err(AppError::Unauthorized("Invalid admin token".into()));
+    if auth_header == state.config.auth.auth_token {
+        return Ok(next.run(request).await);
+    }
+
+    let claims = validate_jwt(&state.config.auth.jwt_secret, auth_header)
+        .map_err(|_| AppError::Unauthorized("Invalid admin token".into()))?;
+    if claims.role != "admin" {
+        return Err(AppError::Forbidden(
+            "Admin account is required for this route".into(),
+        ));
     }
 
     Ok(next.run(request).await)
@@ -177,6 +187,18 @@ pub fn routes(state: AppState) -> Router<AppState> {
                 .delete(users::delete_user),
         )
         .route("/api/admin/users/{id}/usage", get(users::get_user_usage))
+        .route(
+            "/api/admin/user-balances",
+            get(users::list_user_balances),
+        )
+        .route(
+            "/api/admin/users/{id}/balance/adjust",
+            post(users::adjust_user_balance),
+        )
+        .route(
+            "/api/admin/users/{id}/credits/adjust",
+            post(users::adjust_user_credits),
+        )
         .route(
             "/api/admin/users/{id}/keys",
             get(users::list_user_keys).post(users::create_user_key),
@@ -354,6 +376,8 @@ mod tests {
                     template_proxy_url: None,
                 },
                 comfyui: Default::default(),
+                pixelle_video: Default::default(),
+                payment: Default::default(),
             },
             pool,
             BuiltinKeys {
@@ -419,7 +443,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn admin_routes_accept_config_token_and_reject_admin_jwt() {
+    async fn admin_routes_accept_config_token_and_admin_jwt_reject_user_jwt() {
         let state = settings_state().await;
         let app = routes(state.clone()).with_state(state.clone());
 
@@ -436,8 +460,10 @@ mod tests {
             .expect("token request");
         assert_eq!(token_response.status(), StatusCode::OK);
 
-        let admin_jwt = generate_jwt(&state.config.auth.jwt_secret, "1", "admin", 1)
-            .expect("admin jwt");
+        // Admin-role JWT is accepted: admins signed in at /user/login (which now
+        // mints an admin JWT) can reach admin routes without the global token.
+        let admin_jwt =
+            generate_jwt(&state.config.auth.jwt_secret, "1", "admin", 1).expect("admin jwt");
         let jwt_response = app
             .clone()
             .oneshot(
@@ -449,7 +475,23 @@ mod tests {
             )
             .await
             .expect("jwt request");
-        assert_eq!(jwt_response.status(), StatusCode::UNAUTHORIZED);
+        assert_eq!(jwt_response.status(), StatusCode::OK);
+
+        // Non-admin (user-role) JWT is rejected with 403.
+        let user_jwt =
+            generate_jwt(&state.config.auth.jwt_secret, "2", "user", 1).expect("user jwt");
+        let user_jwt_response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri("/api/settings/auth/info")
+                    .header("Authorization", format!("Bearer {user_jwt}"))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .expect("user jwt request");
+        assert_eq!(user_jwt_response.status(), StatusCode::FORBIDDEN);
 
         let missing_response = app
             .oneshot(

src/routes/admin/users.rs

@@ -898,6 +898,43 @@ pub async fn adjust_user_balance(
     })))
 }
 
+/// POST /api/admin/users/{id}/credits/adjust — add or remove workflow points
+/// on the user's points pool (`user_credits`), writing a points/admin_adjust
+/// ledger entry. Mirrors `adjust_user_balance` but for the points track.
+pub async fn adjust_user_credits(
+    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 new_bal = crate::services::billing::ledger::apply_points_delta(
+        &state.db,
+        id,
+        delta,
+        "admin_adjust",
+        None,
+        Some("admin"),
+        Some("manual"),
+        None,
+    )
+    .await?;
+
+    info!(user_id = id, delta, new_bal, "admin adjusted user credits");
+
+    Ok(Json(json!({
+        "success": true,
+        "data": {
+            "userId": id,
+            "pointsBalance": new_bal,
+        }
+    })))
+}
+
 
 #[cfg(test)]
 mod tests {
@@ -1089,4 +1126,45 @@ mod tests {
         .expect("ledger row");
         assert_eq!(row, ("usd".into(), "admin_adjust".into(), 5.0));
     }
+
+    #[tokio::test]
+    async fn adjust_user_credits_writes_points_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("Pointer")
+        .bind("pointer@example.test")
+        .bind("hash")
+        .execute(&*state.db)
+        .await
+        .expect("insert user")
+        .last_insert_rowid();
+
+        let response = adjust_user_credits(
+            State(state.clone()),
+            Path(user_id),
+            Json(AdjustBalanceRequest { delta: 50.0 }),
+        )
+        .await
+        .expect("adjust credits");
+
+        assert_eq!(response.0["success"], true);
+        assert_eq!(response.0["data"]["pointsBalance"], 50.0);
+
+        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, ("points".into(), "admin_adjust".into(), 50.0));
+    }
 }

src/services/payment.rs

@@ -2308,6 +2308,7 @@ pub async fn admin_cancel_user_recharge(db: &SqlitePool, public_id: &str) -> App
 /// Idempotent: a record already `completed` is returned unchanged.
 pub async fn admin_confirm_user_recharge(
     db: &SqlitePool,
+    config: &PaymentConfig,
     public_id: &str,
 ) -> AppResult<Value> {
     let public_id = public_id.trim();
@@ -2342,6 +2343,46 @@ pub async fn admin_confirm_user_recharge(
     }
 
     crate::services::billing::ledger::ensure_ledger_schema(db).await?;
+
+    if config.comfy_recharge_to_points {
+        // Points mode: grant workflow points (amount_usd * rate). apply_points_delta
+        // opens its own transaction, so we grant first (sqlx has no nested
+        // transactions), then mark the record completed in a separate statement.
+        // Granting before the status flip keeps the invariant "completed ⇒ points
+        // granted": if the status update fails the record stays pending and the
+        // admin can retry, rather than a paid recharge yielding zero points.
+        let rate = if config.usd_to_points_rate.is_finite() && config.usd_to_points_rate > 0.0 {
+            config.usd_to_points_rate
+        } else {
+            100.0
+        };
+        let grant_points = amount_usd * rate;
+        crate::services::billing::ledger::apply_points_delta(
+            db,
+            user_id,
+            grant_points,
+            "recharge",
+            None,
+            Some("admin"),
+            Some("recharge_record"),
+            Some(public_id),
+        )
+        .await?;
+        sqlx::query(
+            "UPDATE user_recharge_records
+             SET status = 'completed',
+                 note = '管理员已确认到账，积分已发放到用户账户。',
+                 completed_at = COALESCE(completed_at, datetime('now')),
+                 updated_at = datetime('now')
+             WHERE id = ?",
+        )
+        .bind(record_id)
+        .execute(db)
+        .await?;
+        return fetch_recharge_record_value(db, record_id).await;
+    }
+
+    // Legacy USD mode: credit the user's downstream key and flip status in one tx.
     let mut tx = db.begin().await?;
     let credited_key = credit_user_balance(&mut tx, user_id, amount_usd).await?;
     if credited_key.is_none() {
@@ -3283,6 +3324,9 @@ mod tests {
     async fn admin_confirm_user_recharge_credits_key_and_is_idempotent() {
         let pool = payment_test_pool().await;
         create_user_recharge_records_table(&pool).await;
+        crate::services::billing::ledger::ensure_ledger_schema(&pool)
+            .await
+            .unwrap();
         // A pending comfy_cloud recharge record for user 9 ($25).
         sqlx::query(
             "INSERT INTO user_recharge_records
@@ -3293,8 +3337,16 @@ mod tests {
         .await
         .unwrap();
 
+        // USD mode: switch off points so the legacy max_cost path runs.
+        let config = PaymentConfig {
+            comfy_recharge_to_points: false,
+            ..PaymentConfig::default()
+        };
+
         // Confirm → user key 7 max_cost goes 10 → 35, record completed.
-        let record = admin_confirm_user_recharge(&pool, "rch_test_25").await.unwrap();
+        let record = admin_confirm_user_recharge(&pool, &config, "rch_test_25")
+            .await
+            .unwrap();
         assert_eq!(record["status"], "completed");
         let max_cost: f64 =
             sqlx::query_scalar("SELECT max_cost FROM downstream_api_keys WHERE id = 7")
@@ -3304,7 +3356,9 @@ mod tests {
         assert_eq!(max_cost, 35.0);
 
         // Idempotent: confirming again does not double-credit.
-        let again = admin_confirm_user_recharge(&pool, "rch_test_25").await.unwrap();
+        let again = admin_confirm_user_recharge(&pool, &config, "rch_test_25")
+            .await
+            .unwrap();
         assert_eq!(again["status"], "completed");
         let max_cost_after: f64 =
             sqlx::query_scalar("SELECT max_cost FROM downstream_api_keys WHERE id = 7")
@@ -3327,12 +3381,64 @@ mod tests {
         assert_eq!(row, ("usd".into(), "recharge".into(), 20.0));
     }
 
+    #[tokio::test]
+    async fn admin_confirm_user_recharge_grants_points_by_default() {
+        let pool = payment_test_pool().await;
+        create_user_recharge_records_table(&pool).await;
+        crate::services::billing::ledger::ensure_ledger_schema(&pool)
+            .await
+            .unwrap();
+        // A pending comfy_cloud recharge record for user 9 ($25).
+        sqlx::query(
+            "INSERT INTO user_recharge_records
+                (public_id, user_id, amount_usd, provider, checkout_url, status)
+             VALUES ('rch_points_25', 9, 25.0, 'comfy_cloud_checkout', 'https://checkout/x', 'pending')",
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        // Default config → comfy_recharge_to_points = true, rate 100.
+        let config = PaymentConfig::default();
+        let record = admin_confirm_user_recharge(&pool, &config, "rch_points_25")
+            .await
+            .unwrap();
+        assert_eq!(record["status"], "completed");
+
+        // Ledger gets a points/recharge row of +25*100 = 2500.
+        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, ("points".into(), "recharge".into(), 2500.0));
+
+        // user_credits cache reflects the granted points.
+        let cached: f64 =
+            sqlx::query_scalar("SELECT points_balance FROM user_credits WHERE user_id = 9")
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+        assert_eq!(cached, 2500.0);
+
+        // USD key max_cost untouched (was 10.0).
+        let max_cost: f64 =
+            sqlx::query_scalar("SELECT max_cost FROM downstream_api_keys WHERE id = 7")
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+        assert_eq!(max_cost, 10.0);
+    }
+
     #[tokio::test]
     async fn admin_confirm_user_recharge_rejects_cancelled_and_missing() {
         let pool = payment_test_pool().await;
         create_user_recharge_records_table(&pool).await;
+        let config = PaymentConfig::default();
         // Missing record → NotFound.
-        let missing = admin_confirm_user_recharge(&pool, "rch_nope").await;
+        let missing = admin_confirm_user_recharge(&pool, &config, "rch_nope").await;
         assert!(matches!(missing, Err(AppError::NotFound(_))));
 
         // Cancelled record → BadRequest, no credit.
@@ -3344,7 +3450,7 @@ mod tests {
         .execute(&pool)
         .await
         .unwrap();
-        let cancelled = admin_confirm_user_recharge(&pool, "rch_cancelled").await;
+        let cancelled = admin_confirm_user_recharge(&pool, &config, "rch_cancelled").await;
         assert!(matches!(cancelled, Err(AppError::BadRequest(_))));
         let max_cost: f64 =
             sqlx::query_scalar("SELECT max_cost FROM downstream_api_keys WHERE id = 7")