feat(billing): workflow points gating + charge

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

src/workflow/executor.rs

@@ -13,6 +13,30 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
 
 const DEFAULT_TEXT_TO_VIDEO_MODEL: &str = "seedance-1-0-lite-t2v-250428";
 
+/// Gate a workflow run on the caller's points balance. No-op when the caller
+/// is not being charged (admin), has no user id, or the cost is zero.
+async fn ensure_workflow_points(
+    db: &sqlx::SqlitePool,
+    user_id: Option<i64>,
+    charge_points: bool,
+    cost: f64,
+) -> AppResult<()> {
+    if !charge_points {
+        return Ok(());
+    }
+    let Some(uid) = user_id else {
+        return Ok(());
+    };
+    if cost <= 0.0 {
+        return Ok(());
+    }
+    let bal = crate::services::billing::ledger::points_balance(db, uid).await?;
+    if bal < cost {
+        return Err(AppError::Forbidden("工作流积分不足，请充值".into()));
+    }
+    Ok(())
+}
+
 fn normalize_video_model(value: &str) -> String {
     match value.trim() {
         "seedance-1-0-lite-t2v" => "seedance-1-0-lite-t2v-250428".to_string(),
@@ -81,6 +105,8 @@ pub async fn run_workflow_with_progress(
         .map_err(|err| AppError::BadRequest(format!("invalid workflow graph: {err}")))?;
     let order = graph::validate_and_sort(&graph)?;
     let total = order.len().max(1);
+    let cost = state.config.comfyui.workflow_points_cost;
+    ensure_workflow_points(&state.db, ctx.user_id, ctx.charge_points, cost).await?;
     let (run_db_id, run_public_id) =
         storage::insert_run(state, workflow_id, ctx.user_id, &input_overrides).await?;
     emit_progress(
@@ -161,6 +187,30 @@ pub async fn run_workflow_with_progress(
 
     let final_outputs = json!({ "nodes": store });
     storage::complete_run(state, run_db_id, &final_outputs).await?;
+    if ctx.charge_points && cost > 0.0 {
+        if let Some(uid) = ctx.user_id {
+            if let Err(err) = crate::services::billing::ledger::apply_points_delta(
+                &state.db,
+                uid,
+                -cost,
+                "consume",
+                Some(&format!("workflow:{}", workflow_id)),
+                Some("system"),
+                Some("workflow_run"),
+                Some(&run_public_id),
+            )
+            .await
+            {
+                tracing::warn!(
+                    user_id = uid,
+                    workflow_id,
+                    run = %run_public_id,
+                    error = %err,
+                    "failed to charge workflow points after successful run"
+                );
+            }
+        }
+    }
     emit_progress(
         &progress,
         WorkflowProgressEvent::Finished {
@@ -658,7 +708,11 @@ fn video_request_body(node: &WorkflowNode, prompt: String, image_url: Option<Str
     let mut body = Map::new();
     body.insert(
         "model".to_string(),
-        json!(normalize_video_model(&param_str(node, "model", DEFAULT_TEXT_TO_VIDEO_MODEL))),
+        json!(normalize_video_model(&param_str(
+            node,
+            "model",
+            DEFAULT_TEXT_TO_VIDEO_MODEL
+        ))),
     );
     body.insert("prompt".to_string(), json!(prompt));
     for key in ["ratio", "resolution"] {
@@ -1088,3 +1142,58 @@ fn value_to_string(value: Option<&Value>) -> String {
         Some(other) => other.to_string(),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::services::billing::ledger;
+    use sqlx::sqlite::SqlitePoolOptions;
+
+    async fn mem_pool() -> sqlx::SqlitePool {
+        let pool = SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect("sqlite::memory:")
+            .await
+            .unwrap();
+        ledger::ensure_ledger_schema(&pool).await.unwrap();
+        pool
+    }
+
+    #[tokio::test]
+    async fn ensure_workflow_points_skips_when_not_charging() {
+        let pool = mem_pool().await;
+        // charge_points=false (admin) + zero balance + cost 10 => Ok (no gating).
+        ensure_workflow_points(&pool, Some(9), false, 10.0)
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn ensure_workflow_points_rejects_insufficient_balance() {
+        let pool = mem_pool().await;
+        // user 9 has zero points, cost 10 => Forbidden.
+        let err = ensure_workflow_points(&pool, Some(9), true, 10.0)
+            .await
+            .unwrap_err();
+        match err {
+            AppError::Forbidden(msg) => assert!(msg.contains("积分不足"), "msg was: {msg}"),
+            other => panic!("expected Forbidden, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn ensure_workflow_points_allows_sufficient_balance_and_zero_cost() {
+        let pool = mem_pool().await;
+        ledger::apply_points_delta(&pool, 9, 100.0, "recharge", None, Some("system"), None, None)
+            .await
+            .unwrap();
+        // Balance 100 >= cost 10 => Ok.
+        ensure_workflow_points(&pool, Some(9), true, 10.0)
+            .await
+            .unwrap();
+        // cost 0 => noop Ok regardless of balance.
+        ensure_workflow_points(&pool, Some(9), true, 0.0)
+            .await
+            .unwrap();
+    }
+}

src/workflow/models.rs

@@ -97,6 +97,7 @@ pub struct RunContext {
     pub user_id: Option<i64>,
     pub bearer: String,
     pub base_url: String,
+    pub charge_points: bool,
 }
 
 #[derive(Debug, Clone)]

src/workflow/routes.rs

@@ -163,6 +163,7 @@ async fn run_context(state: &AppState, caller: &WorkflowCaller) -> AppResult<Run
         user_id: caller_user_id(caller),
         bearer,
         base_url: format!("http://{}:{}/v1", host, state.config.server.port),
+        charge_points: !caller_is_admin(caller),
     })
 }