feat(saas): Phase 2 — 模型配置模块
- Provider CRUD (列表/详情/创建/更新/删除) - Model CRUD (列表/详情/创建/更新/删除) - Account API Key 管理 (创建/轮换/撤销/掩码显示) - Usage 统计 (总量/按模型/按天, 支持时间/供应商/模型过滤) - 权限控制 (provider:manage, model:manage) - 3 个新集成测试覆盖 providers/models/keys
This commit is contained in:
iven
411
crates/zclaw-saas/src/model_config/service.rs
Normal file
411
crates/zclaw-saas/src/model_config/service.rs
Normal file
@@ -0,0 +1,411 @@
|
||||
//! 模型配置业务逻辑
|
||||
|
||||
use sqlx::SqlitePool;
|
||||
use crate::error::{SaasError, SaasResult};
|
||||
use super::types::*;
|
||||
|
||||
// ============ Providers ============
|
||||
|
||||
pub async fn list_providers(db: &SqlitePool) -> SaasResult<Vec<ProviderInfo>> {
|
||||
let rows: Vec<(String, String, String, String, String, bool, Option<i64>, Option<i64>, String, String)> =
|
||||
sqlx::query_as(
|
||||
"SELECT id, name, display_name, base_url, api_protocol, enabled, rate_limit_rpm, rate_limit_tpm, created_at, updated_at
|
||||
FROM providers ORDER BY name"
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
|
||||
Ok(rows.into_iter().map(|(id, name, display_name, base_url, api_protocol, enabled, rpm, tpm, created_at, updated_at)| {
|
||||
ProviderInfo { id, name, display_name, base_url, api_protocol, enabled, rate_limit_rpm: rpm, rate_limit_tpm: tpm, created_at, updated_at }
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn get_provider(db: &SqlitePool, provider_id: &str) -> SaasResult<ProviderInfo> {
|
||||
let row: Option<(String, String, String, String, String, bool, Option<i64>, Option<i64>, String, String)> =
|
||||
sqlx::query_as(
|
||||
"SELECT id, name, display_name, base_url, api_protocol, enabled, rate_limit_rpm, rate_limit_tpm, created_at, updated_at
|
||||
FROM providers WHERE id = ?1"
|
||||
)
|
||||
.bind(provider_id)
|
||||
.fetch_optional(db)
|
||||
.await?;
|
||||
|
||||
let (id, name, display_name, base_url, api_protocol, enabled, rpm, tpm, created_at, updated_at) =
|
||||
row.ok_or_else(|| SaasError::NotFound(format!("Provider {} 不存在", provider_id)))?;
|
||||
|
||||
Ok(ProviderInfo { id, name, display_name, base_url, api_protocol, enabled, rate_limit_rpm: rpm, rate_limit_tpm: tpm, created_at, updated_at })
|
||||
}
|
||||
|
||||
pub async fn create_provider(db: &SqlitePool, req: &CreateProviderRequest) -> SaasResult<ProviderInfo> {
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
|
||||
// 检查名称唯一性
|
||||
let existing: Option<(String,)> = sqlx::query_as("SELECT id FROM providers WHERE name = ?1")
|
||||
.bind(&req.name).fetch_optional(db).await?;
|
||||
if existing.is_some() {
|
||||
return Err(SaasError::AlreadyExists(format!("Provider '{}' 已存在", req.name)));
|
||||
}
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO providers (id, name, display_name, api_key, base_url, api_protocol, enabled, rate_limit_rpm, rate_limit_tpm, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1, ?7, ?8, ?9, ?9)"
|
||||
)
|
||||
.bind(&id).bind(&req.name).bind(&req.display_name).bind(&req.api_key)
|
||||
.bind(&req.base_url).bind(&req.api_protocol).bind(&req.rate_limit_rpm).bind(&req.rate_limit_tpm).bind(&now)
|
||||
.execute(db).await?;
|
||||
|
||||
get_provider(db, &id).await
|
||||
}
|
||||
|
||||
pub async fn update_provider(
|
||||
db: &SqlitePool, provider_id: &str, req: &UpdateProviderRequest,
|
||||
) -> 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();
|
||||
|
||||
if let Some(ref v) = req.display_name { updates.push("display_name = ?"); params.push(Box::new(v.clone())); }
|
||||
if let Some(ref v) = req.base_url { updates.push("base_url = ?"); params.push(Box::new(v.clone())); }
|
||||
if let Some(ref v) = req.api_protocol { updates.push("api_protocol = ?"); params.push(Box::new(v.clone())); }
|
||||
if let Some(ref v) = req.api_key { updates.push("api_key = ?"); params.push(Box::new(v.clone())); }
|
||||
if let Some(v) = req.enabled { updates.push("enabled = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.rate_limit_rpm { updates.push("rate_limit_rpm = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.rate_limit_tpm { updates.push("rate_limit_tpm = ?"); params.push(Box::new(v)); }
|
||||
|
||||
if updates.is_empty() {
|
||||
return get_provider(db, provider_id).await;
|
||||
}
|
||||
|
||||
updates.push("updated_at = ?");
|
||||
params.push(Box::new(now.clone()));
|
||||
params.push(Box::new(provider_id.to_string()));
|
||||
|
||||
let sql = format!("UPDATE providers SET {} WHERE id = ?", updates.join(", "));
|
||||
let mut query = sqlx::query(&sql);
|
||||
for p in ¶ms {
|
||||
query = query.bind(format!("{}", p));
|
||||
}
|
||||
query.execute(db).await?;
|
||||
|
||||
get_provider(db, provider_id).await
|
||||
}
|
||||
|
||||
pub async fn delete_provider(db: &SqlitePool, provider_id: &str) -> SaasResult<()> {
|
||||
let result = sqlx::query("DELETE FROM providers WHERE id = ?1")
|
||||
.bind(provider_id).execute(db).await?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(SaasError::NotFound(format!("Provider {} 不存在", provider_id)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ============ Models ============
|
||||
|
||||
pub async fn list_models(db: &SqlitePool, provider_id: Option<&str>) -> SaasResult<Vec<ModelInfo>> {
|
||||
let sql = if provider_id.is_some() {
|
||||
"SELECT id, provider_id, model_id, alias, context_window, max_output_tokens, supports_streaming, supports_vision, enabled, pricing_input, pricing_output, created_at, updated_at
|
||||
FROM models WHERE provider_id = ?1 ORDER BY alias"
|
||||
} else {
|
||||
"SELECT id, provider_id, model_id, alias, context_window, max_output_tokens, supports_streaming, supports_vision, enabled, pricing_input, pricing_output, created_at, updated_at
|
||||
FROM models ORDER BY provider_id, alias"
|
||||
};
|
||||
|
||||
let mut query = sqlx::query_as::<_, (String, String, String, String, i64, i64, bool, bool, bool, f64, f64, String, String)>(sql);
|
||||
if let Some(pid) = provider_id {
|
||||
query = query.bind(pid);
|
||||
}
|
||||
|
||||
let rows = query.fetch_all(db).await?;
|
||||
Ok(rows.into_iter().map(|(id, provider_id, model_id, alias, ctx, max_out, streaming, vision, enabled, pi, po, created_at, updated_at)| {
|
||||
ModelInfo { id, provider_id, model_id, alias, context_window: ctx, max_output_tokens: max_out, supports_streaming: streaming, supports_vision: vision, enabled, pricing_input: pi, pricing_output: po, created_at, updated_at }
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn create_model(db: &SqlitePool, req: &CreateModelRequest) -> SaasResult<ModelInfo> {
|
||||
// 验证 provider 存在
|
||||
let provider = get_provider(db, &req.provider_id).await?;
|
||||
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
|
||||
// 检查 model 唯一性
|
||||
let existing: Option<(String,)> = sqlx::query_as(
|
||||
"SELECT id FROM models WHERE provider_id = ?1 AND model_id = ?2"
|
||||
)
|
||||
.bind(&req.provider_id).bind(&req.model_id)
|
||||
.fetch_optional(db).await?;
|
||||
|
||||
if existing.is_some() {
|
||||
return Err(SaasError::AlreadyExists(format!(
|
||||
"模型 '{}' 已存在于 provider '{}'", req.model_id, provider.name
|
||||
)));
|
||||
}
|
||||
|
||||
let ctx = req.context_window.unwrap_or(8192);
|
||||
let max_out = req.max_output_tokens.unwrap_or(4096);
|
||||
let streaming = req.supports_streaming.unwrap_or(true);
|
||||
let vision = req.supports_vision.unwrap_or(false);
|
||||
let pi = req.pricing_input.unwrap_or(0.0);
|
||||
let po = req.pricing_output.unwrap_or(0.0);
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO models (id, provider_id, model_id, alias, context_window, max_output_tokens, supports_streaming, supports_vision, enabled, pricing_input, pricing_output, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 1, ?9, ?10, ?11, ?11)"
|
||||
)
|
||||
.bind(&id).bind(&req.provider_id).bind(&req.model_id).bind(&req.alias)
|
||||
.bind(ctx).bind(max_out).bind(streaming).bind(vision).bind(pi).bind(po).bind(&now)
|
||||
.execute(db).await?;
|
||||
|
||||
get_model(db, &id).await
|
||||
}
|
||||
|
||||
pub async fn get_model(db: &SqlitePool, model_id: &str) -> SaasResult<ModelInfo> {
|
||||
let row: Option<(String, String, String, String, i64, i64, bool, bool, bool, f64, f64, String, String)> =
|
||||
sqlx::query_as(
|
||||
"SELECT id, provider_id, model_id, alias, context_window, max_output_tokens, supports_streaming, supports_vision, enabled, pricing_input, pricing_output, created_at, updated_at
|
||||
FROM models WHERE id = ?1"
|
||||
)
|
||||
.bind(model_id)
|
||||
.fetch_optional(db)
|
||||
.await?;
|
||||
|
||||
let (id, provider_id, model_id, alias, ctx, max_out, streaming, vision, enabled, pi, po, created_at, updated_at) =
|
||||
row.ok_or_else(|| SaasError::NotFound(format!("模型 {} 不存在", model_id)))?;
|
||||
|
||||
Ok(ModelInfo { id, provider_id, model_id, alias, context_window: ctx, max_output_tokens: max_out, supports_streaming: streaming, supports_vision: vision, enabled, pricing_input: pi, pricing_output: po, created_at, updated_at })
|
||||
}
|
||||
|
||||
pub async fn update_model(
|
||||
db: &SqlitePool, 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();
|
||||
|
||||
if let Some(ref v) = req.alias { updates.push("alias = ?"); params.push(Box::new(v.clone())); }
|
||||
if let Some(v) = req.context_window { updates.push("context_window = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.max_output_tokens { updates.push("max_output_tokens = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.supports_streaming { updates.push("supports_streaming = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.supports_vision { updates.push("supports_vision = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.enabled { updates.push("enabled = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.pricing_input { updates.push("pricing_input = ?"); params.push(Box::new(v)); }
|
||||
if let Some(v) = req.pricing_output { updates.push("pricing_output = ?"); params.push(Box::new(v)); }
|
||||
|
||||
if updates.is_empty() {
|
||||
return get_model(db, model_id).await;
|
||||
}
|
||||
|
||||
updates.push("updated_at = ?");
|
||||
params.push(Box::new(now.clone()));
|
||||
params.push(Box::new(model_id.to_string()));
|
||||
|
||||
let sql = format!("UPDATE models SET {} WHERE id = ?", updates.join(", "));
|
||||
let mut query = sqlx::query(&sql);
|
||||
for p in ¶ms {
|
||||
query = query.bind(format!("{}", p));
|
||||
}
|
||||
query.execute(db).await?;
|
||||
|
||||
get_model(db, model_id).await
|
||||
}
|
||||
|
||||
pub async fn delete_model(db: &SqlitePool, model_id: &str) -> SaasResult<()> {
|
||||
let result = sqlx::query("DELETE FROM models WHERE id = ?1")
|
||||
.bind(model_id).execute(db).await?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(SaasError::NotFound(format!("模型 {} 不存在", model_id)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ============ Account API Keys ============
|
||||
|
||||
pub async fn list_account_api_keys(
|
||||
db: &SqlitePool, account_id: &str, provider_id: Option<&str>,
|
||||
) -> SaasResult<Vec<AccountApiKeyInfo>> {
|
||||
let sql = if provider_id.is_some() {
|
||||
"SELECT id, provider_id, key_label, permissions, enabled, last_used_at, created_at, key_value
|
||||
FROM account_api_keys WHERE account_id = ?1 AND provider_id = ?2 AND revoked_at IS NULL ORDER BY created_at DESC"
|
||||
} else {
|
||||
"SELECT id, provider_id, key_label, permissions, enabled, last_used_at, created_at, key_value
|
||||
FROM account_api_keys WHERE account_id = ?1 AND revoked_at IS NULL ORDER BY created_at DESC"
|
||||
};
|
||||
|
||||
let mut query = sqlx::query_as::<_, (String, String, Option<String>, String, bool, Option<String>, String, String)>(sql)
|
||||
.bind(account_id);
|
||||
if let Some(pid) = provider_id {
|
||||
query = query.bind(pid);
|
||||
}
|
||||
|
||||
let rows = query.fetch_all(db).await?;
|
||||
Ok(rows.into_iter().map(|(id, provider_id, key_label, perms, enabled, last_used, created_at, key_value)| {
|
||||
let permissions: Vec<String> = serde_json::from_str(&perms).unwrap_or_default();
|
||||
let masked = mask_api_key(&key_value);
|
||||
AccountApiKeyInfo { id, provider_id, key_label, permissions, enabled, last_used_at: last_used, created_at, masked_key: masked }
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn create_account_api_key(
|
||||
db: &SqlitePool, account_id: &str, req: &CreateAccountApiKeyRequest,
|
||||
) -> SaasResult<AccountApiKeyInfo> {
|
||||
// 验证 provider 存在
|
||||
get_provider(db, &req.provider_id).await?;
|
||||
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
let permissions = serde_json::to_string(&req.permissions)?;
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO account_api_keys (id, account_id, provider_id, key_value, key_label, permissions, enabled, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1, ?7, ?7)"
|
||||
)
|
||||
.bind(&id).bind(account_id).bind(&req.provider_id).bind(&req.key_value)
|
||||
.bind(&req.key_label).bind(&permissions).bind(&now)
|
||||
.execute(db).await?;
|
||||
|
||||
let masked = mask_api_key(&req.key_value);
|
||||
Ok(AccountApiKeyInfo {
|
||||
id, provider_id: req.provider_id.clone(), key_label: req.key_label.clone(),
|
||||
permissions: req.permissions.clone(), enabled: true, last_used_at: None,
|
||||
created_at: now, masked_key: masked,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn rotate_account_api_key(
|
||||
db: &SqlitePool, key_id: &str, account_id: &str, new_key_value: &str,
|
||||
) -> SaasResult<()> {
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
let result = sqlx::query(
|
||||
"UPDATE account_api_keys SET key_value = ?1, updated_at = ?2 WHERE id = ?3 AND account_id = ?4 AND revoked_at IS NULL"
|
||||
)
|
||||
.bind(new_key_value).bind(&now).bind(key_id).bind(account_id)
|
||||
.execute(db).await?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(SaasError::NotFound("API Key 不存在或已撤销".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn revoke_account_api_key(
|
||||
db: &SqlitePool, key_id: &str, account_id: &str,
|
||||
) -> SaasResult<()> {
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
let result = sqlx::query(
|
||||
"UPDATE account_api_keys SET revoked_at = ?1 WHERE id = ?2 AND account_id = ?3 AND revoked_at IS NULL"
|
||||
)
|
||||
.bind(&now).bind(key_id).bind(account_id)
|
||||
.execute(db).await?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(SaasError::NotFound("API Key 不存在或已撤销".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ============ Usage Statistics ============
|
||||
|
||||
pub async fn get_usage_stats(
|
||||
db: &SqlitePool, account_id: &str, query: &UsageQuery,
|
||||
) -> SaasResult<UsageStats> {
|
||||
let mut where_clauses = vec!["account_id = ?".to_string()];
|
||||
let mut params: Vec<String> = vec![account_id.to_string()];
|
||||
|
||||
if let Some(ref from) = query.from {
|
||||
where_clauses.push("created_at >= ?".to_string());
|
||||
params.push(from.clone());
|
||||
}
|
||||
if let Some(ref to) = query.to {
|
||||
where_clauses.push("created_at <= ?".to_string());
|
||||
params.push(to.clone());
|
||||
}
|
||||
if let Some(ref pid) = query.provider_id {
|
||||
where_clauses.push("provider_id = ?".to_string());
|
||||
params.push(pid.clone());
|
||||
}
|
||||
if let Some(ref mid) = query.model_id {
|
||||
where_clauses.push("model_id = ?".to_string());
|
||||
params.push(mid.clone());
|
||||
}
|
||||
|
||||
let where_sql = where_clauses.join(" AND ");
|
||||
|
||||
// 总量统计
|
||||
let total_sql = format!(
|
||||
"SELECT COUNT(*), COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0)
|
||||
FROM usage_records WHERE {}", where_sql
|
||||
);
|
||||
let mut total_query = sqlx::query_as::<_, (i64, i64, i64)>(&total_sql);
|
||||
for p in ¶ms {
|
||||
total_query = total_query.bind(p);
|
||||
}
|
||||
let (total_requests, total_input, total_output) = total_query.fetch_one(db).await?;
|
||||
|
||||
// 按模型统计
|
||||
let by_model_sql = format!(
|
||||
"SELECT provider_id, model_id, COUNT(*), COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0)
|
||||
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::<_, (String, String, i64, i64, i64)>(&by_model_sql);
|
||||
for p in ¶ms {
|
||||
by_model_query = by_model_query.bind(p);
|
||||
}
|
||||
let by_model_rows = by_model_query.fetch_all(db).await?;
|
||||
let by_model: Vec<ModelUsage> = by_model_rows.into_iter()
|
||||
.map(|(provider_id, model_id, count, input, output)| {
|
||||
ModelUsage { provider_id, model_id, request_count: count, input_tokens: input, output_tokens: output }
|
||||
}).collect();
|
||||
|
||||
// 按天统计 (最近 30 天)
|
||||
let from_30d = (chrono::Utc::now() - chrono::Duration::days(30)).to_rfc3339();
|
||||
let daily_sql = format!(
|
||||
"SELECT DATE(created_at) as day, COUNT(*), COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0)
|
||||
FROM usage_records WHERE account_id = ?1 AND created_at >= ?2
|
||||
GROUP BY DATE(created_at) ORDER BY day DESC LIMIT 30"
|
||||
);
|
||||
let daily_rows: Vec<(String, i64, i64, i64)> = sqlx::query_as(&daily_sql)
|
||||
.bind(account_id).bind(&from_30d)
|
||||
.fetch_all(db).await?;
|
||||
let by_day: Vec<DailyUsage> = daily_rows.into_iter()
|
||||
.map(|(date, count, input, output)| {
|
||||
DailyUsage { date, request_count: count, input_tokens: input, output_tokens: output }
|
||||
}).collect();
|
||||
|
||||
Ok(UsageStats {
|
||||
total_requests,
|
||||
total_input_tokens: total_input,
|
||||
total_output_tokens: total_output,
|
||||
by_model,
|
||||
by_day,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn record_usage(
|
||||
db: &SqlitePool, account_id: &str, provider_id: &str, model_id: &str,
|
||||
input_tokens: i64, output_tokens: i64, latency_ms: Option<i64>,
|
||||
status: &str, error_message: Option<&str>,
|
||||
) -> SaasResult<()> {
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
sqlx::query(
|
||||
"INSERT INTO usage_records (account_id, provider_id, model_id, input_tokens, output_tokens, latency_ms, status, error_message, created_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)"
|
||||
)
|
||||
.bind(account_id).bind(provider_id).bind(model_id)
|
||||
.bind(input_tokens).bind(output_tokens).bind(latency_ms)
|
||||
.bind(status).bind(error_message).bind(&now)
|
||||
.execute(db).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ============ Helpers ============
|
||||
|
||||
fn mask_api_key(key: &str) -> String {
|
||||
if key.len() <= 8 {
|
||||
return "*".repeat(key.len());
|
||||
}
|
||||
format!("{}...{}", &key[..4], &key[key.len()-4..])
|
||||
}
|
||||
Reference in New Issue
Block a user