Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a89edf6f75 | |||
| 5d42671172 | |||
| ca9e45c59c |
+127
@@ -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(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+49
-7
@@ -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(
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
+1094
-14
File diff suppressed because it is too large
Load Diff
+87
-46
@@ -13,10 +13,9 @@ use crate::{
|
||||
use axum::{
|
||||
body::Bytes,
|
||||
extract::{
|
||||
DefaultBodyLimit,
|
||||
multipart::Multipart,
|
||||
ws::{Message, WebSocketUpgrade},
|
||||
FromRequest, Path, Query, State,
|
||||
DefaultBodyLimit, FromRequest, Path, Query, State,
|
||||
},
|
||||
http::{header, HeaderMap, HeaderValue, Method, StatusCode, Uri},
|
||||
response::{Html, IntoResponse, Response},
|
||||
@@ -1311,7 +1310,10 @@ fn is_boosapi_extension_node_type(node_type: &str) -> bool {
|
||||
)
|
||||
}
|
||||
|
||||
fn workflow_node_defs_with_models(models: &[String], include_compat_core: bool) -> Map<String, Value> {
|
||||
fn workflow_node_defs_with_models(
|
||||
models: &[String],
|
||||
include_compat_core: bool,
|
||||
) -> Map<String, Value> {
|
||||
let core_defs = include_compat_core
|
||||
.then(core_node_definitions)
|
||||
.unwrap_or_default();
|
||||
@@ -1865,9 +1867,7 @@ fn cache_control_for_path(path: &StdPath) -> &'static str {
|
||||
.to_ascii_lowercase();
|
||||
match ext.as_str() {
|
||||
"html" | "js" | "mjs" | "css" => "no-cache",
|
||||
"woff2" | "png" | "jpg" | "jpeg" | "webp" | "svg" => {
|
||||
"public, max-age=604800"
|
||||
}
|
||||
"woff2" | "png" | "jpg" | "jpeg" | "webp" | "svg" => "public, max-age=604800",
|
||||
_ => "public, max-age=3600",
|
||||
}
|
||||
}
|
||||
@@ -1921,7 +1921,8 @@ async fn get_setting(
|
||||
Path(id): Path<String>,
|
||||
) -> AppResult<Json<Value>> {
|
||||
let _ = auth_caller(&state, &headers).await?;
|
||||
if state.config.comfyui.uses_cloud_backend() && id == "Comfy.Workflow.ShowMissingModelsWarning" {
|
||||
if state.config.comfyui.uses_cloud_backend() && id == "Comfy.Workflow.ShowMissingModelsWarning"
|
||||
{
|
||||
return Ok(Json(Value::Bool(false)));
|
||||
}
|
||||
Ok(Json(Value::Null))
|
||||
@@ -2936,7 +2937,10 @@ fn repair_resize_image_mask_node_inputs_without_workflow(
|
||||
inputs
|
||||
.entry("resize_type".to_string())
|
||||
.or_insert_with(|| Value::String("scale dimensions".to_string()));
|
||||
inputs.insert("resize_type.crop".to_string(), Value::String("center".to_string()));
|
||||
inputs.insert(
|
||||
"resize_type.crop".to_string(),
|
||||
Value::String("center".to_string()),
|
||||
);
|
||||
tracing::warn!(
|
||||
node_id = %node_id,
|
||||
"repaired ResizeImageMaskNode crop prompt input without workflow metadata"
|
||||
@@ -3186,14 +3190,11 @@ fn dynamic_combo_value_matches_field(value: &Value, field: &DynamicComboField) -
|
||||
.as_str()
|
||||
.is_some_and(|text| field.options.iter().any(|option| option == text)),
|
||||
"INT" => dynamic_combo_value_as_i64(value).is_some_and(|number| {
|
||||
field
|
||||
.min
|
||||
.is_none_or(|min| (number as f64) >= min)
|
||||
field.min.is_none_or(|min| (number as f64) >= min)
|
||||
&& field.max.is_none_or(|max| (number as f64) <= max)
|
||||
}),
|
||||
"FLOAT" => dynamic_combo_value_as_f64(value).is_some_and(|number| {
|
||||
field.min.is_none_or(|min| number >= min)
|
||||
&& field.max.is_none_or(|max| number <= max)
|
||||
field.min.is_none_or(|min| number >= min) && field.max.is_none_or(|max| number <= max)
|
||||
}),
|
||||
"BOOLEAN" => dynamic_combo_value_as_bool(value).is_some(),
|
||||
_ => true,
|
||||
@@ -3239,28 +3240,42 @@ fn dynamic_combo_value_as_i64(value: &Value) -> Option<i64> {
|
||||
.as_i64()
|
||||
.or_else(|| value.as_u64().and_then(|number| i64::try_from(number).ok()))
|
||||
.or_else(|| value.as_f64().map(|number| number.round() as i64))
|
||||
.or_else(|| value.as_str().and_then(|text| text.trim().parse::<i64>().ok()))
|
||||
.or_else(|| {
|
||||
value
|
||||
.as_str()
|
||||
.and_then(|text| text.trim().parse::<i64>().ok())
|
||||
})
|
||||
}
|
||||
|
||||
fn dynamic_combo_value_as_f64(value: &Value) -> Option<f64> {
|
||||
value
|
||||
.as_f64()
|
||||
.or_else(|| value.as_str().and_then(|text| text.trim().parse::<f64>().ok()))
|
||||
value.as_f64().or_else(|| {
|
||||
value
|
||||
.as_str()
|
||||
.and_then(|text| text.trim().parse::<f64>().ok())
|
||||
})
|
||||
}
|
||||
|
||||
fn dynamic_combo_value_as_bool(value: &Value) -> Option<bool> {
|
||||
value.as_bool().or_else(|| {
|
||||
value.as_str().and_then(|text| match text.trim().to_ascii_lowercase().as_str() {
|
||||
"true" | "1" | "yes" | "on" => Some(true),
|
||||
"false" | "0" | "no" | "off" => Some(false),
|
||||
_ => None,
|
||||
})
|
||||
value
|
||||
.as_str()
|
||||
.and_then(|text| match text.trim().to_ascii_lowercase().as_str() {
|
||||
"true" | "1" | "yes" | "on" => Some(true),
|
||||
"false" | "0" | "no" | "off" => Some(false),
|
||||
_ => None,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
fn clamp_i64_to_field(value: i64, field: &DynamicComboField) -> i64 {
|
||||
let min = field.min.map(|value| value.ceil() as i64).unwrap_or(i64::MIN);
|
||||
let max = field.max.map(|value| value.floor() as i64).unwrap_or(i64::MAX);
|
||||
let min = field
|
||||
.min
|
||||
.map(|value| value.ceil() as i64)
|
||||
.unwrap_or(i64::MIN);
|
||||
let max = field
|
||||
.max
|
||||
.map(|value| value.floor() as i64)
|
||||
.unwrap_or(i64::MAX);
|
||||
value.clamp(min, max)
|
||||
}
|
||||
|
||||
@@ -3284,9 +3299,7 @@ fn repair_wan2_api_nodes(prompt: &mut Value) {
|
||||
.unwrap_or_default()
|
||||
.to_string();
|
||||
let field_names = match class_type.as_str() {
|
||||
"Wan2ImageToVideoApi" => {
|
||||
&["prompt", "negative_prompt", "resolution", "duration"][..]
|
||||
}
|
||||
"Wan2ImageToVideoApi" => &["prompt", "negative_prompt", "resolution", "duration"][..],
|
||||
"Wan2TextToVideoApi" | "Wan2ReferenceVideoApi" => &[
|
||||
"prompt",
|
||||
"negative_prompt",
|
||||
@@ -3789,7 +3802,11 @@ async fn upload_image_to_native_backend(
|
||||
let content_type = headers
|
||||
.get(header::CONTENT_TYPE)
|
||||
.and_then(|value| value.to_str().ok())
|
||||
.filter(|value| value.to_ascii_lowercase().starts_with("multipart/form-data"))
|
||||
.filter(|value| {
|
||||
value
|
||||
.to_ascii_lowercase()
|
||||
.starts_with("multipart/form-data")
|
||||
})
|
||||
.ok_or_else(|| AppError::BadRequest("multipart/form-data content-type is required".into()))?
|
||||
.to_string();
|
||||
let response = native_workflow_request(state, Method::POST, "/upload/image", None)
|
||||
@@ -4179,10 +4196,16 @@ async fn native_workflow_raw_response(
|
||||
) -> AppResult<Response> {
|
||||
let is_prompt_post = method == Method::POST && path.split('?').next() == Some("/prompt");
|
||||
let mut request = native_workflow_request(state, method, path, None);
|
||||
if let Some(content_type) = headers.get(header::CONTENT_TYPE).and_then(|value| value.to_str().ok()) {
|
||||
if let Some(content_type) = headers
|
||||
.get(header::CONTENT_TYPE)
|
||||
.and_then(|value| value.to_str().ok())
|
||||
{
|
||||
request = request.header(reqwest::header::CONTENT_TYPE, content_type);
|
||||
}
|
||||
if let Some(accept) = headers.get(header::ACCEPT).and_then(|value| value.to_str().ok()) {
|
||||
if let Some(accept) = headers
|
||||
.get(header::ACCEPT)
|
||||
.and_then(|value| value.to_str().ok())
|
||||
{
|
||||
request = request.header(reqwest::header::ACCEPT, accept);
|
||||
}
|
||||
if !body.is_empty() {
|
||||
@@ -4861,7 +4884,10 @@ async fn experiment_models_in_folder(
|
||||
native_items.extend(items);
|
||||
}
|
||||
}
|
||||
return Ok(Json(merge_model_payload_items(api_model_items, native_items)));
|
||||
return Ok(Json(merge_model_payload_items(
|
||||
api_model_items,
|
||||
native_items,
|
||||
)));
|
||||
}
|
||||
let path = format!("/experiment/models/{}", simple_urlencode(&folder));
|
||||
return native_workflow_json_cached(
|
||||
@@ -4909,7 +4935,10 @@ async fn experiment_models_in_folder(
|
||||
}
|
||||
}
|
||||
}
|
||||
return Ok(Json(merge_model_payload_items(native_items, api_model_items)));
|
||||
return Ok(Json(merge_model_payload_items(
|
||||
native_items,
|
||||
api_model_items,
|
||||
)));
|
||||
}
|
||||
let caller =
|
||||
auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
|
||||
@@ -4926,7 +4955,10 @@ async fn experiment_models_in_folder(
|
||||
}));
|
||||
}
|
||||
}
|
||||
Ok(Json(merge_model_payload_items(api_model_items, local_items)))
|
||||
Ok(Json(merge_model_payload_items(
|
||||
api_model_items,
|
||||
local_items,
|
||||
)))
|
||||
}
|
||||
|
||||
async fn view_metadata(
|
||||
@@ -5228,9 +5260,14 @@ async fn workflow_context(state: &AppState, caller: &WorkflowCaller) -> AppResul
|
||||
.bind(user_id)
|
||||
.fetch_optional(&*state.db)
|
||||
.await?;
|
||||
maybe_key
|
||||
.map(|key| key.key)
|
||||
.unwrap_or_else(|| state.builtin_keys.comfyui_agent.1.clone())
|
||||
// Must run on the user's own key so consumption is billed to them.
|
||||
// Falling back to the global builtin key would mix every user's
|
||||
// cost into one shared account, breaking per-user accounting.
|
||||
maybe_key.map(|key| key.key).ok_or_else(|| {
|
||||
AppError::Forbidden(
|
||||
"当前用户没有可用的 API Key,无法计费执行工作流,请先创建 API Key".into(),
|
||||
)
|
||||
})?
|
||||
}
|
||||
};
|
||||
let host = if state.config.server.host == "0.0.0.0" || state.config.server.host == "::" {
|
||||
@@ -5242,6 +5279,7 @@ async fn workflow_context(state: &AppState, caller: &WorkflowCaller) -> AppResul
|
||||
user_id: caller_user_id(caller),
|
||||
bearer,
|
||||
base_url: format!("http://{}:{}/v1", host, state.config.server.port),
|
||||
charge_points: !caller_is_admin(caller),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5812,20 +5850,23 @@ mod tests {
|
||||
.filter_map(|item| item.get("name").and_then(Value::as_str))
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
assert_eq!(
|
||||
names,
|
||||
vec![
|
||||
"wan2.1-t2v",
|
||||
"local-video-model.safetensors",
|
||||
]
|
||||
);
|
||||
assert_eq!(names, vec!["wan2.1-t2v", "local-video-model.safetensors",]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn workflow_ui_runtime_assets_revalidate_after_install() {
|
||||
assert_eq!(cache_control_for_path(StdPath::new("index.html")), "no-cache");
|
||||
assert_eq!(cache_control_for_path(StdPath::new("assets/index.js")), "no-cache");
|
||||
assert_eq!(cache_control_for_path(StdPath::new("assets/index.css")), "no-cache");
|
||||
assert_eq!(
|
||||
cache_control_for_path(StdPath::new("index.html")),
|
||||
"no-cache"
|
||||
);
|
||||
assert_eq!(
|
||||
cache_control_for_path(StdPath::new("assets/index.js")),
|
||||
"no-cache"
|
||||
);
|
||||
assert_eq!(
|
||||
cache_control_for_path(StdPath::new("assets/index.css")),
|
||||
"no-cache"
|
||||
);
|
||||
assert_eq!(
|
||||
cache_control_for_path(StdPath::new("assets/logo.svg")),
|
||||
"public, max-age=604800"
|
||||
|
||||
+110
-4
@@ -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")
|
||||
|
||||
+110
-1
@@ -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(¶m_str(node, "model", DEFAULT_TEXT_TO_VIDEO_MODEL))),
|
||||
json!(normalize_video_model(¶m_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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user