iven/zclaw_openfang
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(auth): 添加异步密码哈希和验证函数
Some checks failed
CI / Lint & TypeCheck (push) Has been cancelled
Details
CI / Unit Tests (push) Has been cancelled
Details
CI / Build Frontend (push) Has been cancelled
Details
CI / Rust Check (push) Has been cancelled
Details
CI / Security Scan (push) Has been cancelled
Details
CI / E2E Tests (push) Has been cancelled
Details

Browse Source 
refactor(relay): 复用HTTP客户端和请求体序列化结果

feat(kernel): 添加获取单个审批记录的方法

fix(store): 改进SaaS连接错误分类和降级处理

docs: 更新审计文档和系统架构文档

refactor(prompt): 优化SQL查询参数化绑定

refactor(migration): 使用静态SQL和COALESCE更新配置项

feat(commands): 添加审批执行状态追踪和事件通知

chore: 更新启动脚本以支持Admin后台

fix(auth-guard): 优化授权状态管理和错误处理

refactor(db): 使用异步密码哈希函数

refactor(totp): 使用异步密码验证函数

style: 清理无用文件和注释

docs: 更新功能全景和审计文档

refactor(service): 优化HTTP客户端重用和请求处理

fix(connection): 改进SaaS不可用时的降级处理

refactor(handlers): 使用异步密码验证函数

chore: 更新依赖和工具链配置
This commit is contained in:
iven
2026-03-29 21:45:29 +08:00
parent b7ec317d2c
commit 7de294375b
34 changed files with 2041 additions and 894 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
crates/zclaw-saas/src/model_config/service.rs
View File

@@ -104,36 +104,38 @@ pub async fn update_provider(
db: &PgPool, provider_id: &str, req: &UpdateProviderRequest, enc_key: &[u8; 32],
) -> SaasResult<ProviderInfo> {
let now = chrono::Utc::now().to_rfc3339();
let mut updates = Vec::new();
let mut params: Vec<Box<dyn std::fmt::Display + Send + Sync>> = Vec::new();
let mut param_idx = 1;
if let Some(ref v) = req.display_name { updates.push(format!("display_name = ${}", param_idx)); params.push(Box::new(v.clone())); param_idx += 1; }
if let Some(ref v) = req.base_url { updates.push(format!("base_url = ${}", param_idx)); params.push(Box::new(v.clone())); param_idx += 1; }
if let Some(ref v) = req.api_protocol { updates.push(format!("api_protocol = ${}", param_idx)); params.push(Box::new(v.clone())); param_idx += 1; }
if let Some(ref v) = req.api_key {
let encrypted = if v.is_empty() { String::new() } else { crypto::encrypt_value(v, enc_key)? };
updates.push(format!("api_key = ${}", param_idx)); params.push(Box::new(encrypted)); param_idx += 1;
}
if let Some(v) = req.enabled { updates.push(format!("enabled = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.rate_limit_rpm { updates.push(format!("rate_limit_rpm = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.rate_limit_tpm { updates.push(format!("rate_limit_tpm = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
// Encrypt api_key upfront if provided
let encrypted_api_key = match req.api_key {
Some(ref v) if !v.is_empty() => Some(crypto::encrypt_value(v, enc_key)?),
Some(ref v) if v.is_empty() => Some(String::new()),
_ => None,
};
if updates.is_empty() {
return get_provider(db, provider_id).await;
}
updates.push(format!("updated_at = ${}", param_idx));
params.push(Box::new(now.clone()));
param_idx += 1;
params.push(Box::new(provider_id.to_string()));
let sql = format!("UPDATE providers SET {} WHERE id = ${}", updates.join(", "), param_idx);
let mut query = sqlx::query(&sql);
for p in &params {
query = query.bind(format!("{}", p));
}
query.execute(db).await?;
// COALESCE pattern: all updatable fields in a single static SQL.
// NULL parameters leave the column unchanged.
sqlx::query(
"UPDATE providers SET
display_name = COALESCE($1, display_name),
base_url = COALESCE($2, base_url),
api_protocol = COALESCE($3, api_protocol),
api_key = COALESCE($4, api_key),
enabled = COALESCE($5, enabled),
rate_limit_rpm = COALESCE($6, rate_limit_rpm),
rate_limit_tpm = COALESCE($7, rate_limit_tpm),
updated_at = $8
WHERE id = $9"
)
.bind(req.display_name.as_deref())
.bind(req.base_url.as_deref())
.bind(req.api_protocol.as_deref())
.bind(encrypted_api_key.as_deref())
.bind(req.enabled)
.bind(req.rate_limit_rpm)
.bind(req.rate_limit_tpm)
.bind(&now)
.bind(provider_id)
.execute(db).await?;
get_provider(db, provider_id).await
}
@@ -245,34 +247,33 @@ pub async fn update_model(
db: &PgPool, model_id: &str, req: &UpdateModelRequest,
) -> SaasResult<ModelInfo> {
let now = chrono::Utc::now().to_rfc3339();
let mut updates = Vec::new();
let mut params: Vec<Box<dyn std::fmt::Display + Send + Sync>> = Vec::new();
let mut param_idx = 1;
if let Some(ref v) = req.alias { updates.push(format!("alias = ${}", param_idx)); params.push(Box::new(v.clone())); param_idx += 1; }
if let Some(v) = req.context_window { updates.push(format!("context_window = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.max_output_tokens { updates.push(format!("max_output_tokens = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.supports_streaming { updates.push(format!("supports_streaming = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.supports_vision { updates.push(format!("supports_vision = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.enabled { updates.push(format!("enabled = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.pricing_input { updates.push(format!("pricing_input = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if let Some(v) = req.pricing_output { updates.push(format!("pricing_output = ${}", param_idx)); params.push(Box::new(v)); param_idx += 1; }
if updates.is_empty() {
return get_model(db, model_id).await;
}
updates.push(format!("updated_at = ${}", param_idx));
params.push(Box::new(now.clone()));
param_idx += 1;
params.push(Box::new(model_id.to_string()));
let sql = format!("UPDATE models SET {} WHERE id = ${}", updates.join(", "), param_idx);
let mut query = sqlx::query(&sql);
for p in &params {
query = query.bind(format!("{}", p));
}
query.execute(db).await?;
// COALESCE pattern: all updatable fields in a single static SQL.
// NULL parameters leave the column unchanged.
sqlx::query(
"UPDATE models SET
alias = COALESCE($1, alias),
context_window = COALESCE($2, context_window),
max_output_tokens = COALESCE($3, max_output_tokens),
supports_streaming = COALESCE($4, supports_streaming),
supports_vision = COALESCE($5, supports_vision),
enabled = COALESCE($6, enabled),
pricing_input = COALESCE($7, pricing_input),
pricing_output = COALESCE($8, pricing_output),
updated_at = $9
WHERE id = $10"
)
.bind(req.alias.as_deref())
.bind(req.context_window)
.bind(req.max_output_tokens)
.bind(req.supports_streaming)
.bind(req.supports_vision)
.bind(req.enabled)
.bind(req.pricing_input)
.bind(req.pricing_output)
.bind(&now)
.bind(model_id)
.execute(db).await?;
get_model(db, model_id).await
}
@@ -401,58 +402,33 @@ pub async fn revoke_account_api_key(
pub async fn get_usage_stats(
db: &PgPool, account_id: &str, query: &UsageQuery,
) -> SaasResult<UsageStats> {
let mut param_idx = 1;
let mut where_clauses = vec![format!("account_id = ${}", param_idx)];
let mut params: Vec<String> = vec![account_id.to_string()];
param_idx += 1;
// Static SQL with conditional filter pattern:
// account_id is always required; optional filters use ($N IS NULL OR col = $N).
let total_sql = "SELECT COUNT(*)::bigint, COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0)
FROM usage_records WHERE account_id = $1 AND ($2 IS NULL OR created_at >= $2) AND ($3 IS NULL OR created_at <= $3) AND ($4 IS NULL OR provider_id = $4) AND ($5 IS NULL OR model_id = $5)";
if let Some(ref from) = query.from {
where_clauses.push(format!("created_at >= ${}", param_idx));
params.push(from.clone());
param_idx += 1;
}
if let Some(ref to) = query.to {
where_clauses.push(format!("created_at <= ${}", param_idx));
params.push(to.clone());
param_idx += 1;
}
if let Some(ref pid) = query.provider_id {
where_clauses.push(format!("provider_id = ${}", param_idx));
params.push(pid.clone());
param_idx += 1;
}
if let Some(ref mid) = query.model_id {
where_clauses.push(format!("model_id = ${}", param_idx));
params.push(mid.clone());
}
let where_sql = where_clauses.join(" AND ");
// 总量统计
let total_sql = format!(
"SELECT COUNT(*)::bigint, COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0)
FROM usage_records WHERE {}", where_sql
);
let mut total_query = sqlx::query(&total_sql);
for p in &params {
total_query = total_query.bind(p);
}
let row = total_query.fetch_one(db).await?;
let row = sqlx::query(total_sql)
.bind(account_id)
.bind(&query.from)
.bind(&query.to)
.bind(&query.provider_id)
.bind(&query.model_id)
.fetch_one(db).await?;
let total_requests: i64 = row.try_get(0).unwrap_or(0);
let total_input: i64 = row.try_get(1).unwrap_or(0);
let total_output: i64 = row.try_get(2).unwrap_or(0);
// 按模型统计
let by_model_sql = format!(
"SELECT provider_id, model_id, COUNT(*)::bigint AS request_count, COALESCE(SUM(input_tokens), 0) AS input_tokens, COALESCE(SUM(output_tokens), 0) AS output_tokens
FROM usage_records WHERE {} GROUP BY provider_id, model_id ORDER BY COUNT(*) DESC LIMIT 20",
where_sql
);
let mut by_model_query = sqlx::query_as::<_, UsageByModelRow>(&by_model_sql);
for p in &params {
by_model_query = by_model_query.bind(p);
}
let by_model_rows = by_model_query.fetch_all(db).await?;
let by_model_sql = "SELECT provider_id, model_id, COUNT(*)::bigint AS request_count, COALESCE(SUM(input_tokens), 0) AS input_tokens, COALESCE(SUM(output_tokens), 0) AS output_tokens
FROM usage_records WHERE account_id = $1 AND ($2 IS NULL OR created_at >= $2) AND ($3 IS NULL OR created_at <= $3) AND ($4 IS NULL OR provider_id = $4) AND ($5 IS NULL OR model_id = $5) GROUP BY provider_id, model_id ORDER BY COUNT(*) DESC LIMIT 20";
let by_model_rows: Vec<UsageByModelRow> = sqlx::query_as(by_model_sql)
.bind(account_id)
.bind(&query.from)
.bind(&query.to)
.bind(&query.provider_id)
.bind(&query.model_id)
.fetch_all(db).await?;
let by_model: Vec<ModelUsage> = by_model_rows.into_iter()
.map(|r| {
ModelUsage { provider_id: r.provider_id, model_id: r.model_id, request_count: r.request_count, input_tokens: r.input_tokens, output_tokens: r.output_tokens }