Files
cc-switch/src-tauri/src/services/usage_stats.rs
T
Jason aa5e58d060 fix(usage): correct cache cost semantics and silence pricing warn storm
- Split CostCalculator into per-app cache semantics: Anthropic's
  input_tokens is already fresh input, while Codex/Gemini include
  cached tokens in their prompt count. The old shared formula
  double-subtracted cache_read for Claude, under-billing input cost.
- Backfill now reads cost_multiplier from the per-log snapshot column
  instead of re-querying providers.meta, so historical rows are no
  longer rewritten with the current multiplier.
- Move the "pricing not found" warn out of find_model_pricing_row;
  emit it only when a brand new log is written, and skip placeholder
  models (unknown / empty / null / none) entirely.
- Broaden model id normalization: strip namespace prefixes
  (anthropic./openai./global./bedrock.), bedrock-style -vN suffixes,
  reasoning effort suffixes (-low/-medium/-high/-xhigh/-minimal),
  Claude Desktop's claude-<non-anthropic> wrapper, dot-to-dash for
  Claude, and try a LIKE prefix match for Claude short route ids
  (e.g. claude-haiku-4-5 -> claude-haiku-4-5-20251001).
- Fall back to request_model when the stored model is missing, so
  early Codex session rows with model=unknown can still be priced.
2026-05-13 17:51:35 +08:00

3125 lines
110 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! 使用统计服务
//!
//! 提供使用量数据的聚合查询功能
use crate::database::{lock_conn, Database};
use crate::error::AppError;
use crate::services::sql_helpers::fresh_input_sql;
use chrono::{Local, NaiveDate, TimeZone, Timelike};
use rusqlite::{params, Connection, OptionalExtension};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::str::FromStr;
/// 使用量汇总
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageSummary {
pub total_requests: u64,
pub total_cost: String,
pub total_input_tokens: u64,
pub total_output_tokens: u64,
pub total_cache_creation_tokens: u64,
pub total_cache_read_tokens: u64,
pub success_rate: f32,
/// input + output + cache_creation + cache_read — the total tokens
/// actually processed by the model (including cache hits). Used as the
/// headline "real consumption" number in the usage hero.
pub real_total_tokens: u64,
/// cache_read / (input + cache_creation + cache_read). Range 0.01.0.
/// Reported as a fraction; multiply by 100 in UI for percentage display.
pub cache_hit_rate: f64,
}
/// Per-app-type usage summary used by the dashboard breakdown rail.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageSummaryByApp {
pub app_type: String,
pub summary: UsageSummary,
}
/// Helper: compute (real_total, hit_rate) from the four token counters.
/// All inputs must already be cache-normalized (i.e. input excludes cache).
fn derive_real_total_and_hit_rate(
fresh_input: u64,
output: u64,
cache_creation: u64,
cache_read: u64,
) -> (u64, f64) {
let real_total = fresh_input + output + cache_creation + cache_read;
let cacheable_input = fresh_input + cache_creation + cache_read;
let hit_rate = if cacheable_input > 0 {
cache_read as f64 / cacheable_input as f64
} else {
0.0
};
(real_total, hit_rate)
}
/// 每日统计
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DailyStats {
pub date: String,
pub request_count: u64,
pub total_cost: String,
pub total_tokens: u64,
pub total_input_tokens: u64,
pub total_output_tokens: u64,
pub total_cache_creation_tokens: u64,
pub total_cache_read_tokens: u64,
}
/// Provider 统计
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProviderStats {
pub provider_id: String,
pub provider_name: String,
pub request_count: u64,
pub total_tokens: u64,
pub total_cost: String,
pub success_rate: f32,
pub avg_latency_ms: u64,
}
/// 模型统计
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ModelStats {
pub model: String,
pub request_count: u64,
pub total_tokens: u64,
pub total_cost: String,
pub avg_cost_per_request: String,
}
/// 请求日志过滤器
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LogFilters {
pub app_type: Option<String>,
pub provider_name: Option<String>,
pub model: Option<String>,
pub status_code: Option<u16>,
pub start_date: Option<i64>,
pub end_date: Option<i64>,
}
/// 分页请求日志响应
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PaginatedLogs {
pub data: Vec<RequestLogDetail>,
pub total: u32,
pub page: u32,
pub page_size: u32,
}
/// 请求日志详情
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RequestLogDetail {
pub request_id: String,
pub provider_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub provider_name: Option<String>,
pub app_type: String,
pub model: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub request_model: Option<String>,
pub cost_multiplier: String,
pub input_tokens: u32,
pub output_tokens: u32,
pub cache_read_tokens: u32,
pub cache_creation_tokens: u32,
pub input_cost_usd: String,
pub output_cost_usd: String,
pub cache_read_cost_usd: String,
pub cache_creation_cost_usd: String,
pub total_cost_usd: String,
pub is_streaming: bool,
pub latency_ms: u64,
pub first_token_ms: Option<u64>,
pub duration_ms: Option<u64>,
pub status_code: u16,
pub error_message: Option<String>,
pub created_at: i64,
#[serde(skip_serializing_if = "Option::is_none")]
pub data_source: Option<String>,
}
/// 把 24 列的查询结果映射为 `RequestLogDetail`。
///
/// 调用方的 SELECT **必须**按以下顺序返回 24 列:
/// `request_id, provider_id, provider_name, app_type, model, request_model,
/// cost_multiplier, input_tokens, output_tokens, cache_read_tokens,
/// cache_creation_tokens, input_cost_usd, output_cost_usd, cache_read_cost_usd,
/// cache_creation_cost_usd, total_cost_usd, is_streaming, latency_ms,
/// first_token_ms, duration_ms, status_code, error_message, created_at,
/// data_source`
///
/// 不需要 provider_name 时(如 backfillSELECT `NULL AS provider_name` 占位即可。
fn row_to_request_log_detail(row: &rusqlite::Row<'_>) -> rusqlite::Result<RequestLogDetail> {
Ok(RequestLogDetail {
request_id: row.get(0)?,
provider_id: row.get(1)?,
provider_name: row.get(2)?,
app_type: row.get(3)?,
model: row.get(4)?,
request_model: row.get(5)?,
cost_multiplier: row
.get::<_, Option<String>>(6)?
.unwrap_or_else(|| "1".to_string()),
input_tokens: row.get::<_, i64>(7)? as u32,
output_tokens: row.get::<_, i64>(8)? as u32,
cache_read_tokens: row.get::<_, i64>(9)? as u32,
cache_creation_tokens: row.get::<_, i64>(10)? as u32,
input_cost_usd: row.get(11)?,
output_cost_usd: row.get(12)?,
cache_read_cost_usd: row.get(13)?,
cache_creation_cost_usd: row.get(14)?,
total_cost_usd: row.get(15)?,
is_streaming: row.get::<_, i64>(16)? != 0,
latency_ms: row.get::<_, i64>(17)? as u64,
first_token_ms: row.get::<_, Option<i64>>(18)?.map(|v| v as u64),
duration_ms: row.get::<_, Option<i64>>(19)?.map(|v| v as u64),
status_code: row.get::<_, i64>(20)? as u16,
error_message: row.get(21)?,
created_at: row.get(22)?,
data_source: row.get(23)?,
})
}
/// SQL fragment: resolve provider_name with fallback for session-based entries.
/// Session logs use placeholder provider_ids (e.g., `_session`, `_<app>_session`)
/// that don't exist in the providers table — the CASE expression below is the
/// authoritative mapping from placeholder to readable name.
fn provider_name_coalesce(log_alias: &str, provider_alias: &str) -> String {
format!(
"COALESCE({provider_alias}.name, CASE {log_alias}.provider_id \
WHEN '_session' THEN 'Claude (Session)' \
WHEN '_codex_session' THEN 'Codex (Session)' \
WHEN '_gemini_session' THEN 'Gemini (Session)' \
ELSE {log_alias}.provider_id END)"
)
}
pub(crate) const SESSION_PROXY_DEDUP_WINDOW_SECONDS: i64 = 10 * 60;
/// SQL 片段:把指定别名的 `data_source` 包成 COALESCENULL 视作 'proxy'。
///
/// 防御 schema v9 之前可能写入的 NULL data_source 行(见
/// `tests::create_legacy_nullable_logs_table`)。所有用到 data_source 的查询
/// 都应通过此 helper 生成片段,避免遗漏。
fn data_source_expr(log_alias: &str) -> String {
format!("COALESCE({log_alias}.data_source, 'proxy')")
}
pub(crate) fn effective_usage_log_filter(log_alias: &str) -> String {
let data_source = data_source_expr(log_alias);
let proxy_data_source = data_source_expr("proxy_dedup");
format!(
"NOT (
{data_source} IN ('session_log', 'codex_session', 'gemini_session')
AND EXISTS (
SELECT 1
FROM proxy_request_logs proxy_dedup
WHERE {proxy_data_source} = 'proxy'
AND proxy_dedup.app_type = {log_alias}.app_type
AND proxy_dedup.status_code >= 200
AND proxy_dedup.status_code < 300
AND proxy_dedup.input_tokens = {log_alias}.input_tokens
AND proxy_dedup.output_tokens = {log_alias}.output_tokens
AND proxy_dedup.cache_read_tokens = {log_alias}.cache_read_tokens
AND (
proxy_dedup.cache_creation_tokens = {log_alias}.cache_creation_tokens
OR (
{log_alias}.cache_creation_tokens = 0
AND {data_source} IN ('codex_session', 'gemini_session')
)
)
AND proxy_dedup.created_at BETWEEN
{log_alias}.created_at - {SESSION_PROXY_DEDUP_WINDOW_SECONDS}
AND {log_alias}.created_at + {SESSION_PROXY_DEDUP_WINDOW_SECONDS}
AND (
LOWER(proxy_dedup.model) = LOWER({log_alias}.model)
OR LOWER(proxy_dedup.model) = 'unknown'
OR LOWER({log_alias}.model) = 'unknown'
)
)
)"
)
}
/// 跨源去重指纹键。
///
/// `cache_creation_tokens`Codex/Gemini session 日志不暴露该字段,调用方传 0
/// 表示"未知",匹配器会放行 proxy 侧任意 cache_creation_tokens 值。
#[derive(Debug, Clone, Copy)]
pub(crate) struct DedupKey<'a> {
pub app_type: &'a str,
pub model: &'a str,
pub input_tokens: u32,
pub output_tokens: u32,
pub cache_read_tokens: u32,
pub cache_creation_tokens: u32,
pub created_at: i64,
}
/// session 日志写入前的统一去重判定。
///
/// 命中以下任一条件即跳过插入:① `request_id` 已存在;② 时间窗口内存在
/// 与 `key` 匹配的 proxy 日志(指纹去重)。
pub(crate) fn should_skip_session_insert(
conn: &Connection,
request_id: &str,
key: &DedupKey,
) -> Result<bool, AppError> {
if proxy_request_id_exists(conn, request_id)? {
return Ok(true);
}
has_matching_proxy_usage_log(conn, key)
}
fn proxy_request_id_exists(conn: &Connection, request_id: &str) -> Result<bool, AppError> {
conn.query_row(
"SELECT EXISTS(SELECT 1 FROM proxy_request_logs WHERE request_id = ?1)",
params![request_id],
|row| row.get::<_, bool>(0),
)
.map_err(|e| AppError::Database(format!("查询 request_id 失败: {e}")))
}
pub(crate) fn has_matching_proxy_usage_log(
conn: &Connection,
key: &DedupKey,
) -> Result<bool, AppError> {
let allow_missing_cache_creation =
matches!(key.app_type, "codex" | "gemini") && key.cache_creation_tokens == 0;
let l_data_source = data_source_expr("l");
let sql = format!(
"SELECT EXISTS (
SELECT 1
FROM proxy_request_logs l
WHERE {l_data_source} = 'proxy'
AND l.app_type = ?1
AND l.status_code >= 200
AND l.status_code < 300
AND l.input_tokens = ?3
AND l.output_tokens = ?4
AND l.cache_read_tokens = ?5
AND (l.cache_creation_tokens = ?6 OR ?9 = 1)
AND l.created_at BETWEEN ?7 - ?8 AND ?7 + ?8
AND (
LOWER(l.model) = LOWER(?2)
OR LOWER(l.model) = 'unknown'
OR LOWER(?2) = 'unknown'
)
)"
);
conn.query_row(
&sql,
params![
key.app_type,
key.model,
key.input_tokens as i64,
key.output_tokens as i64,
key.cache_read_tokens as i64,
key.cache_creation_tokens as i64,
key.created_at,
SESSION_PROXY_DEDUP_WINDOW_SECONDS,
allow_missing_cache_creation as i64,
],
|row| row.get::<_, bool>(0),
)
.map_err(|e| AppError::Database(format!("查询重复代理用量日志失败: {e}")))
}
#[derive(Debug, Clone, Default)]
struct RollupDateBounds {
start: Option<String>,
end: Option<String>,
is_empty: bool,
}
fn local_datetime_from_timestamp(ts: i64) -> Result<chrono::DateTime<Local>, AppError> {
Local
.timestamp_opt(ts, 0)
.single()
.ok_or_else(|| AppError::Database(format!("无法解析本地时间戳: {ts}")))
}
fn compute_rollup_date_bounds(
start_ts: Option<i64>,
end_ts: Option<i64>,
) -> Result<RollupDateBounds, AppError> {
let start = match start_ts {
Some(ts) => {
let local = local_datetime_from_timestamp(ts)?;
let day = local.date_naive();
if local.time().num_seconds_from_midnight() == 0 {
Some(day.format("%Y-%m-%d").to_string())
} else {
day.succ_opt()
.map(|next| next.format("%Y-%m-%d").to_string())
}
}
None => None,
};
let end = match end_ts {
Some(ts) => {
let local = local_datetime_from_timestamp(ts)?;
let day = local.date_naive();
if local.time().hour() == 23 && local.time().minute() == 59 {
Some(day.format("%Y-%m-%d").to_string())
} else {
day.pred_opt()
.map(|prev| prev.format("%Y-%m-%d").to_string())
}
}
None => None,
};
let is_empty = matches!((&start, &end), (Some(start), Some(end)) if start > end);
Ok(RollupDateBounds {
start,
end,
is_empty,
})
}
fn push_rollup_date_filters(
conditions: &mut Vec<String>,
params: &mut Vec<Box<dyn rusqlite::ToSql>>,
column: &str,
bounds: &RollupDateBounds,
) {
if bounds.is_empty {
conditions.push("1 = 0".to_string());
return;
}
if let Some(start) = &bounds.start {
conditions.push(format!("{column} >= ?"));
params.push(Box::new(start.clone()));
}
if let Some(end) = &bounds.end {
conditions.push(format!("{column} <= ?"));
params.push(Box::new(end.clone()));
}
}
fn local_day_start_rfc3339(day: NaiveDate) -> String {
let local_midnight = day
.and_hms_opt(0, 0, 0)
.and_then(|naive| match Local.from_local_datetime(&naive) {
chrono::LocalResult::Single(dt) => Some(dt),
chrono::LocalResult::Ambiguous(earliest, _) => Some(earliest),
chrono::LocalResult::None => None,
})
.unwrap_or_else(Local::now);
local_midnight.to_rfc3339()
}
impl Database {
/// 获取使用量汇总
pub fn get_usage_summary(
&self,
start_date: Option<i64>,
end_date: Option<i64>,
app_type: Option<&str>,
) -> Result<UsageSummary, AppError> {
let conn = lock_conn!(self.conn);
// Build detail WHERE clause
let mut conditions = vec![effective_usage_log_filter("l")];
let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(start) = start_date {
conditions.push("l.created_at >= ?".to_string());
params_vec.push(Box::new(start));
}
if let Some(end) = end_date {
conditions.push("l.created_at <= ?".to_string());
params_vec.push(Box::new(end));
}
if let Some(at) = app_type {
conditions.push("l.app_type = ?".to_string());
params_vec.push(Box::new(at.to_string()));
}
let where_clause = if conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", conditions.join(" AND "))
};
// Only include rolled-up rows for full local days that are fully covered by the range.
let mut rollup_conditions: Vec<String> = Vec::new();
let mut rollup_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
let rollup_bounds = compute_rollup_date_bounds(start_date, end_date)?;
push_rollup_date_filters(
&mut rollup_conditions,
&mut rollup_params,
"date",
&rollup_bounds,
);
if let Some(at) = app_type {
rollup_conditions.push("app_type = ?".to_string());
rollup_params.push(Box::new(at.to_string()));
}
let rollup_where = if rollup_conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", rollup_conditions.join(" AND "))
};
let fresh_input_detail = fresh_input_sql("l");
let fresh_input_rollup = fresh_input_sql("");
let sql = format!(
"SELECT
COALESCE(d.total_requests, 0) + COALESCE(r.total_requests, 0),
COALESCE(d.total_cost, 0) + COALESCE(r.total_cost, 0),
COALESCE(d.total_input_tokens, 0) + COALESCE(r.total_input_tokens, 0),
COALESCE(d.total_output_tokens, 0) + COALESCE(r.total_output_tokens, 0),
COALESCE(d.total_cache_creation_tokens, 0) + COALESCE(r.total_cache_creation_tokens, 0),
COALESCE(d.total_cache_read_tokens, 0) + COALESCE(r.total_cache_read_tokens, 0),
COALESCE(d.success_count, 0) + COALESCE(r.success_count, 0)
FROM
(SELECT
COUNT(*) as total_requests,
COALESCE(SUM(CAST(l.total_cost_usd AS REAL)), 0) as total_cost,
COALESCE(SUM({fresh_input_detail}), 0) as total_input_tokens,
COALESCE(SUM(l.output_tokens), 0) as total_output_tokens,
COALESCE(SUM(l.cache_creation_tokens), 0) as total_cache_creation_tokens,
COALESCE(SUM(l.cache_read_tokens), 0) as total_cache_read_tokens,
COALESCE(SUM(CASE WHEN l.status_code >= 200 AND l.status_code < 300 THEN 1 ELSE 0 END), 0) as success_count
FROM proxy_request_logs l {where_clause}) d,
(SELECT
COALESCE(SUM(request_count), 0) as total_requests,
COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0) as total_cost,
COALESCE(SUM({fresh_input_rollup}), 0) as total_input_tokens,
COALESCE(SUM(output_tokens), 0) as total_output_tokens,
COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation_tokens,
COALESCE(SUM(cache_read_tokens), 0) as total_cache_read_tokens,
COALESCE(SUM(success_count), 0) as success_count
FROM usage_daily_rollups {rollup_where}) r"
);
// Combine params: detail params first, then rollup params
let mut all_params: Vec<Box<dyn rusqlite::ToSql>> = params_vec;
all_params.extend(rollup_params);
let param_refs: Vec<&dyn rusqlite::ToSql> = all_params.iter().map(|p| p.as_ref()).collect();
let result = conn.query_row(&sql, param_refs.as_slice(), |row| {
let total_requests: i64 = row.get(0)?;
let total_cost: f64 = row.get(1)?;
let total_input_tokens: i64 = row.get(2)?;
let total_output_tokens: i64 = row.get(3)?;
let total_cache_creation_tokens: i64 = row.get(4)?;
let total_cache_read_tokens: i64 = row.get(5)?;
let success_count: i64 = row.get(6)?;
let success_rate = if total_requests > 0 {
(success_count as f32 / total_requests as f32) * 100.0
} else {
0.0
};
let (real_total_tokens, cache_hit_rate) = derive_real_total_and_hit_rate(
total_input_tokens as u64,
total_output_tokens as u64,
total_cache_creation_tokens as u64,
total_cache_read_tokens as u64,
);
Ok(UsageSummary {
total_requests: total_requests as u64,
total_cost: format!("{total_cost:.6}"),
total_input_tokens: total_input_tokens as u64,
total_output_tokens: total_output_tokens as u64,
total_cache_creation_tokens: total_cache_creation_tokens as u64,
total_cache_read_tokens: total_cache_read_tokens as u64,
success_rate,
real_total_tokens,
cache_hit_rate,
})
})?;
Ok(result)
}
/// 按 app_type 维度拆分的使用量汇总,用于 Dashboard 的分应用展示条。
/// 返回所有有数据的 app_type,按 real_total_tokens 降序。
///
/// Single SQL with `GROUP BY app_type` — avoids the N+1 round-trip that
/// would result from invoking `get_usage_summary` once per app_type.
pub fn get_usage_summary_by_app(
&self,
start_date: Option<i64>,
end_date: Option<i64>,
) -> Result<Vec<UsageSummaryByApp>, AppError> {
let conn = lock_conn!(self.conn);
let mut detail_conditions = vec![effective_usage_log_filter("l")];
let mut detail_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(start) = start_date {
detail_conditions.push("l.created_at >= ?".to_string());
detail_params.push(Box::new(start));
}
if let Some(end) = end_date {
detail_conditions.push("l.created_at <= ?".to_string());
detail_params.push(Box::new(end));
}
let detail_where = format!("WHERE {}", detail_conditions.join(" AND "));
let rollup_bounds = compute_rollup_date_bounds(start_date, end_date)?;
let mut rollup_conditions: Vec<String> = Vec::new();
let mut rollup_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
push_rollup_date_filters(
&mut rollup_conditions,
&mut rollup_params,
"date",
&rollup_bounds,
);
let rollup_where = if rollup_conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", rollup_conditions.join(" AND "))
};
let fresh_input_detail = fresh_input_sql("l");
let fresh_input_rollup = fresh_input_sql("");
let sql = format!(
"SELECT app_type,
SUM(req_count) as req_count,
SUM(cost) as cost,
SUM(input_t) as input_t,
SUM(output_t) as output_t,
SUM(cache_create_t) as cache_create_t,
SUM(cache_read_t) as cache_read_t,
SUM(success_count) as success_count
FROM (
SELECT l.app_type,
COUNT(*) as req_count,
COALESCE(SUM(CAST(l.total_cost_usd AS REAL)), 0) as cost,
COALESCE(SUM({fresh_input_detail}), 0) as input_t,
COALESCE(SUM(l.output_tokens), 0) as output_t,
COALESCE(SUM(l.cache_creation_tokens), 0) as cache_create_t,
COALESCE(SUM(l.cache_read_tokens), 0) as cache_read_t,
COALESCE(SUM(CASE WHEN l.status_code >= 200 AND l.status_code < 300 THEN 1 ELSE 0 END), 0) as success_count
FROM proxy_request_logs l {detail_where}
GROUP BY l.app_type
UNION ALL
SELECT app_type,
COALESCE(SUM(request_count), 0),
COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0),
COALESCE(SUM({fresh_input_rollup}), 0),
COALESCE(SUM(output_tokens), 0),
COALESCE(SUM(cache_creation_tokens), 0),
COALESCE(SUM(cache_read_tokens), 0),
COALESCE(SUM(success_count), 0)
FROM usage_daily_rollups {rollup_where}
GROUP BY app_type
)
GROUP BY app_type"
);
let mut combined: Vec<Box<dyn rusqlite::ToSql>> = detail_params;
combined.extend(rollup_params);
let refs: Vec<&dyn rusqlite::ToSql> = combined.iter().map(|p| p.as_ref()).collect();
let mut stmt = conn.prepare(&sql)?;
let rows = stmt.query_map(refs.as_slice(), |row| {
let app_type: String = row.get(0)?;
let total_requests: i64 = row.get(1)?;
let total_cost: f64 = row.get(2)?;
let total_input_tokens: i64 = row.get(3)?;
let total_output_tokens: i64 = row.get(4)?;
let total_cache_creation_tokens: i64 = row.get(5)?;
let total_cache_read_tokens: i64 = row.get(6)?;
let success_count: i64 = row.get(7)?;
let success_rate = if total_requests > 0 {
(success_count as f32 / total_requests as f32) * 100.0
} else {
0.0
};
let (real_total_tokens, cache_hit_rate) = derive_real_total_and_hit_rate(
total_input_tokens as u64,
total_output_tokens as u64,
total_cache_creation_tokens as u64,
total_cache_read_tokens as u64,
);
Ok(UsageSummaryByApp {
app_type,
summary: UsageSummary {
total_requests: total_requests as u64,
total_cost: format!("{total_cost:.6}"),
total_input_tokens: total_input_tokens as u64,
total_output_tokens: total_output_tokens as u64,
total_cache_creation_tokens: total_cache_creation_tokens as u64,
total_cache_read_tokens: total_cache_read_tokens as u64,
success_rate,
real_total_tokens,
cache_hit_rate,
},
})
})?;
let mut summaries = Vec::new();
for row in rows {
let item = row?;
if item.summary.total_requests == 0 && item.summary.real_total_tokens == 0 {
continue;
}
summaries.push(item);
}
summaries.sort_by(|a, b| {
b.summary
.real_total_tokens
.cmp(&a.summary.real_total_tokens)
});
Ok(summaries)
}
/// 获取每日趋势(滑动窗口,<=24h 按小时,>24h 按天,窗口与汇总一致)
pub fn get_daily_trends(
&self,
start_date: Option<i64>,
end_date: Option<i64>,
app_type: Option<&str>,
) -> Result<Vec<DailyStats>, AppError> {
let conn = lock_conn!(self.conn);
let end_ts = end_date.unwrap_or_else(|| Local::now().timestamp());
let mut start_ts = start_date.unwrap_or_else(|| end_ts - 24 * 60 * 60);
if start_ts >= end_ts {
start_ts = end_ts - 24 * 60 * 60;
}
let duration = end_ts - start_ts;
if duration <= 24 * 60 * 60 {
let bucket_seconds: i64 = 60 * 60;
let mut bucket_count: i64 = if duration <= 0 {
1
} else {
(duration + bucket_seconds - 1) / bucket_seconds
};
if bucket_count < 1 {
bucket_count = 1;
}
let app_type_filter = if app_type.is_some() {
"AND l.app_type = ?4"
} else {
""
};
let effective_filter = effective_usage_log_filter("l");
let fresh_input = fresh_input_sql("l");
let sql = format!(
"SELECT
CAST((l.created_at - ?1) / ?3 AS INTEGER) as bucket_idx,
COUNT(*) as request_count,
COALESCE(SUM(CAST(l.total_cost_usd AS REAL)), 0) as total_cost,
COALESCE(SUM({fresh_input} + l.output_tokens), 0) as total_tokens,
COALESCE(SUM({fresh_input}), 0) as total_input_tokens,
COALESCE(SUM(l.output_tokens), 0) as total_output_tokens,
COALESCE(SUM(l.cache_creation_tokens), 0) as total_cache_creation_tokens,
COALESCE(SUM(l.cache_read_tokens), 0) as total_cache_read_tokens
FROM proxy_request_logs l
WHERE l.created_at >= ?1 AND l.created_at <= ?2
AND {effective_filter} {app_type_filter}
GROUP BY bucket_idx
ORDER BY bucket_idx ASC"
);
let mut stmt = conn.prepare(&sql)?;
let row_mapper = |row: &rusqlite::Row| {
Ok((
row.get::<_, i64>(0)?,
DailyStats {
date: String::new(),
request_count: row.get::<_, i64>(1)? as u64,
total_cost: format!("{:.6}", row.get::<_, f64>(2)?),
total_tokens: row.get::<_, i64>(3)? as u64,
total_input_tokens: row.get::<_, i64>(4)? as u64,
total_output_tokens: row.get::<_, i64>(5)? as u64,
total_cache_creation_tokens: row.get::<_, i64>(6)? as u64,
total_cache_read_tokens: row.get::<_, i64>(7)? as u64,
},
))
};
let mut map: HashMap<i64, DailyStats> = HashMap::new();
let rows = if let Some(at) = app_type {
stmt.query_map(params![start_ts, end_ts, bucket_seconds, at], row_mapper)?
} else {
stmt.query_map(params![start_ts, end_ts, bucket_seconds], row_mapper)?
};
for row in rows {
let (mut bucket_idx, stat) = row?;
if bucket_idx < 0 {
continue;
}
if bucket_idx >= bucket_count {
bucket_idx = bucket_count - 1;
}
map.insert(bucket_idx, stat);
}
let mut stats = Vec::with_capacity(bucket_count as usize);
for i in 0..bucket_count {
let bucket_start_ts = start_ts + i * bucket_seconds;
let bucket_start = local_datetime_from_timestamp(bucket_start_ts)?;
let date = bucket_start.to_rfc3339();
if let Some(mut stat) = map.remove(&i) {
stat.date = date;
stats.push(stat);
} else {
stats.push(DailyStats {
date,
request_count: 0,
total_cost: "0.000000".to_string(),
total_tokens: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_cache_creation_tokens: 0,
total_cache_read_tokens: 0,
});
}
}
return Ok(stats);
}
let start_day = local_datetime_from_timestamp(start_ts)?.date_naive();
let end_day = local_datetime_from_timestamp(end_ts)?.date_naive();
let bucket_count = (end_day.signed_duration_since(start_day).num_days() + 1) as usize;
let app_type_filter = if app_type.is_some() {
"AND l.app_type = ?3"
} else {
""
};
let effective_filter = effective_usage_log_filter("l");
let fresh_input = fresh_input_sql("l");
let detail_sql = format!(
"SELECT
date(l.created_at, 'unixepoch', 'localtime') as bucket_date,
COUNT(*) as request_count,
COALESCE(SUM(CAST(l.total_cost_usd AS REAL)), 0) as total_cost,
COALESCE(SUM({fresh_input} + l.output_tokens), 0) as total_tokens,
COALESCE(SUM({fresh_input}), 0) as total_input_tokens,
COALESCE(SUM(l.output_tokens), 0) as total_output_tokens,
COALESCE(SUM(l.cache_creation_tokens), 0) as total_cache_creation_tokens,
COALESCE(SUM(l.cache_read_tokens), 0) as total_cache_read_tokens
FROM proxy_request_logs l
WHERE l.created_at >= ?1 AND l.created_at <= ?2
AND {effective_filter} {app_type_filter}
GROUP BY bucket_date
ORDER BY bucket_date ASC"
);
let mut detail_stmt = conn.prepare(&detail_sql)?;
let detail_row_mapper = |row: &rusqlite::Row| {
Ok((
row.get::<_, String>(0)?,
DailyStats {
date: String::new(),
request_count: row.get::<_, i64>(1)? as u64,
total_cost: format!("{:.6}", row.get::<_, f64>(2)?),
total_tokens: row.get::<_, i64>(3)? as u64,
total_input_tokens: row.get::<_, i64>(4)? as u64,
total_output_tokens: row.get::<_, i64>(5)? as u64,
total_cache_creation_tokens: row.get::<_, i64>(6)? as u64,
total_cache_read_tokens: row.get::<_, i64>(7)? as u64,
},
))
};
let mut map: HashMap<NaiveDate, DailyStats> = HashMap::new();
let detail_rows = if let Some(at) = app_type {
detail_stmt.query_map(params![start_ts, end_ts, at], detail_row_mapper)?
} else {
detail_stmt.query_map(params![start_ts, end_ts], detail_row_mapper)?
};
for row in detail_rows {
let (bucket_date, stat) = row?;
let date = NaiveDate::parse_from_str(&bucket_date, "%Y-%m-%d")
.map_err(|err| AppError::Database(format!("解析趋势日期失败: {err}")))?;
map.insert(date, stat);
}
let rollup_bounds = compute_rollup_date_bounds(Some(start_ts), Some(end_ts))?;
let mut rollup_conditions = Vec::new();
let mut rollup_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
push_rollup_date_filters(
&mut rollup_conditions,
&mut rollup_params,
"date",
&rollup_bounds,
);
if let Some(at) = app_type {
rollup_conditions.push("app_type = ?".to_string());
rollup_params.push(Box::new(at.to_string()));
}
let rollup_where = if rollup_conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", rollup_conditions.join(" AND "))
};
let fresh_input_rollup = fresh_input_sql("");
let rollup_sql = format!(
"SELECT
date,
COALESCE(SUM(request_count), 0),
COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0),
COALESCE(SUM({fresh_input_rollup} + output_tokens), 0),
COALESCE(SUM({fresh_input_rollup}), 0),
COALESCE(SUM(output_tokens), 0),
COALESCE(SUM(cache_creation_tokens), 0),
COALESCE(SUM(cache_read_tokens), 0)
FROM usage_daily_rollups
{rollup_where}
GROUP BY date
ORDER BY date ASC"
);
let mut rollup_stmt = conn.prepare(&rollup_sql)?;
let rollup_row_mapper = |row: &rusqlite::Row| {
Ok((
row.get::<_, String>(0)?,
(
row.get::<_, i64>(1)? as u64,
row.get::<_, f64>(2)?,
row.get::<_, i64>(3)? as u64,
row.get::<_, i64>(4)? as u64,
row.get::<_, i64>(5)? as u64,
row.get::<_, i64>(6)? as u64,
row.get::<_, i64>(7)? as u64,
),
))
};
let rollup_param_refs: Vec<&dyn rusqlite::ToSql> =
rollup_params.iter().map(|param| param.as_ref()).collect();
let rollup_rows = rollup_stmt.query_map(rollup_param_refs.as_slice(), rollup_row_mapper)?;
for row in rollup_rows {
let (bucket_date, (req, cost, tok, inp, out, cc, cr)) = row?;
let date = NaiveDate::parse_from_str(&bucket_date, "%Y-%m-%d")
.map_err(|err| AppError::Database(format!("解析 rollup 趋势日期失败: {err}")))?;
let entry = map.entry(date).or_insert_with(|| DailyStats {
date: String::new(),
request_count: 0,
total_cost: "0.000000".to_string(),
total_tokens: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_cache_creation_tokens: 0,
total_cache_read_tokens: 0,
});
entry.request_count += req;
let existing_cost: f64 = entry.total_cost.parse().unwrap_or(0.0);
entry.total_cost = format!("{:.6}", existing_cost + cost);
entry.total_tokens += tok;
entry.total_input_tokens += inp;
entry.total_output_tokens += out;
entry.total_cache_creation_tokens += cc;
entry.total_cache_read_tokens += cr;
}
let mut stats = Vec::with_capacity(bucket_count);
let mut current_day = start_day;
for _ in 0..bucket_count {
let date = local_day_start_rfc3339(current_day);
if let Some(mut stat) = map.remove(&current_day) {
stat.date = date;
stats.push(stat);
} else {
stats.push(DailyStats {
date,
request_count: 0,
total_cost: "0.000000".to_string(),
total_tokens: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_cache_creation_tokens: 0,
total_cache_read_tokens: 0,
});
}
current_day = current_day.succ_opt().unwrap_or(current_day);
}
Ok(stats)
}
/// 获取 Provider 统计
pub fn get_provider_stats(
&self,
start_date: Option<i64>,
end_date: Option<i64>,
app_type: Option<&str>,
) -> Result<Vec<ProviderStats>, AppError> {
let conn = lock_conn!(self.conn);
let mut detail_conditions = vec![effective_usage_log_filter("l")];
let mut detail_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(start) = start_date {
detail_conditions.push("l.created_at >= ?".to_string());
detail_params.push(Box::new(start));
}
if let Some(end) = end_date {
detail_conditions.push("l.created_at <= ?".to_string());
detail_params.push(Box::new(end));
}
if let Some(at) = app_type {
detail_conditions.push("l.app_type = ?".to_string());
detail_params.push(Box::new(at.to_string()));
}
let detail_where = if detail_conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", detail_conditions.join(" AND "))
};
let mut rollup_conditions = Vec::new();
let mut rollup_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
let rollup_bounds = compute_rollup_date_bounds(start_date, end_date)?;
push_rollup_date_filters(
&mut rollup_conditions,
&mut rollup_params,
"r.date",
&rollup_bounds,
);
if let Some(at) = app_type {
rollup_conditions.push("r.app_type = ?".to_string());
rollup_params.push(Box::new(at.to_string()));
}
let rollup_where = if rollup_conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", rollup_conditions.join(" AND "))
};
// UNION detail logs + rollup data, then aggregate
let detail_pname = provider_name_coalesce("l", "p");
let rollup_pname = provider_name_coalesce("r", "p2");
let fresh_input_detail = fresh_input_sql("l");
let fresh_input_rollup = fresh_input_sql("r");
let sql = format!(
"SELECT
provider_id, app_type, provider_name,
SUM(request_count) as request_count,
SUM(total_tokens) as total_tokens,
SUM(total_cost) as total_cost,
SUM(success_count) as success_count,
CASE WHEN SUM(request_count) > 0
THEN SUM(latency_sum) / SUM(request_count)
ELSE 0 END as avg_latency
FROM (
SELECT l.provider_id, l.app_type,
{detail_pname} as provider_name,
COUNT(*) as request_count,
COALESCE(SUM({fresh_input_detail} + l.output_tokens), 0) as total_tokens,
COALESCE(SUM(CAST(l.total_cost_usd AS REAL)), 0) as total_cost,
COALESCE(SUM(CASE WHEN l.status_code >= 200 AND l.status_code < 300 THEN 1 ELSE 0 END), 0) as success_count,
COALESCE(SUM(l.latency_ms), 0) as latency_sum
FROM proxy_request_logs l
LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type
{detail_where}
GROUP BY l.provider_id, l.app_type
UNION ALL
SELECT r.provider_id, r.app_type,
{rollup_pname} as provider_name,
COALESCE(SUM(r.request_count), 0),
COALESCE(SUM({fresh_input_rollup} + r.output_tokens), 0),
COALESCE(SUM(CAST(r.total_cost_usd AS REAL)), 0),
COALESCE(SUM(r.success_count), 0),
COALESCE(SUM(r.avg_latency_ms * r.request_count), 0)
FROM usage_daily_rollups r
LEFT JOIN providers p2 ON r.provider_id = p2.id AND r.app_type = p2.app_type
{rollup_where}
GROUP BY r.provider_id, r.app_type
)
GROUP BY provider_id, app_type
ORDER BY total_cost DESC"
);
let mut stmt = conn.prepare(&sql)?;
let mut params: Vec<Box<dyn rusqlite::ToSql>> = detail_params;
params.extend(rollup_params);
let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
let row_mapper = |row: &rusqlite::Row| {
let request_count: i64 = row.get(3)?;
let success_count: i64 = row.get(6)?;
let success_rate = if request_count > 0 {
(success_count as f32 / request_count as f32) * 100.0
} else {
0.0
};
Ok(ProviderStats {
provider_id: row.get(0)?,
provider_name: row.get(2)?,
request_count: request_count as u64,
total_tokens: row.get::<_, i64>(4)? as u64,
total_cost: format!("{:.6}", row.get::<_, f64>(5)?),
success_rate,
avg_latency_ms: row.get::<_, f64>(7)? as u64,
})
};
let rows = stmt.query_map(param_refs.as_slice(), row_mapper)?;
let mut stats = Vec::new();
for row in rows {
stats.push(row?);
}
Ok(stats)
}
/// 获取模型统计
pub fn get_model_stats(
&self,
start_date: Option<i64>,
end_date: Option<i64>,
app_type: Option<&str>,
) -> Result<Vec<ModelStats>, AppError> {
let conn = lock_conn!(self.conn);
let mut detail_conditions = vec![effective_usage_log_filter("l")];
let mut detail_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(start) = start_date {
detail_conditions.push("l.created_at >= ?".to_string());
detail_params.push(Box::new(start));
}
if let Some(end) = end_date {
detail_conditions.push("l.created_at <= ?".to_string());
detail_params.push(Box::new(end));
}
if let Some(at) = app_type {
detail_conditions.push("l.app_type = ?".to_string());
detail_params.push(Box::new(at.to_string()));
}
let detail_where = if detail_conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", detail_conditions.join(" AND "))
};
let mut rollup_conditions = Vec::new();
let mut rollup_params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
let rollup_bounds = compute_rollup_date_bounds(start_date, end_date)?;
push_rollup_date_filters(
&mut rollup_conditions,
&mut rollup_params,
"r.date",
&rollup_bounds,
);
if let Some(at) = app_type {
rollup_conditions.push("r.app_type = ?".to_string());
rollup_params.push(Box::new(at.to_string()));
}
let rollup_where = if rollup_conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", rollup_conditions.join(" AND "))
};
// UNION detail logs + rollup data
let fresh_input_detail = fresh_input_sql("l");
let fresh_input_rollup = fresh_input_sql("r");
let sql = format!(
"SELECT
model,
SUM(request_count) as request_count,
SUM(total_tokens) as total_tokens,
SUM(total_cost) as total_cost
FROM (
SELECT l.model,
COUNT(*) as request_count,
COALESCE(SUM({fresh_input_detail} + l.output_tokens), 0) as total_tokens,
COALESCE(SUM(CAST(l.total_cost_usd AS REAL)), 0) as total_cost
FROM proxy_request_logs l
{detail_where}
GROUP BY l.model
UNION ALL
SELECT r.model,
COALESCE(SUM(r.request_count), 0),
COALESCE(SUM({fresh_input_rollup} + r.output_tokens), 0),
COALESCE(SUM(CAST(r.total_cost_usd AS REAL)), 0)
FROM usage_daily_rollups r
{rollup_where}
GROUP BY r.model
)
GROUP BY model
ORDER BY total_cost DESC"
);
let mut stmt = conn.prepare(&sql)?;
let mut params: Vec<Box<dyn rusqlite::ToSql>> = detail_params;
params.extend(rollup_params);
let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
let row_mapper = |row: &rusqlite::Row| {
let request_count: i64 = row.get(1)?;
let total_cost: f64 = row.get(3)?;
let avg_cost = if request_count > 0 {
total_cost / request_count as f64
} else {
0.0
};
Ok(ModelStats {
model: row.get(0)?,
request_count: request_count as u64,
total_tokens: row.get::<_, i64>(2)? as u64,
total_cost: format!("{total_cost:.6}"),
avg_cost_per_request: format!("{avg_cost:.6}"),
})
};
let rows = stmt.query_map(param_refs.as_slice(), row_mapper)?;
let mut stats = Vec::new();
for row in rows {
stats.push(row?);
}
Ok(stats)
}
/// 获取请求日志列表(分页)
pub fn get_request_logs(
&self,
filters: &LogFilters,
page: u32,
page_size: u32,
) -> Result<PaginatedLogs, AppError> {
let conn = lock_conn!(self.conn);
let mut conditions = vec![effective_usage_log_filter("l")];
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(ref app_type) = filters.app_type {
conditions.push("l.app_type = ?".to_string());
params.push(Box::new(app_type.clone()));
}
if let Some(ref provider_name) = filters.provider_name {
conditions.push("p.name LIKE ?".to_string());
params.push(Box::new(format!("%{provider_name}%")));
}
if let Some(ref model) = filters.model {
conditions.push("l.model LIKE ?".to_string());
params.push(Box::new(format!("%{model}%")));
}
if let Some(status) = filters.status_code {
conditions.push("l.status_code = ?".to_string());
params.push(Box::new(status as i64));
}
if let Some(start) = filters.start_date {
conditions.push("l.created_at >= ?".to_string());
params.push(Box::new(start));
}
if let Some(end) = filters.end_date {
conditions.push("l.created_at <= ?".to_string());
params.push(Box::new(end));
}
let where_clause = if conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", conditions.join(" AND "))
};
// 获取总数
let count_sql = format!(
"SELECT COUNT(*) FROM proxy_request_logs l
LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type
{where_clause}"
);
let count_params: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
let total: u32 = conn.query_row(&count_sql, count_params.as_slice(), |row| {
row.get::<_, i64>(0).map(|v| v as u32)
})?;
// 获取数据
let offset = page * page_size;
params.push(Box::new(page_size as i64));
params.push(Box::new(offset as i64));
let logs_pname = provider_name_coalesce("l", "p");
let sql = format!(
"SELECT l.request_id, l.provider_id, {logs_pname} as provider_name, l.app_type, l.model,
l.request_model, l.cost_multiplier,
l.input_tokens, l.output_tokens, l.cache_read_tokens, l.cache_creation_tokens,
l.input_cost_usd, l.output_cost_usd, l.cache_read_cost_usd, l.cache_creation_cost_usd, l.total_cost_usd,
l.is_streaming, l.latency_ms, l.first_token_ms, l.duration_ms,
l.status_code, l.error_message, l.created_at, l.data_source
FROM proxy_request_logs l
LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type
{where_clause}
ORDER BY l.created_at DESC
LIMIT ? OFFSET ?"
);
let mut stmt = conn.prepare(&sql)?;
let params_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
let rows = stmt.query_map(params_refs.as_slice(), row_to_request_log_detail)?;
let mut logs = Vec::new();
let mut pricing_cache = HashMap::new();
for row in rows {
let mut log = row?;
Self::maybe_backfill_log_costs(&conn, &mut log, &mut pricing_cache)?;
logs.push(log);
}
Ok(PaginatedLogs {
data: logs,
total,
page,
page_size,
})
}
/// 获取单个请求详情
pub fn get_request_detail(
&self,
request_id: &str,
) -> Result<Option<RequestLogDetail>, AppError> {
let conn = lock_conn!(self.conn);
let detail_pname = provider_name_coalesce("l", "p");
let detail_sql = format!(
"SELECT l.request_id, l.provider_id, {detail_pname} as provider_name, l.app_type, l.model,
l.request_model, l.cost_multiplier,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd, total_cost_usd,
is_streaming, latency_ms, first_token_ms, duration_ms,
status_code, error_message, created_at, l.data_source
FROM proxy_request_logs l
LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type
WHERE l.request_id = ?"
);
let result = conn.query_row(&detail_sql, [request_id], row_to_request_log_detail);
match result {
Ok(mut detail) => {
let mut pricing_cache = HashMap::new();
Self::maybe_backfill_log_costs(&conn, &mut detail, &mut pricing_cache)?;
Ok(Some(detail))
}
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(AppError::Database(e.to_string())),
}
}
/// 检查 Provider 使用限额
pub fn check_provider_limits(
&self,
provider_id: &str,
app_type: &str,
) -> Result<ProviderLimitStatus, AppError> {
let conn = lock_conn!(self.conn);
// 获取 provider 的限额设置
let (limit_daily, limit_monthly) = conn
.query_row(
"SELECT meta FROM providers WHERE id = ? AND app_type = ?",
params![provider_id, app_type],
|row| {
let meta_str: String = row.get(0)?;
Ok(meta_str)
},
)
.ok()
.and_then(|meta_str| serde_json::from_str::<serde_json::Value>(&meta_str).ok())
.map(|meta| {
let daily = meta
.get("limitDailyUsd")
.and_then(|v| v.as_str())
.and_then(|s| s.parse::<f64>().ok());
let monthly = meta
.get("limitMonthlyUsd")
.and_then(|v| v.as_str())
.and_then(|s| s.parse::<f64>().ok());
(daily, monthly)
})
.unwrap_or((None, None));
// 计算今日使用量 (detail logs + rollup)
let daily_usage: f64 = conn
.query_row(
"SELECT COALESCE(SUM(cost), 0) FROM (
SELECT CAST(total_cost_usd AS REAL) as cost
FROM proxy_request_logs
WHERE provider_id = ? AND app_type = ?
AND date(datetime(created_at, 'unixepoch', 'localtime')) = date('now', 'localtime')
UNION ALL
SELECT CAST(total_cost_usd AS REAL)
FROM usage_daily_rollups
WHERE provider_id = ? AND app_type = ?
AND date = date('now', 'localtime')
)",
params![provider_id, app_type, provider_id, app_type],
|row| row.get(0),
)
.unwrap_or(0.0);
// 计算本月使用量 (detail logs + rollup)
let monthly_usage: f64 = conn
.query_row(
"SELECT COALESCE(SUM(cost), 0) FROM (
SELECT CAST(total_cost_usd AS REAL) as cost
FROM proxy_request_logs
WHERE provider_id = ? AND app_type = ?
AND strftime('%Y-%m', datetime(created_at, 'unixepoch', 'localtime')) = strftime('%Y-%m', 'now', 'localtime')
UNION ALL
SELECT CAST(total_cost_usd AS REAL)
FROM usage_daily_rollups
WHERE provider_id = ? AND app_type = ?
AND strftime('%Y-%m', date) = strftime('%Y-%m', 'now', 'localtime')
)",
params![provider_id, app_type, provider_id, app_type],
|row| row.get(0),
)
.unwrap_or(0.0);
let daily_exceeded = limit_daily
.map(|limit| daily_usage >= limit)
.unwrap_or(false);
let monthly_exceeded = limit_monthly
.map(|limit| monthly_usage >= limit)
.unwrap_or(false);
Ok(ProviderLimitStatus {
provider_id: provider_id.to_string(),
daily_usage: format!("{daily_usage:.6}"),
daily_limit: limit_daily.map(|l| format!("{l:.2}")),
daily_exceeded,
monthly_usage: format!("{monthly_usage:.6}"),
monthly_limit: limit_monthly.map(|l| format!("{l:.2}")),
monthly_exceeded,
})
}
}
/// Provider 限额状态
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProviderLimitStatus {
pub provider_id: String,
pub daily_usage: String,
pub daily_limit: Option<String>,
pub daily_exceeded: bool,
pub monthly_usage: String,
pub monthly_limit: Option<String>,
pub monthly_exceeded: bool,
}
#[derive(Clone)]
struct PricingInfo {
input: rust_decimal::Decimal,
output: rust_decimal::Decimal,
cache_read: rust_decimal::Decimal,
cache_creation: rust_decimal::Decimal,
}
impl Database {
/// Recalculate stored zero-cost usage rows once pricing becomes available.
pub(crate) fn backfill_missing_usage_costs(&self) -> Result<u64, AppError> {
let conn = lock_conn!(self.conn);
Self::backfill_missing_usage_costs_on_conn(&conn)
}
fn backfill_missing_usage_costs_on_conn(conn: &Connection) -> Result<u64, AppError> {
let mut logs = {
let mut stmt = conn.prepare(
"SELECT request_id, provider_id, NULL AS provider_name, app_type, model, request_model,
cost_multiplier,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_cost_usd, output_cost_usd, cache_read_cost_usd,
cache_creation_cost_usd, total_cost_usd, is_streaming, latency_ms,
first_token_ms, duration_ms, status_code, error_message, created_at,
data_source
FROM proxy_request_logs
WHERE CAST(total_cost_usd AS REAL) <= 0
AND (input_tokens > 0 OR output_tokens > 0
OR cache_read_tokens > 0 OR cache_creation_tokens > 0)",
)?;
let rows = stmt.query_map([], row_to_request_log_detail)?;
rows.collect::<Result<Vec<_>, _>>()?
};
if logs.is_empty() {
return Ok(0);
}
let tx = conn
.unchecked_transaction()
.map_err(|e| AppError::Database(format!("启动用量成本回填事务失败: {e}")))?;
let mut updated = 0u64;
let mut pricing_cache = HashMap::new();
for log in &mut logs {
if Self::maybe_backfill_log_costs(&tx, log, &mut pricing_cache)? {
updated += 1;
}
}
tx.commit()
.map_err(|e| AppError::Database(format!("提交用量成本回填事务失败: {e}")))?;
if updated > 0 {
log::info!("已回填 {updated} 条缺失的用量成本");
}
Ok(updated)
}
/// 尝试为单条 log 回填成本字段。返回是否实际写入(true=已 UPDATEfalse=跳过)。
fn maybe_backfill_log_costs(
conn: &Connection,
log: &mut RequestLogDetail,
pricing_cache: &mut HashMap<String, PricingInfo>,
) -> Result<bool, AppError> {
let existing_cost = rust_decimal::Decimal::from_str(&log.total_cost_usd)
.unwrap_or(rust_decimal::Decimal::ZERO);
let has_cost = existing_cost > rust_decimal::Decimal::ZERO;
let has_usage = log.input_tokens > 0
|| log.output_tokens > 0
|| log.cache_read_tokens > 0
|| log.cache_creation_tokens > 0;
if has_cost || !has_usage {
return Ok(false);
}
let pricing = match Self::get_log_model_pricing_cached(conn, pricing_cache, log)? {
Some(info) => info,
None => return Ok(false),
};
let multiplier =
rust_decimal::Decimal::from_str(&log.cost_multiplier).unwrap_or_else(|e| {
log::warn!(
"历史用量倍率解析失败 request_id={}: {} - {e}",
log.request_id,
log.cost_multiplier
);
rust_decimal::Decimal::ONE
});
let million = rust_decimal::Decimal::from(1_000_000u64);
// 与 CostCalculator::calculate_for_app 保持一致的计算逻辑:
// 1. Codex/Gemini 的 input_tokens 包含 cache_read_tokens,需要扣除后按输入价计费
// 2. Claude/Anthropic 的 input_tokens 已经是 fresh input,不能再次扣减
// 3. 各项成本是基础成本(不含倍率),倍率只作用于最终总价
let input_includes_cache_read = matches!(log.app_type.as_str(), "codex" | "gemini");
let billable_input_tokens = if input_includes_cache_read {
(log.input_tokens as u64).saturating_sub(log.cache_read_tokens as u64)
} else {
log.input_tokens as u64
};
let input_cost =
rust_decimal::Decimal::from(billable_input_tokens) * pricing.input / million;
let output_cost =
rust_decimal::Decimal::from(log.output_tokens as u64) * pricing.output / million;
let cache_read_cost = rust_decimal::Decimal::from(log.cache_read_tokens as u64)
* pricing.cache_read
/ million;
let cache_creation_cost = rust_decimal::Decimal::from(log.cache_creation_tokens as u64)
* pricing.cache_creation
/ million;
// 总成本 = 基础成本之和 × 倍率
let base_total = input_cost + output_cost + cache_read_cost + cache_creation_cost;
let total_cost = base_total * multiplier;
log.input_cost_usd = format!("{input_cost:.6}");
log.output_cost_usd = format!("{output_cost:.6}");
log.cache_read_cost_usd = format!("{cache_read_cost:.6}");
log.cache_creation_cost_usd = format!("{cache_creation_cost:.6}");
log.total_cost_usd = format!("{total_cost:.6}");
conn.execute(
"UPDATE proxy_request_logs
SET input_cost_usd = ?1,
output_cost_usd = ?2,
cache_read_cost_usd = ?3,
cache_creation_cost_usd = ?4,
total_cost_usd = ?5
WHERE request_id = ?6",
params![
log.input_cost_usd,
log.output_cost_usd,
log.cache_read_cost_usd,
log.cache_creation_cost_usd,
log.total_cost_usd,
log.request_id
],
)
.map_err(|e| AppError::Database(format!("更新请求成本失败: {e}")))?;
Ok(true)
}
fn get_model_pricing_cached(
conn: &Connection,
cache: &mut HashMap<String, PricingInfo>,
model: &str,
) -> Result<Option<PricingInfo>, AppError> {
if let Some(info) = cache.get(model) {
return Ok(Some(info.clone()));
}
let row = find_model_pricing_row(conn, model)?;
let Some((input, output, cache_read, cache_creation)) = row else {
return Ok(None);
};
let pricing = PricingInfo {
input: rust_decimal::Decimal::from_str(&input)
.map_err(|e| AppError::Database(format!("解析输入价格失败: {e}")))?,
output: rust_decimal::Decimal::from_str(&output)
.map_err(|e| AppError::Database(format!("解析输出价格失败: {e}")))?,
cache_read: rust_decimal::Decimal::from_str(&cache_read)
.map_err(|e| AppError::Database(format!("解析缓存读取价格失败: {e}")))?,
cache_creation: rust_decimal::Decimal::from_str(&cache_creation)
.map_err(|e| AppError::Database(format!("解析缓存写入价格失败: {e}")))?,
};
cache.insert(model.to_string(), pricing.clone());
Ok(Some(pricing))
}
fn get_log_model_pricing_cached(
conn: &Connection,
cache: &mut HashMap<String, PricingInfo>,
log: &RequestLogDetail,
) -> Result<Option<PricingInfo>, AppError> {
if let Some(pricing) = Self::get_model_pricing_cached(conn, cache, &log.model)? {
return Ok(Some(pricing));
}
let Some(request_model) = log.request_model.as_deref() else {
return Ok(None);
};
if request_model == log.model {
return Ok(None);
}
Self::get_model_pricing_cached(conn, cache, request_model)
}
}
pub(crate) fn find_model_pricing_row(
conn: &Connection,
model_id: &str,
) -> Result<Option<(String, String, String, String)>, AppError> {
let candidates = model_pricing_candidates(model_id);
if candidates.is_empty() {
return Ok(None);
}
for candidate in &candidates {
if let Some(row) = query_model_pricing_exact(conn, candidate)? {
return Ok(Some(row));
}
}
for candidate in &candidates {
if should_try_pricing_prefix_match(candidate) {
if let Some(row) = query_model_pricing_prefix(conn, candidate)? {
return Ok(Some(row));
}
}
}
Ok(None)
}
pub(crate) fn is_placeholder_pricing_model(model_id: &str) -> bool {
let normalized = model_id.trim().to_ascii_lowercase();
normalized.is_empty() || matches!(normalized.as_str(), "unknown" | "null" | "none")
}
fn query_model_pricing_exact(
conn: &Connection,
model_id: &str,
) -> Result<Option<(String, String, String, String)>, AppError> {
conn.query_row(
"SELECT input_cost_per_million, output_cost_per_million,
cache_read_cost_per_million, cache_creation_cost_per_million
FROM model_pricing
WHERE model_id = ?1",
[model_id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
},
)
.optional()
.map_err(|e| AppError::Database(format!("查询模型定价失败: {e}")))
}
fn query_model_pricing_prefix(
conn: &Connection,
model_id: &str,
) -> Result<Option<(String, String, String, String)>, AppError> {
let pattern = format!("{model_id}-%");
conn.query_row(
"SELECT input_cost_per_million, output_cost_per_million,
cache_read_cost_per_million, cache_creation_cost_per_million
FROM model_pricing
WHERE model_id LIKE ?1
ORDER BY LENGTH(model_id) ASC
LIMIT 1",
[pattern],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
},
)
.optional()
.map_err(|e| AppError::Database(format!("查询模型前缀定价失败: {e}")))
}
fn model_pricing_candidates(model_id: &str) -> Vec<String> {
let cleaned = clean_model_id_for_pricing(model_id);
if is_placeholder_pricing_model(&cleaned) {
return Vec::new();
}
let mut candidates = Vec::new();
let mut queue = vec![cleaned];
while let Some(candidate) = queue.pop() {
if !push_unique_candidate(&mut candidates, candidate.clone()) {
continue;
}
if let Some(stripped) = strip_known_model_namespace(&candidate) {
queue.push(stripped);
}
if let Some(stripped) = strip_claude_desktop_non_anthropic_prefix(&candidate) {
queue.push(stripped);
}
if let Some(stripped) = strip_bedrock_model_version_suffix(&candidate) {
queue.push(stripped);
}
if let Some(stripped) = strip_reasoning_effort_suffix(&candidate) {
queue.push(stripped);
}
if candidate.starts_with("claude-") && candidate.contains('.') {
queue.push(candidate.replace('.', "-"));
}
}
candidates
}
fn clean_model_id_for_pricing(model_id: &str) -> String {
let normalized = model_id
.rsplit_once('/')
.map_or(model_id, |(_, r)| r)
.split(':')
.next()
.unwrap_or(model_id)
.trim()
.replace('@', "-")
.to_ascii_lowercase();
normalized
.trim_end_matches(crate::claude_desktop_config::ONE_M_CONTEXT_MARKER)
.trim()
.to_string()
}
fn push_unique_candidate(candidates: &mut Vec<String>, candidate: String) -> bool {
if candidate.is_empty() || candidates.iter().any(|existing| existing == &candidate) {
return false;
}
candidates.push(candidate);
true
}
fn strip_known_model_namespace(model_id: &str) -> Option<String> {
if let Some(pos) = model_id.rfind("claude-") {
if pos > 0 {
return Some(model_id[pos..].to_string());
}
}
for marker in [
"openai.",
"anthropic.",
"google.",
"moonshot.",
"moonshotai.",
"bedrock.",
"global.",
] {
if let Some(stripped) = model_id.strip_prefix(marker) {
return Some(stripped.to_string());
}
}
None
}
fn strip_claude_desktop_non_anthropic_prefix(model_id: &str) -> Option<String> {
const NON_ANTHROPIC_MARKERS: &[&str] = &[
"abab",
"ark-code",
"arctic",
"astron",
"codex",
"command-r",
"deepseek",
"doubao",
"ernie",
"gemini",
"gemma",
"glm",
"gpt",
"grok",
"hermes",
"hy3",
"hunyuan",
"jamba",
"kimi",
"lfm",
"llama",
"longcat",
"mercury",
"mimo",
"minimax",
"mistral",
"mixtral",
"moonshot",
"nemotron",
"nova-",
"openai",
"qianfan",
"qwen",
"seed-",
"solar",
"stepfun",
];
let rest = model_id.strip_prefix("claude-")?;
NON_ANTHROPIC_MARKERS
.iter()
.any(|marker| rest.starts_with(marker))
.then(|| rest.to_string())
}
fn strip_bedrock_model_version_suffix(model_id: &str) -> Option<String> {
let (base, suffix) = model_id.rsplit_once("-v")?;
(!base.is_empty() && !suffix.is_empty() && suffix.chars().all(|c| c.is_ascii_digit()))
.then(|| base.to_string())
}
fn strip_reasoning_effort_suffix(model_id: &str) -> Option<String> {
for suffix in ["-minimal", "-low", "-medium", "-high", "-xhigh"] {
if let Some(stripped) = model_id.strip_suffix(suffix) {
if !stripped.is_empty() {
return Some(stripped.to_string());
}
}
}
None
}
fn should_try_pricing_prefix_match(model_id: &str) -> bool {
model_id.starts_with("claude-") && model_id.matches('-').count() >= 3
}
#[cfg(test)]
mod tests {
use super::*;
fn local_ts(year: i32, month: u32, day: u32, hour: u32, minute: u32, second: u32) -> i64 {
match Local.with_ymd_and_hms(year, month, day, hour, minute, second) {
chrono::LocalResult::Single(dt) => dt.timestamp(),
chrono::LocalResult::Ambiguous(earliest, _) => earliest.timestamp(),
chrono::LocalResult::None => panic!("valid local datetime"),
}
}
#[allow(clippy::too_many_arguments)]
fn insert_usage_log(
conn: &Connection,
request_id: &str,
app_type: &str,
provider_id: &str,
model: &str,
data_source: &str,
created_at: i64,
input_tokens: i64,
output_tokens: i64,
cache_read_tokens: i64,
cache_creation_tokens: i64,
status_code: i64,
total_cost_usd: &str,
) -> Result<(), AppError> {
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model, request_model,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd,
total_cost_usd, latency_ms, status_code, created_at, data_source
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, '0', '0', '0', '0', ?, 100, ?, ?, ?)",
params![
request_id,
provider_id,
app_type,
model,
model,
input_tokens,
output_tokens,
cache_read_tokens,
cache_creation_tokens,
total_cost_usd,
status_code,
created_at,
data_source
],
)?;
Ok(())
}
fn create_legacy_nullable_logs_table(conn: &Connection) -> Result<(), AppError> {
conn.execute(
"CREATE TABLE proxy_request_logs (
request_id TEXT PRIMARY KEY,
app_type TEXT NOT NULL,
model TEXT NOT NULL,
input_tokens INTEGER NOT NULL,
output_tokens INTEGER NOT NULL,
cache_read_tokens INTEGER NOT NULL,
cache_creation_tokens INTEGER NOT NULL,
status_code INTEGER NOT NULL,
created_at INTEGER NOT NULL,
data_source TEXT
)",
[],
)?;
Ok(())
}
#[test]
fn test_effective_filter_keeps_legacy_null_data_source_proxy_rows() -> Result<(), AppError> {
let conn = Connection::open_in_memory()?;
create_legacy_nullable_logs_table(&conn)?;
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, app_type, model, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, status_code, created_at, data_source
) VALUES ('legacy-proxy', 'codex', 'gpt-5.5', 10, 2, 1, 0, 200, 1000, NULL)",
[],
)?;
let filter = effective_usage_log_filter("l");
let sql = format!("SELECT COUNT(*) FROM proxy_request_logs l WHERE {filter}");
let count: i64 = conn.query_row(&sql, [], |row| row.get(0))?;
assert_eq!(count, 1);
Ok(())
}
#[test]
fn test_matching_proxy_log_treats_legacy_null_data_source_as_proxy() -> Result<(), AppError> {
let conn = Connection::open_in_memory()?;
create_legacy_nullable_logs_table(&conn)?;
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, app_type, model, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, status_code, created_at, data_source
) VALUES ('legacy-proxy', 'codex', 'gpt-5.5', 10, 2, 1, 0, 200, 1000, NULL)",
[],
)?;
let key = DedupKey {
app_type: "codex",
model: "gpt-5.5",
input_tokens: 10,
output_tokens: 2,
cache_read_tokens: 1,
cache_creation_tokens: 0,
created_at: 1000,
};
assert!(has_matching_proxy_usage_log(&conn, &key)?);
Ok(())
}
#[test]
fn test_backfill_missing_usage_costs_uses_new_gpt_5_5_pricing() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
insert_usage_log(
&conn,
"codex-gpt-5-5-zero-cost",
"codex",
"_codex_session",
"gpt-5.5",
"codex_session",
1000,
1_000_000,
1_000_000,
0,
0,
200,
"0",
)?;
}
assert_eq!(db.backfill_missing_usage_costs()?, 1);
let conn = lock_conn!(db.conn);
let (input_cost, output_cost, total_cost): (String, String, String) = conn.query_row(
"SELECT input_cost_usd, output_cost_usd, total_cost_usd
FROM proxy_request_logs WHERE request_id = 'codex-gpt-5-5-zero-cost'",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
assert_eq!(input_cost, "5.000000");
assert_eq!(output_cost, "30.000000");
assert_eq!(total_cost, "35.000000");
Ok(())
}
#[test]
fn test_backfill_missing_usage_costs_uses_stored_multiplier() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
insert_usage_log(
&conn,
"codex-gpt-5-5-multiplier",
"codex",
"_codex_session",
"gpt-5.5",
"codex_session",
1000,
1_000_000,
0,
0,
0,
200,
"0",
)?;
conn.execute(
"UPDATE proxy_request_logs
SET cost_multiplier = '1.5'
WHERE request_id = 'codex-gpt-5-5-multiplier'",
[],
)?;
}
assert_eq!(db.backfill_missing_usage_costs()?, 1);
let conn = lock_conn!(db.conn);
let (input_cost, total_cost): (String, String) = conn.query_row(
"SELECT input_cost_usd, total_cost_usd
FROM proxy_request_logs WHERE request_id = 'codex-gpt-5-5-multiplier'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(input_cost, "5.000000");
assert_eq!(total_cost, "7.500000");
Ok(())
}
#[test]
fn test_backfill_missing_usage_costs_falls_back_to_request_model() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model, request_model,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd,
total_cost_usd, latency_ms, status_code, created_at, data_source
) VALUES (
'codex-request-model-fallback', '_codex_session', 'codex', 'unknown', 'gpt-5.5',
1000000, 0, 0, 0,
'0', '0', '0', '0',
'0', 100, 200, 1000, 'codex_session'
)",
[],
)?;
}
assert_eq!(db.backfill_missing_usage_costs()?, 1);
let conn = lock_conn!(db.conn);
let total_cost: String = conn.query_row(
"SELECT total_cost_usd
FROM proxy_request_logs WHERE request_id = 'codex-request-model-fallback'",
[],
|row| row.get(0),
)?;
assert_eq!(total_cost, "5.000000");
Ok(())
}
#[test]
fn test_backfill_missing_usage_costs_keeps_claude_fresh_input() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
insert_usage_log(
&conn,
"claude-cache-fresh-input",
"claude",
"_session",
"claude-haiku-4-5",
"session_log",
1000,
100,
0,
200,
0,
200,
"0",
)?;
}
assert_eq!(db.backfill_missing_usage_costs()?, 1);
let conn = lock_conn!(db.conn);
let (input_cost, cache_read_cost, total_cost): (String, String, String) = conn.query_row(
"SELECT input_cost_usd, cache_read_cost_usd, total_cost_usd
FROM proxy_request_logs WHERE request_id = 'claude-cache-fresh-input'",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
assert_eq!(input_cost, "0.000100");
assert_eq!(cache_read_cost, "0.000020");
assert_eq!(total_cost, "0.000120");
Ok(())
}
#[test]
fn test_get_usage_summary() -> Result<(), AppError> {
let db = Database::memory()?;
// 插入测试数据
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params!["req1", "p1", "claude", "claude-3", 100, 50, "0.01", 100, 200, 1000],
)?;
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params!["req2", "p1", "claude", "claude-3", 200, 100, "0.02", 150, 200, 2000],
)?;
}
let summary = db.get_usage_summary(None, None, None)?;
assert_eq!(summary.total_requests, 2);
assert_eq!(summary.success_rate, 100.0);
Ok(())
}
#[test]
fn test_get_usage_summary_excludes_partial_rollup_boundary_days() -> Result<(), AppError> {
let db = Database::memory()?;
let start = local_ts(2024, 1, 1, 12, 0, 0);
let end = local_ts(2024, 1, 3, 12, 0, 0);
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-01-01",
"claude",
"p1",
"claude-3",
10,
10,
1000,
500,
0,
0,
"1.00",
100
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-01-02",
"claude",
"p1",
"claude-3",
20,
19,
2000,
1000,
0,
0,
"2.00",
120
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-01-03",
"claude",
"p1",
"claude-3",
30,
29,
3000,
1500,
0,
0,
"3.00",
140
],
)?;
}
let summary = db.get_usage_summary(Some(start), Some(end), Some("claude"))?;
assert_eq!(summary.total_requests, 20);
assert_eq!(summary.total_input_tokens, 2000);
assert_eq!(summary.total_output_tokens, 1000);
Ok(())
}
#[test]
fn test_get_usage_summary_includes_end_day_rollup_for_minute_precision_end_time(
) -> Result<(), AppError> {
let db = Database::memory()?;
let start = local_ts(2024, 1, 1, 0, 0, 0);
let end = local_ts(2024, 1, 2, 23, 59, 0);
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-01-01",
"claude",
"p1",
"claude-3",
10,
10,
1000,
500,
0,
0,
"1.00",
100
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-01-02",
"claude",
"p1",
"claude-3",
20,
19,
2000,
1000,
0,
0,
"2.00",
120
],
)?;
}
let summary = db.get_usage_summary(Some(start), Some(end), Some("claude"))?;
assert_eq!(summary.total_requests, 30);
assert_eq!(summary.total_input_tokens, 3000);
assert_eq!(summary.total_output_tokens, 1500);
Ok(())
}
#[test]
fn test_effective_usage_dedup_prefers_proxy_for_session_sources() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
insert_usage_log(
&conn,
"codex-proxy",
"codex",
"openai",
"GPT-5.4",
"proxy",
10_000,
100,
20,
10,
7,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"codex-session-dup",
"codex",
"_codex_session",
"gpt-5.4",
"codex_session",
10_060,
100,
20,
10,
0,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"claude-proxy",
"claude",
"openai-compatible",
"claude-sonnet-4-5",
"proxy",
25_000,
300,
60,
20,
5,
200,
"0.30",
)?;
insert_usage_log(
&conn,
"claude-session-dup",
"claude",
"_session",
"claude-sonnet-4-5",
"session_log",
25_060,
300,
60,
20,
5,
200,
"0.30",
)?;
insert_usage_log(
&conn,
"gemini-proxy",
"gemini",
"google",
"gemini-2.5-pro",
"proxy",
20_000,
200,
40,
30,
0,
200,
"0.20",
)?;
insert_usage_log(
&conn,
"gemini-session-dup",
"gemini",
"_gemini_session",
"gemini-2.5-pro",
"gemini_session",
20_060,
200,
40,
30,
0,
200,
"0.20",
)?;
insert_usage_log(
&conn,
"codex-session-only",
"codex",
"_codex_session",
"gpt-5.4",
"codex_session",
30_000,
50,
5,
0,
0,
200,
"0.02",
)?;
}
let summary = db.get_usage_summary(None, None, None)?;
assert_eq!(summary.total_requests, 4);
// codex-proxy contributes 100-10=90; gemini-proxy contributes 200-30=170
// (both cache-inclusive providers). claude-proxy=300, codex-session-only=50.
// 90 + 170 + 300 + 50 = 610.
assert_eq!(summary.total_input_tokens, 610);
assert_eq!(summary.total_output_tokens, 125);
assert_eq!(summary.total_cache_read_tokens, 60);
assert_eq!(summary.total_cache_creation_tokens, 12);
// real_total = fresh_input(610) + output(125) + cache_create(12) + cache_read(60) = 807
assert_eq!(summary.real_total_tokens, 807);
// hit_rate = 60 / (610 + 12 + 60) = 60 / 682
let expected_hit_rate = 60.0_f64 / 682.0_f64;
assert!((summary.cache_hit_rate - expected_hit_rate).abs() < 1e-9);
let trends = db.get_daily_trends(Some(0), Some(40_000), None)?;
assert_eq!(trends.iter().map(|stat| stat.request_count).sum::<u64>(), 4);
let provider_stats = db.get_provider_stats(None, None, None)?;
assert_eq!(
provider_stats
.iter()
.map(|stat| stat.request_count)
.sum::<u64>(),
4
);
assert!(provider_stats
.iter()
.any(|stat| stat.provider_id == "_codex_session" && stat.request_count == 1));
assert!(!provider_stats
.iter()
.any(|stat| stat.provider_id == "_gemini_session"));
assert!(!provider_stats
.iter()
.any(|stat| stat.provider_id == "_session"));
let model_stats = db.get_model_stats(None, None, None)?;
assert_eq!(
model_stats
.iter()
.map(|stat| stat.request_count)
.sum::<u64>(),
4
);
let logs = db.get_request_logs(&LogFilters::default(), 0, 10)?;
let request_ids: Vec<&str> = logs
.data
.iter()
.map(|log| log.request_id.as_str())
.collect();
assert_eq!(logs.total, 4);
assert!(request_ids.contains(&"codex-proxy"));
assert!(request_ids.contains(&"claude-proxy"));
assert!(request_ids.contains(&"gemini-proxy"));
assert!(request_ids.contains(&"codex-session-only"));
assert!(!request_ids.contains(&"codex-session-dup"));
assert!(!request_ids.contains(&"claude-session-dup"));
assert!(!request_ids.contains(&"gemini-session-dup"));
let breakdown = crate::services::session_usage::get_data_source_breakdown(&db)?;
let proxy_count = breakdown
.iter()
.find(|item| item.data_source == "proxy")
.map(|item| item.request_count);
let codex_session_count = breakdown
.iter()
.find(|item| item.data_source == "codex_session")
.map(|item| item.request_count);
let gemini_session_count = breakdown
.iter()
.find(|item| item.data_source == "gemini_session")
.map(|item| item.request_count);
let session_log_count = breakdown
.iter()
.find(|item| item.data_source == "session_log")
.map(|item| item.request_count);
assert_eq!(proxy_count, Some(3));
assert_eq!(codex_session_count, Some(1));
assert_eq!(gemini_session_count, None);
assert_eq!(session_log_count, None);
Ok(())
}
#[test]
fn test_effective_usage_dedup_keeps_non_matching_session_rows() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
insert_usage_log(
&conn,
"proxy-base",
"codex",
"openai",
"gpt-5.4",
"proxy",
10_000,
100,
20,
10,
0,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"session-outside-window",
"codex",
"_codex_session",
"gpt-5.4",
"codex_session",
10_601,
100,
20,
10,
0,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"session-token-mismatch",
"codex",
"_codex_session",
"gpt-5.4",
"codex_session",
10_060,
101,
20,
10,
0,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"session-app-mismatch",
"gemini",
"_gemini_session",
"gpt-5.4",
"gemini_session",
10_060,
100,
20,
10,
0,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"session-model-mismatch",
"codex",
"_codex_session",
"different-model",
"codex_session",
10_060,
100,
20,
10,
0,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"proxy-error",
"codex",
"openai",
"gpt-5.4",
"proxy",
20_000,
300,
60,
0,
0,
500,
"0.00",
)?;
insert_usage_log(
&conn,
"session-matches-error-proxy",
"codex",
"_codex_session",
"gpt-5.4",
"codex_session",
20_060,
300,
60,
0,
0,
200,
"0.30",
)?;
insert_usage_log(
&conn,
"claude-proxy-cache-creation",
"claude",
"anthropic",
"claude-sonnet-4-5",
"proxy",
30_000,
100,
20,
10,
5,
200,
"0.10",
)?;
insert_usage_log(
&conn,
"claude-session-cache-creation-mismatch",
"claude",
"_session",
"claude-sonnet-4-5",
"session_log",
30_060,
100,
20,
10,
0,
200,
"0.10",
)?;
}
let summary = db.get_usage_summary(None, None, None)?;
assert_eq!(summary.total_requests, 9);
let logs = db.get_request_logs(&LogFilters::default(), 0, 10)?;
let request_ids: Vec<&str> = logs
.data
.iter()
.map(|log| log.request_id.as_str())
.collect();
assert_eq!(logs.total, 9);
assert!(request_ids.contains(&"session-outside-window"));
assert!(request_ids.contains(&"session-token-mismatch"));
assert!(request_ids.contains(&"session-app-mismatch"));
assert!(request_ids.contains(&"session-model-mismatch"));
assert!(request_ids.contains(&"session-matches-error-proxy"));
assert!(request_ids.contains(&"claude-session-cache-creation-mismatch"));
Ok(())
}
#[test]
fn test_get_model_stats() -> Result<(), AppError> {
let db = Database::memory()?;
// 插入测试数据
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"req1",
"p1",
"claude",
"claude-3-sonnet",
100,
50,
"0.01",
100,
200,
1000
],
)?;
}
let stats = db.get_model_stats(None, None, None)?;
assert_eq!(stats.len(), 1);
assert_eq!(stats[0].model, "claude-3-sonnet");
assert_eq!(stats[0].request_count, 1);
Ok(())
}
#[test]
fn test_get_provider_stats_with_time_filter() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params!["old", "p1", "claude", "claude-3", 100, 50, "0.01", 100, 200, 1000],
)?;
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params!["new", "p1", "claude", "claude-3", 200, 75, "0.02", 120, 200, 2000],
)?;
}
let stats = db.get_provider_stats(Some(1500), Some(2500), Some("claude"))?;
assert_eq!(stats.len(), 1);
assert_eq!(stats[0].provider_id, "p1");
assert_eq!(stats[0].request_count, 1);
assert_eq!(stats[0].total_tokens, 275);
Ok(())
}
#[test]
fn test_get_provider_stats_excludes_partial_rollup_boundary_days() -> Result<(), AppError> {
let db = Database::memory()?;
let start = local_ts(2024, 2, 1, 12, 0, 0);
let end = local_ts(2024, 2, 3, 12, 0, 0);
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-02-01",
"claude",
"p-rollup",
"claude-3",
5,
5,
500,
250,
0,
0,
"0.50",
100
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-02-02",
"claude",
"p-rollup",
"claude-3",
8,
7,
800,
400,
0,
0,
"0.80",
120
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-02-03",
"claude",
"p-rollup",
"claude-3",
12,
11,
1200,
600,
0,
0,
"1.20",
140
],
)?;
}
let stats = db.get_provider_stats(Some(start), Some(end), Some("claude"))?;
assert_eq!(stats.len(), 1);
assert_eq!(stats[0].provider_id, "p-rollup");
assert_eq!(stats[0].request_count, 8);
assert_eq!(stats[0].total_tokens, 1200);
Ok(())
}
#[test]
fn test_get_daily_trends_respects_shorter_than_24_hours() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"req-short",
"p1",
"claude",
"claude-3",
100,
50,
"0.01",
100,
200,
10_800
],
)?;
}
let stats = db.get_daily_trends(Some(0), Some(15 * 60 * 60), Some("claude"))?;
assert_eq!(stats.len(), 15);
assert_eq!(stats[3].request_count, 1);
Ok(())
}
#[test]
fn test_get_daily_trends_groups_ranges_longer_than_24_hours_by_local_day(
) -> Result<(), AppError> {
let db = Database::memory()?;
let start = local_ts(2024, 3, 1, 12, 0, 0);
let end = local_ts(2024, 3, 3, 12, 0, 0);
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"day-1-detail",
"p1",
"claude",
"claude-3",
100,
50,
"0.01",
100,
200,
local_ts(2024, 3, 1, 13, 0, 0)
],
)?;
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, total_cost_usd,
latency_ms, status_code, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"day-3-detail",
"p1",
"claude",
"claude-3",
200,
75,
"0.02",
110,
200,
local_ts(2024, 3, 3, 10, 0, 0)
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-03-02",
"claude",
"p1",
"claude-3",
4,
4,
400,
200,
0,
0,
"0.40",
120
],
)?;
}
let stats = db.get_daily_trends(Some(start), Some(end), Some("claude"))?;
assert_eq!(stats.len(), 3);
assert_eq!(stats[0].request_count, 1);
assert_eq!(stats[0].total_tokens, 150);
assert_eq!(stats[1].request_count, 4);
assert_eq!(stats[1].total_tokens, 600);
assert_eq!(stats[2].request_count, 1);
assert_eq!(stats[2].total_tokens, 275);
Ok(())
}
#[test]
fn test_get_model_stats_excludes_partial_rollup_boundary_days() -> Result<(), AppError> {
let db = Database::memory()?;
let start = local_ts(2024, 4, 1, 12, 0, 0);
let end = local_ts(2024, 4, 3, 12, 0, 0);
{
let conn = lock_conn!(db.conn);
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-04-01",
"claude",
"p1",
"claude-3-haiku",
6,
6,
600,
300,
0,
0,
"0.60",
100
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-04-02",
"claude",
"p1",
"claude-3-haiku",
9,
8,
900,
450,
0,
0,
"0.90",
110
],
)?;
conn.execute(
"INSERT INTO usage_daily_rollups (
date, app_type, provider_id, model,
request_count, success_count, input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
params![
"2024-04-03",
"claude",
"p1",
"claude-3-haiku",
12,
11,
1200,
600,
0,
0,
"1.20",
130
],
)?;
}
let stats = db.get_model_stats(Some(start), Some(end), Some("claude"))?;
assert_eq!(stats.len(), 1);
assert_eq!(stats[0].model, "claude-3-haiku");
assert_eq!(stats[0].request_count, 9);
assert_eq!(stats[0].total_tokens, 1350);
Ok(())
}
#[test]
fn test_model_pricing_matching() -> Result<(), AppError> {
let db = Database::memory()?;
let conn = lock_conn!(db.conn);
// 准备额外定价数据,覆盖前缀/后缀清洗场景
conn.execute(
"INSERT OR REPLACE INTO model_pricing (
model_id, display_name, input_cost_per_million, output_cost_per_million,
cache_read_cost_per_million, cache_creation_cost_per_million
) VALUES (?, ?, ?, ?, ?, ?)",
params![
"claude-haiku-4.5",
"Claude Haiku 4.5",
"1.0",
"2.0",
"0.0",
"0.0"
],
)?;
// 测试精确匹配(seed_model_pricing 已预置 claude-sonnet-4-5-20250929
let result = find_model_pricing_row(&conn, "claude-sonnet-4-5-20250929")?;
assert!(
result.is_some(),
"应该能精确匹配 claude-sonnet-4-5-20250929"
);
// 清洗:去除前缀和冒号后缀
let result = find_model_pricing_row(&conn, "anthropic/claude-haiku-4.5")?;
assert!(
result.is_some(),
"带前缀的模型 anthropic/claude-haiku-4.5 应能匹配到 claude-haiku-4.5"
);
let result = find_model_pricing_row(&conn, "moonshotai/kimi-k2-0905:exa")?;
assert!(
result.is_some(),
"带前缀+冒号后缀的模型应清洗后匹配到 kimi-k2-0905"
);
// 清洗:@ 替换为 -seed_model_pricing 已预置 gpt-5.2-codex-low
let result = find_model_pricing_row(&conn, "gpt-5.2-codex@low")?;
assert!(
result.is_some(),
"带 @ 分隔符的模型 gpt-5.2-codex@low 应能匹配到 gpt-5.2-codex-low"
);
let result = find_model_pricing_row(&conn, "OpenAI/GPT-5.5@HIGH")?;
assert!(
result.is_some(),
"大小写混合的 GPT-5.5 模型应能归一化匹配到 gpt-5.5-high"
);
// Claude Desktop route 短 ID:应通过前缀匹配到带日期的定价
let result = find_model_pricing_row(&conn, "claude-haiku-4-5")?;
assert!(
result.is_some(),
"Claude Desktop 短路由 claude-haiku-4-5 应能匹配到 claude-haiku-4-5-20251001"
);
// Claude Desktop 旧版/异常包装的非 Anthropic routeclaude-gpt-5.5 → gpt-5.5
let result = find_model_pricing_row(&conn, "claude-gpt-5.5")?;
assert!(
result.is_some(),
"带 claude- 包装的非 Anthropic 模型应能剥离后匹配到真实模型定价"
);
// Bedrock/Vertex 常见形态:provider 前缀 + -vN 后缀 + :0 修饰
let result =
find_model_pricing_row(&conn, "global.anthropic.claude-haiku-4-5-20251001-v1:0")?;
assert!(
result.is_some(),
"Bedrock/Vertex 风格 Claude 模型 ID 应能归一化到基础 Claude 模型定价"
);
// Reasoning effort 后缀:没有专门价格时回退到基础模型
let result = find_model_pricing_row(&conn, "gpt-5.4@low")?;
assert!(
result.is_some(),
"缺少专门 effort 价格时应回退到 gpt-5.4 基础模型定价"
);
// Kimi Code 是订阅/额度模型,不应伪装成公开按 token 计费模型
let result = find_model_pricing_row(&conn, "kimi-for-coding")?;
assert!(result.is_none(), "kimi-for-coding 没有固定 token 单价");
// 测试不存在的模型
let result = find_model_pricing_row(&conn, "unknown-model-123")?;
assert!(result.is_none(), "不应该匹配不存在的模型");
Ok(())
}
}