diff --git a/backend/src/admin_queries/accounts.rs b/backend/src/admin_queries/accounts.rs deleted file mode 100644 index 40cdbbf42..000000000 --- a/backend/src/admin_queries/accounts.rs +++ /dev/null @@ -1,589 +0,0 @@ -//! 账号列表跨域查询与 quota/usage join。 - -use std::{collections::HashMap, sync::Arc}; - -use async_trait::async_trait; -use chrono::{DateTime, Duration, Utc}; -use thiserror::Error; - -use crate::{ - fleet::{ - account::{AccountListSort, AccountStatus}, - manage::{AccountManageService, ManagedAccount}, - quota::{ClassifiedQuotaWindow, QuotaSnapshot}, - refresh::{TokenRefreshService, TokenRefresher, token_refresh_status_eligible}, - }, - infra::{format::nonnegative_i64_to_u64, json::NumberedPage}, - telemetry::{ - account_usage::query::{AccountUsageQueryService, AccountUsageRecord}, - billing, - buckets::query::{ModelUsageWindow, UsageBucketWindow}, - }, -}; - -const ACCOUNT_STATS_PAGE_LIMIT: u32 = 200; - -#[derive(Debug, Clone)] -pub struct AccountListQuery { - pub page: u32, - pub page_size: u32, - pub search: Option, - pub status: Option, - pub sort: Option, -} - -#[derive(Debug, Clone)] -pub struct AccountListResult { - pub page: NumberedPage, - pub summary: AccountSummary, -} - -#[derive(Debug, Clone)] -pub struct AccountListItem { - pub account: ManagedAccount, - pub usage: Option, - pub quota: Option, - pub models: Vec, - pub token_refreshing: bool, -} - -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub struct AccountSummary { - pub total: u64, - pub active: u64, - pub quota_exhausted: u64, - pub attention: u64, -} - -#[derive(Debug, Clone)] -pub struct AccountQuotaReadModel { - pub fetched_at: Option>, - pub windows: Vec, -} - -#[derive(Debug, Clone)] -pub struct AccountQuotaWindowReadModel { - pub quota: ClassifiedQuotaWindow, - pub local_usage: Option, -} - -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub struct AccountQuotaWindowLocalUsage { - pub request_count: i64, - pub input_tokens: i64, - pub output_tokens: i64, - pub cached_tokens: i64, -} - -#[derive(Debug, Clone)] -pub struct AccountQuotaUsageWindow { - pub key: String, - pub start: DateTime, - pub end: DateTime, - pub window_seconds: u64, -} - -impl AccountQuotaReadModel { - pub fn from_snapshot(snapshot: QuotaSnapshot, fetched_at: Option>) -> Self { - Self { - fetched_at, - windows: snapshot - .classified_windows() - .into_iter() - .map(|quota| AccountQuotaWindowReadModel { - quota, - local_usage: None, - }) - .collect(), - } - } - - pub fn usage_windows(&self) -> Vec { - self.windows - .iter() - .filter_map(AccountQuotaWindowReadModel::usage_window) - .collect() - } - - fn apply_local_usage( - &mut self, - usage_by_window: &HashMap, - ) { - for window in &mut self.windows { - if window.usage_window().is_some() { - window.local_usage = Some( - usage_by_window - .get(&window.quota.key) - .copied() - .unwrap_or_default(), - ); - } - } - } -} - -impl AccountQuotaWindowReadModel { - fn usage_window(&self) -> Option { - let end = self.quota.window.reset_datetime()?; - let window_seconds = self.quota.window.window_seconds()?; - let seconds = i64::try_from(window_seconds).ok()?; - let start = end.checked_sub_signed(Duration::seconds(seconds))?; - (start <= end).then(|| AccountQuotaUsageWindow { - key: self.quota.key.clone(), - start, - end, - window_seconds, - }) - } -} - -#[derive(Debug, Clone)] -pub struct AccountModelUsage { - pub model: String, - pub request_count: i64, - pub error_count: i64, - pub input_tokens: i64, - pub output_tokens: i64, - pub cached_tokens: i64, - pub billing_amount_usd: f64, - pub last_used_at: Option>, -} - -#[derive(Debug, Error)] -pub enum AccountListQueryError { - #[error("failed to list accounts")] - Accounts, - #[error("failed to load account usage")] - Usage, - #[error("failed to load account quota usage")] - QuotaUsage, - #[error("failed to load account model usage")] - ModelUsage, - #[error("failed to load token refresh activity")] - RefreshActivity, -} - -#[async_trait] -pub trait RefreshActivityQuery: Send + Sync + 'static { - async fn refreshing_account_ids( - &self, - account_ids: &[String], - now: DateTime, - ) -> Result, AccountListQueryError>; -} - -#[async_trait] -impl RefreshActivityQuery for TokenRefreshService -where - R: TokenRefresher, -{ - async fn refreshing_account_ids( - &self, - account_ids: &[String], - now: DateTime, - ) -> Result, AccountListQueryError> { - TokenRefreshService::refreshing_account_ids(self, account_ids, now) - .await - .map_err(|_| AccountListQueryError::RefreshActivity) - } -} - -#[derive(Clone)] -pub struct AccountListQueryService { - accounts: Arc, - usage: Arc, - refresh_activity: Arc, -} - -impl AccountListQueryService { - pub fn new( - accounts: Arc, - usage: Arc, - refresh_activity: Arc, - ) -> Self { - Self { - accounts, - usage, - refresh_activity, - } - } - - pub async fn query( - &self, - query: AccountListQuery, - ) -> Result { - let page = self - .accounts - .list_page( - query.page, - query.page_size, - query.search, - query.status, - query.sort, - ) - .await - .map_err(|_| AccountListQueryError::Accounts)?; - let all_accounts = self.list_all_accounts().await?; - let summary = account_summary(&all_accounts); - let mut quota_by_account = self.quota_by_account().await?; - let account_ids = page - .items - .iter() - .map(|account| account.id.clone()) - .collect::>(); - let usage_records = self - .usage - .list_by_account_ids(&account_ids) - .await - .map_err(|_| AccountListQueryError::Usage)?; - let refreshing = self - .refresh_activity - .refreshing_account_ids(&account_ids, Utc::now()) - .await?; - - let quota_windows = quota_usage_windows(&page.items, "a_by_account); - let quota_usage = self.quota_local_usage("a_windows).await?; - for account in &page.items { - if let Some(quota) = quota_by_account.get_mut(&account.id) { - quota.apply_local_usage(quota_usage.get(&account.id).unwrap_or(&HashMap::new())); - } - } - let selected_windows = selected_quota_windows(&page.items, "a_by_account); - let usage_by_account = usage_records - .into_iter() - .map(|usage| { - let usage = apply_selected_window_usage(usage, &selected_windows, "a_usage); - (usage.account_id.clone(), usage) - }) - .collect::>(); - let models_by_account = self.model_usage(&usage_by_account).await?; - - let items = page - .items - .into_iter() - .map(|account| { - let account_id = account.id.clone(); - AccountListItem { - token_refreshing: token_refresh_status_eligible(account.status) - && refreshing.contains(&account_id), - usage: usage_by_account.get(&account_id).cloned(), - quota: quota_by_account.remove(&account_id), - models: models_by_account - .get(&account_id) - .cloned() - .unwrap_or_default(), - account, - } - }) - .collect(); - Ok(AccountListResult { - page: NumberedPage { - items, - total: page.total, - page: page.page, - page_size: page.page_size, - }, - summary, - }) - } - - pub async fn enrich_account( - &self, - account: ManagedAccount, - quota: AccountQuotaReadModel, - ) -> Result { - let account_id = account.id.clone(); - let mut quota_by_account = HashMap::from([(account_id.clone(), quota)]); - let windows = quota_usage_windows(std::slice::from_ref(&account), "a_by_account); - let quota_usage = self.quota_local_usage(&windows).await?; - if let Some(quota) = quota_by_account.get_mut(&account_id) { - quota.apply_local_usage(quota_usage.get(&account_id).unwrap_or(&HashMap::new())); - } - let selected = selected_quota_windows(std::slice::from_ref(&account), "a_by_account); - let mut usage = self - .usage - .list_by_account_ids(std::slice::from_ref(&account_id)) - .await - .map_err(|_| AccountListQueryError::Usage)? - .into_iter() - .next() - .map(|usage| apply_selected_window_usage(usage, &selected, "a_usage)); - let usage_by_account = usage - .iter() - .cloned() - .map(|usage| (usage.account_id.clone(), usage)) - .collect::>(); - let models = self - .model_usage(&usage_by_account) - .await? - .remove(&account_id) - .unwrap_or_default(); - let refreshing = self - .refresh_activity - .refreshing_account_ids(std::slice::from_ref(&account_id), Utc::now()) - .await?; - Ok(AccountListItem { - token_refreshing: token_refresh_status_eligible(account.status) - && refreshing.contains(&account_id), - account, - usage: usage.take(), - quota: quota_by_account.remove(&account_id), - models, - }) - } - - async fn list_all_accounts(&self) -> Result, AccountListQueryError> { - let mut page_number = 1; - let mut accounts = Vec::new(); - loop { - let page = self - .accounts - .list_page(page_number, ACCOUNT_STATS_PAGE_LIMIT, None, None, None) - .await - .map_err(|_| AccountListQueryError::Accounts)?; - let total = page.total; - accounts.extend(page.items); - if accounts.len() as u64 >= total || total == 0 { - return Ok(accounts); - } - page_number = page_number.saturating_add(1); - } - } - - async fn quota_by_account( - &self, - ) -> Result, AccountListQueryError> { - let snapshots = self - .accounts - .quota_snapshots() - .await - .map_err(|_| AccountListQueryError::Accounts)?; - Ok(snapshots - .into_iter() - .filter_map(|snapshot| { - let quota = QuotaSnapshot::from_json(&snapshot.quota_json).ok()?; - Some(( - snapshot.account_id, - AccountQuotaReadModel::from_snapshot(quota, snapshot.quota_fetched_at), - )) - }) - .collect()) - } - - async fn quota_local_usage( - &self, - windows: &[AccountQuotaWindowSelection], - ) -> Result>, AccountListQueryError> - { - let queries = windows - .iter() - .map(|selection| UsageBucketWindow { - account_id: selection.account_id.clone(), - key: selection.window.key.clone(), - start: selection.window.start, - end: selection.window.end, - }) - .collect::>(); - self.usage - .usage_by_windows(&queries) - .await - .map_err(|_| AccountListQueryError::QuotaUsage) - .map(|usage| { - usage - .into_iter() - .map(|(account_id, windows)| { - let windows = windows - .into_iter() - .map(|(key, usage)| { - ( - key, - AccountQuotaWindowLocalUsage { - request_count: usage.request_count, - input_tokens: usage.input_tokens, - output_tokens: usage.output_tokens, - cached_tokens: usage.cached_tokens, - }, - ) - }) - .collect(); - (account_id, windows) - }) - .collect() - }) - } - - async fn model_usage( - &self, - usage_by_account: &HashMap, - ) -> Result>, AccountListQueryError> { - let now = Utc::now(); - let windows = usage_by_account - .values() - .filter_map(|usage| { - current_usage_window(usage, now).map(|(start, end)| ModelUsageWindow { - account_id: usage.account_id.clone(), - start, - end, - }) - }) - .collect::>(); - if windows.is_empty() { - return Ok(HashMap::new()); - } - let rows = self - .usage - .model_usage_by_windows(&windows) - .await - .map_err(|_| AccountListQueryError::ModelUsage)?; - let mut records = HashMap::<(String, String), AccountModelUsage>::new(); - for row in rows { - let billing_amount_usd = billing::calculate_billing_amount( - nonnegative_i64_to_u64(row.input_tokens), - nonnegative_i64_to_u64(row.output_tokens), - nonnegative_i64_to_u64(row.cached_tokens), - nonnegative_i64_to_u64(row.cache_write_tokens), - &row.model, - row.service_tier.as_deref(), - ); - let record = records - .entry((row.account_id.clone(), row.model.clone())) - .or_insert_with(|| AccountModelUsage { - model: row.model, - request_count: 0, - error_count: 0, - input_tokens: 0, - output_tokens: 0, - cached_tokens: 0, - billing_amount_usd: 0.0, - last_used_at: None, - }); - record.request_count += row.request_count; - record.error_count += row.error_count; - record.input_tokens += row.input_tokens; - record.output_tokens += row.output_tokens; - record.cached_tokens += row.cached_tokens; - record.billing_amount_usd += billing_amount_usd; - record.last_used_at = record.last_used_at.max(row.last_used_at); - } - let mut by_account = HashMap::>::new(); - for ((account_id, _), record) in records { - by_account.entry(account_id).or_default().push(record); - } - for records in by_account.values_mut() { - records.sort_by(|left, right| { - right - .request_count - .cmp(&left.request_count) - .then_with(|| right.last_used_at.cmp(&left.last_used_at)) - .then_with(|| left.model.cmp(&right.model)) - }); - } - Ok(by_account) - } -} - -#[derive(Debug, Clone)] -struct AccountQuotaWindowSelection { - account_id: String, - window: AccountQuotaUsageWindow, -} - -fn account_summary(accounts: &[ManagedAccount]) -> AccountSummary { - AccountSummary { - total: accounts.len() as u64, - active: accounts - .iter() - .filter(|account| account.status == AccountStatus::Active) - .count() as u64, - quota_exhausted: accounts - .iter() - .filter(|account| account.status == AccountStatus::QuotaExhausted) - .count() as u64, - attention: accounts - .iter() - .filter(|account| { - matches!( - account.status, - AccountStatus::Expired | AccountStatus::Disabled | AccountStatus::Banned - ) - }) - .count() as u64, - } -} - -fn quota_usage_windows( - accounts: &[ManagedAccount], - quota_by_account: &HashMap, -) -> Vec { - accounts - .iter() - .filter_map(|account| { - quota_by_account - .get(&account.id) - .map(|quota| (account.id.as_str(), quota)) - }) - .flat_map(|(account_id, quota)| { - quota - .usage_windows() - .into_iter() - .map(move |window| AccountQuotaWindowSelection { - account_id: account_id.to_string(), - window, - }) - }) - .collect() -} - -fn selected_quota_windows( - accounts: &[ManagedAccount], - quota_by_account: &HashMap, -) -> HashMap { - accounts - .iter() - .filter_map(|account| { - let window = quota_by_account - .get(&account.id)? - .usage_windows() - .into_iter() - .max_by(|left, right| { - left.window_seconds - .cmp(&right.window_seconds) - .then_with(|| left.end.cmp(&right.end)) - .then_with(|| left.key.cmp(&right.key)) - })?; - Some((account.id.clone(), window)) - }) - .collect() -} - -fn apply_selected_window_usage( - mut usage: AccountUsageRecord, - selected: &HashMap, - quota_usage: &HashMap>, -) -> AccountUsageRecord { - let Some(window) = selected.get(&usage.account_id) else { - return usage; - }; - let stats = quota_usage - .get(&usage.account_id) - .and_then(|windows| windows.get(&window.key)) - .copied() - .unwrap_or_default(); - usage.window_request_count = stats.request_count; - usage.window_input_tokens = stats.input_tokens; - usage.window_output_tokens = stats.output_tokens; - usage.window_cached_tokens = stats.cached_tokens; - usage.window_started_at = Some(window.start); - usage.window_reset_at = Some(window.end); - usage.limit_window_seconds = Some(window.window_seconds); - usage -} - -fn current_usage_window( - usage: &AccountUsageRecord, - now: DateTime, -) -> Option<(DateTime, DateTime)> { - let start = usage.window_started_at?; - let end = usage.window_reset_at.unwrap_or(now); - (start <= end).then_some((start, end)) -} diff --git a/backend/src/admin_queries/dashboard.rs b/backend/src/admin_queries/dashboard.rs deleted file mode 100644 index f04225095..000000000 --- a/backend/src/admin_queries/dashboard.rs +++ /dev/null @@ -1,345 +0,0 @@ -//! Dashboard 跨域查询与账号/quota join。 - -use std::{collections::HashMap, sync::Arc}; - -use chrono::{DateTime, Duration, Utc}; -use thiserror::Error; - -use crate::{ - admin_queries::accounts::RefreshActivityQuery, - fleet::{ - account::{Account, AccountStatus}, - pool::{AccountCapacitySummary, AccountPoolService}, - quota::QuotaSnapshot, - refresh::token_refresh_status_eligible, - store::AccountStore, - }, - infra::time::{china_day_start, china_quarter_hour_start}, - settings::service::SettingsService, - telemetry::{ - account_usage::query::{ - AccountUsageQueryService, AccountUsageTimeBucket, RetainedUsageSummary, - }, - usage::{ - insights::RequestHealthTimeBucket, - query::{UsageQueryFilter, UsageQueryService}, - types::UsageRecord, - }, - }, -}; - -const ACCOUNT_USAGE_LIMIT: u32 = 4; -const USAGE_RECORD_LIMIT: u32 = 10; -const TIME_BUCKET_MINUTES: i64 = 15; -const TIME_BUCKET_SLOTS: i64 = 7 * 24 * 4; - -/// Dashboard 消费的请求画像快照端口。 -pub trait DashboardWireProfileQuery: Send + Sync { - fn snapshot(&self) -> DashboardWireProfile; -} - -/// Dashboard 消费的 Desktop 发布观测端口。 -pub trait DashboardDesktopReleaseQuery: Send + Sync { - fn snapshot(&self) -> DashboardDesktopReleaseSnapshot; -} - -#[derive(Debug, Clone)] -pub struct DashboardWireProfile { - pub originator: String, - pub codex_version: String, - pub desktop_version: String, - pub desktop_build: String, - pub os_type: String, - pub os_version: String, - pub arch: String, - pub terminal: String, - pub user_agent: String, - pub verified_at: DateTime, -} - -#[derive(Debug, Clone, Default)] -pub struct DashboardDesktopReleaseSnapshot { - pub checked_at: Option>, - pub latest: Option, - pub last_error: Option, -} - -#[derive(Debug, Clone)] -pub struct DashboardDesktopRelease { - pub version: String, - pub build: String, - pub published_at: Option>, - pub minimum_system_version: Option, - pub hardware_requirements: Option, - pub download_url: Option, - pub download_size: Option, - pub signature_present: bool, -} - -#[derive(Debug, Clone)] -pub struct DashboardReadModel { - pub account_counts: DashboardAccountCounts, - pub retained_usage: RetainedUsageSummary, - pub time_buckets: Vec, - pub health_buckets: Vec, - pub account_usage: Vec, - pub usage_records: Vec, - pub account_emails: HashMap, - pub pool: DashboardPoolSummary, - pub capacity: AccountCapacitySummary, - pub rotation_strategy: String, - pub wire_profile: DashboardWireProfile, - pub desktop_release: DashboardDesktopReleaseSnapshot, -} - -#[derive(Debug, Clone, Copy, Default)] -pub struct DashboardAccountCounts { - pub total: u64, - pub enabled: u64, - pub abnormal: u64, -} - -#[derive(Debug, Clone)] -pub struct DashboardAccountUsage { - pub id: String, - pub email: String, - pub plan_type: Option, - pub total_tokens: u64, - pub quota_used_percent: Option, - pub last_used_at: DateTime, -} - -#[derive(Debug, Clone, Copy, Default)] -pub struct DashboardPoolSummary { - pub total: usize, - pub active: usize, - pub expired: usize, - pub quota_exhausted: usize, - pub refreshing: usize, - pub disabled: usize, - pub banned: usize, -} - -#[derive(Debug, Error)] -pub enum DashboardQueryError { - #[error("failed to load dashboard {0}")] - Data(&'static str), -} - -#[derive(Clone)] -pub struct DashboardQueryService { - accounts: Arc, - usage: Arc, - usage_records: Arc, - account_pool: Arc, - refresh_activity: Arc, - settings: Arc, - wire_profile: Arc, - desktop_release: Arc, -} - -pub struct DashboardQueryServiceParts { - pub accounts: Arc, - pub usage: Arc, - pub usage_records: Arc, - pub account_pool: Arc, - pub refresh_activity: Arc, - pub settings: Arc, - pub wire_profile: Arc, - pub desktop_release: Arc, -} - -impl DashboardQueryService { - pub fn new(parts: DashboardQueryServiceParts) -> Self { - Self { - accounts: parts.accounts, - usage: parts.usage, - usage_records: parts.usage_records, - account_pool: parts.account_pool, - refresh_activity: parts.refresh_activity, - settings: parts.settings, - wire_profile: parts.wire_profile, - desktop_release: parts.desktop_release, - } - } - - pub async fn summary( - &self, - now: DateTime, - ) -> Result { - let accounts = self - .accounts - .list_pool_accounts() - .await - .map_err(|_| DashboardQueryError::Data("accounts"))?; - let capacity = self.account_pool.capacity_summary_now().await; - let filter = UsageQueryFilter { - start_time: Some(china_day_start(now)), - end_time: Some(now), - ..UsageQueryFilter::default() - }; - let retained_usage = self - .usage - .retained_summary() - .await - .map_err(|_| DashboardQueryError::Data("retained usage summary"))?; - let account_usage_records = self - .usage_records - .account_usage(filter.clone(), ACCOUNT_USAGE_LIMIT) - .await - .map_err(|_| DashboardQueryError::Data("account usage ranking"))?; - let usage_records = self - .usage_records - .list_recent(USAGE_RECORD_LIMIT, filter) - .await - .map_err(|_| DashboardQueryError::Data("recent usage records"))?; - let time_buckets = self.time_buckets(now).await?; - let health_buckets = self.health_buckets(now).await?; - let account_ids = accounts - .iter() - .map(|account| account.id.clone()) - .collect::>(); - let refreshing = self - .refresh_activity - .refreshing_account_ids(&account_ids, now) - .await - .map_err(|_| DashboardQueryError::Data("refreshing accounts"))?; - let quota_used = self.quota_used_by_account(&account_usage_records).await?; - let account_emails = self - .usage_records - .account_email_map(&usage_records) - .await - .map_err(|_| DashboardQueryError::Data("usage record accounts"))?; - - Ok(DashboardReadModel { - account_counts: account_counts(&accounts), - retained_usage, - time_buckets, - health_buckets, - account_usage: account_usage(&accounts, &account_usage_records, "a_used), - usage_records, - account_emails, - pool: pool_summary(&accounts, &refreshing), - capacity, - rotation_strategy: self.settings.current().rotation_strategy, - wire_profile: self.wire_profile.snapshot(), - desktop_release: self.desktop_release.snapshot(), - }) - } - - pub async fn time_buckets( - &self, - now: DateTime, - ) -> Result, DashboardQueryError> { - let current_slot = china_quarter_hour_start(now); - let start = current_slot - Duration::minutes(TIME_BUCKET_MINUTES * (TIME_BUCKET_SLOTS - 1)); - self.usage - .time_buckets(start, now) - .await - .map_err(|_| DashboardQueryError::Data("time buckets")) - } - - async fn health_buckets( - &self, - now: DateTime, - ) -> Result, DashboardQueryError> { - self.usage_records - .health_timeline(china_day_start(now), now) - .await - .map_err(|_| DashboardQueryError::Data("health timeline")) - } - - async fn quota_used_by_account( - &self, - usage: &[crate::telemetry::usage::query::UsageRecordAccountUsage], - ) -> Result, DashboardQueryError> { - let mut quota_used = HashMap::with_capacity(usage.len()); - for usage in usage { - let quota_json = self - .accounts - .get_quota_json(&usage.account_id) - .await - .map_err(|_| DashboardQueryError::Data("account quota"))?; - let Some(quota) = quota_json - .as_deref() - .and_then(|value| QuotaSnapshot::from_json(value).ok()) - else { - continue; - }; - if let Some(used_percent) = quota.representative_used_percent() { - quota_used.insert(usage.account_id.clone(), used_percent); - } - } - Ok(quota_used) - } -} - -fn account_counts(accounts: &[Account]) -> DashboardAccountCounts { - DashboardAccountCounts { - total: accounts.len() as u64, - enabled: accounts - .iter() - .filter(|account| account.status == AccountStatus::Active) - .count() as u64, - abnormal: accounts - .iter() - .filter(|account| account.status != AccountStatus::Active) - .count() as u64, - } -} - -fn pool_summary( - accounts: &[Account], - refreshing: &std::collections::HashSet, -) -> DashboardPoolSummary { - let mut summary = DashboardPoolSummary { - total: accounts.len(), - ..DashboardPoolSummary::default() - }; - for account in accounts { - match account.status { - status if token_refresh_status_eligible(status) && refreshing.contains(&account.id) => { - summary.refreshing += 1; - } - AccountStatus::Active => summary.active += 1, - AccountStatus::Expired => summary.expired += 1, - AccountStatus::QuotaExhausted => summary.quota_exhausted += 1, - AccountStatus::Disabled => summary.disabled += 1, - AccountStatus::Banned => summary.banned += 1, - } - } - summary -} - -fn account_usage( - accounts: &[Account], - usage_records: &[crate::telemetry::usage::query::UsageRecordAccountUsage], - quota_used: &HashMap, -) -> Vec { - let accounts = accounts - .iter() - .map(|account| (account.id.as_str(), account)) - .collect::>(); - usage_records - .iter() - .map(|usage| { - let account = accounts.get(usage.account_id.as_str()).copied(); - DashboardAccountUsage { - id: usage.account_id.clone(), - email: account - .and_then(|account| account.email.clone()) - .unwrap_or_else(|| usage.account_id.clone()), - plan_type: account.and_then(|account| account.plan_type.clone()), - total_tokens: usage.total_tokens, - quota_used_percent: quota_used.get(&usage.account_id).copied().or_else(|| { - matches!( - account.map(|account| account.status), - Some(AccountStatus::QuotaExhausted) - ) - .then_some(100.0) - }), - last_used_at: usage.last_used_at, - } - }) - .collect() -} diff --git a/backend/src/admin_queries/mod.rs b/backend/src/admin_queries/mod.rs deleted file mode 100644 index 5aeef657f..000000000 --- a/backend/src/admin_queries/mod.rs +++ /dev/null @@ -1,4 +0,0 @@ -//! 管理端跨域只读用例。 - -pub mod accounts; -pub mod dashboard; diff --git a/backend/src/api/admin/accounts_routes/mod.rs b/backend/src/api/admin/accounts_routes/mod.rs index 47a03d418..ec7a6ab81 100644 --- a/backend/src/api/admin/accounts_routes/mod.rs +++ b/backend/src/api/admin/accounts_routes/mod.rs @@ -15,7 +15,6 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::{ - admin_queries::accounts::{AccountListQuery, AccountSummary}, api::AppState, api::admin::{ response::{ @@ -448,8 +447,7 @@ pub(crate) async fn accounts( let sort = account_list_sort(params.sort_by, params.sort_direction)?; let result = state .services - .admin_queries - .accounts + .account_list .query(AccountListQuery { page, page_size, @@ -501,7 +499,7 @@ mod import_routes; mod lifecycle_routes; mod oauth_routes; mod probe_routes; -mod query; +pub(crate) mod query; mod quota_presenter; mod quota_routes; diff --git a/backend/src/api/admin/accounts_routes/query.rs b/backend/src/api/admin/accounts_routes/query.rs index c32e8b095..36a1e2743 100644 --- a/backend/src/api/admin/accounts_routes/query.rs +++ b/backend/src/api/admin/accounts_routes/query.rs @@ -1,6 +1,594 @@ +use std::{collections::HashMap, sync::Arc}; + +use async_trait::async_trait; +use chrono::{DateTime, Duration, Utc}; +use thiserror::Error; + use super::*; -use crate::admin_queries::accounts::{AccountListItem, AccountModelUsage}; +use crate::{ + fleet::{ + manage::AccountManageService, + quota::{ClassifiedQuotaWindow, QuotaSnapshot}, + refresh::{TokenRefreshService, TokenRefresher, token_refresh_status_eligible}, + }, + infra::json::NumberedPage, + telemetry::{ + account_usage::query::AccountUsageQueryService, + billing, + buckets::query::{ModelUsageWindow, UsageBucketWindow}, + }, +}; + +const ACCOUNT_STATS_PAGE_LIMIT: u32 = 200; + +#[derive(Debug, Clone)] +pub(crate) struct AccountListQuery { + pub(crate) page: u32, + pub(crate) page_size: u32, + pub(crate) search: Option, + pub(crate) status: Option, + pub(crate) sort: Option, +} + +#[derive(Debug, Clone)] +pub(crate) struct AccountListResult { + pub(crate) page: NumberedPage, + pub(crate) summary: AccountSummary, +} + +#[derive(Debug, Clone)] +pub(crate) struct AccountListItem { + pub(crate) account: ManagedAccount, + pub(crate) usage: Option, + pub(crate) quota: Option, + pub(crate) models: Vec, + pub(crate) token_refreshing: bool, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct AccountSummary { + pub(crate) total: u64, + pub(crate) active: u64, + pub(crate) quota_exhausted: u64, + pub(crate) attention: u64, +} + +#[derive(Debug, Clone)] +pub(crate) struct AccountQuotaReadModel { + pub(crate) fetched_at: Option>, + pub(crate) windows: Vec, +} + +#[derive(Debug, Clone)] +pub(crate) struct AccountQuotaWindowReadModel { + pub(crate) quota: ClassifiedQuotaWindow, + pub(crate) local_usage: Option, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct AccountQuotaWindowLocalUsage { + pub(crate) request_count: i64, + pub(crate) input_tokens: i64, + pub(crate) output_tokens: i64, + pub(crate) cached_tokens: i64, +} + +#[derive(Debug, Clone)] +pub(crate) struct AccountQuotaUsageWindow { + pub(crate) key: String, + pub(crate) start: DateTime, + pub(crate) end: DateTime, + pub(crate) window_seconds: u64, +} + +impl AccountQuotaReadModel { + pub(crate) fn from_snapshot( + snapshot: QuotaSnapshot, + fetched_at: Option>, + ) -> Self { + Self { + fetched_at, + windows: snapshot + .classified_windows() + .into_iter() + .map(|quota| AccountQuotaWindowReadModel { + quota, + local_usage: None, + }) + .collect(), + } + } + + pub(crate) fn usage_windows(&self) -> Vec { + self.windows + .iter() + .filter_map(AccountQuotaWindowReadModel::usage_window) + .collect() + } + + fn apply_local_usage( + &mut self, + usage_by_window: &HashMap, + ) { + for window in &mut self.windows { + if window.usage_window().is_some() { + window.local_usage = Some( + usage_by_window + .get(&window.quota.key) + .copied() + .unwrap_or_default(), + ); + } + } + } +} + +impl AccountQuotaWindowReadModel { + fn usage_window(&self) -> Option { + let end = self.quota.window.reset_datetime()?; + let window_seconds = self.quota.window.window_seconds()?; + let seconds = i64::try_from(window_seconds).ok()?; + let start = end.checked_sub_signed(Duration::seconds(seconds))?; + (start <= end).then(|| AccountQuotaUsageWindow { + key: self.quota.key.clone(), + start, + end, + window_seconds, + }) + } +} + +#[derive(Debug, Clone)] +pub(crate) struct AccountModelUsage { + pub(crate) model: String, + pub(crate) request_count: i64, + pub(crate) error_count: i64, + pub(crate) input_tokens: i64, + pub(crate) output_tokens: i64, + pub(crate) cached_tokens: i64, + pub(crate) billing_amount_usd: f64, + pub(crate) last_used_at: Option>, +} + +#[derive(Debug, Error)] +pub(crate) enum AccountListQueryError { + #[error("failed to list accounts")] + Accounts, + #[error("failed to load account usage")] + Usage, + #[error("failed to load account quota usage")] + QuotaUsage, + #[error("failed to load account model usage")] + ModelUsage, + #[error("failed to load token refresh activity")] + RefreshActivity, +} + +#[async_trait] +pub(crate) trait RefreshActivityQuery: Send + Sync + 'static { + async fn refreshing_account_ids( + &self, + account_ids: &[String], + now: DateTime, + ) -> Result, AccountListQueryError>; +} + +#[async_trait] +impl RefreshActivityQuery for TokenRefreshService +where + R: TokenRefresher, +{ + async fn refreshing_account_ids( + &self, + account_ids: &[String], + now: DateTime, + ) -> Result, AccountListQueryError> { + TokenRefreshService::refreshing_account_ids(self, account_ids, now) + .await + .map_err(|_| AccountListQueryError::RefreshActivity) + } +} + +#[derive(Clone)] +pub(crate) struct AccountListQueryService { + accounts: Arc, + usage: Arc, + refresh_activity: Arc, +} + +impl AccountListQueryService { + pub(crate) fn new( + accounts: Arc, + usage: Arc, + refresh_activity: Arc, + ) -> Self { + Self { + accounts, + usage, + refresh_activity, + } + } + + pub(crate) async fn query( + &self, + query: AccountListQuery, + ) -> Result { + let page = self + .accounts + .list_page( + query.page, + query.page_size, + query.search, + query.status, + query.sort, + ) + .await + .map_err(|_| AccountListQueryError::Accounts)?; + let all_accounts = self.list_all_accounts().await?; + let summary = account_summary(&all_accounts); + let mut quota_by_account = self.quota_by_account().await?; + let account_ids = page + .items + .iter() + .map(|account| account.id.clone()) + .collect::>(); + let usage_records = self + .usage + .list_by_account_ids(&account_ids) + .await + .map_err(|_| AccountListQueryError::Usage)?; + let refreshing = self + .refresh_activity + .refreshing_account_ids(&account_ids, Utc::now()) + .await?; + + let quota_windows = quota_usage_windows(&page.items, "a_by_account); + let quota_usage = self.quota_local_usage("a_windows).await?; + for account in &page.items { + if let Some(quota) = quota_by_account.get_mut(&account.id) { + quota.apply_local_usage(quota_usage.get(&account.id).unwrap_or(&HashMap::new())); + } + } + let selected_windows = selected_quota_windows(&page.items, "a_by_account); + let usage_by_account = usage_records + .into_iter() + .map(|usage| { + let usage = apply_selected_window_usage(usage, &selected_windows, "a_usage); + (usage.account_id.clone(), usage) + }) + .collect::>(); + let models_by_account = self.model_usage(&usage_by_account).await?; + + let items = page + .items + .into_iter() + .map(|account| { + let account_id = account.id.clone(); + AccountListItem { + token_refreshing: token_refresh_status_eligible(account.status) + && refreshing.contains(&account_id), + usage: usage_by_account.get(&account_id).cloned(), + quota: quota_by_account.remove(&account_id), + models: models_by_account + .get(&account_id) + .cloned() + .unwrap_or_default(), + account, + } + }) + .collect(); + Ok(AccountListResult { + page: NumberedPage { + items, + total: page.total, + page: page.page, + page_size: page.page_size, + }, + summary, + }) + } + + pub(crate) async fn enrich_account( + &self, + account: ManagedAccount, + quota: AccountQuotaReadModel, + ) -> Result { + let account_id = account.id.clone(); + let mut quota_by_account = HashMap::from([(account_id.clone(), quota)]); + let windows = quota_usage_windows(std::slice::from_ref(&account), "a_by_account); + let quota_usage = self.quota_local_usage(&windows).await?; + if let Some(quota) = quota_by_account.get_mut(&account_id) { + quota.apply_local_usage(quota_usage.get(&account_id).unwrap_or(&HashMap::new())); + } + let selected = selected_quota_windows(std::slice::from_ref(&account), "a_by_account); + let mut usage = self + .usage + .list_by_account_ids(std::slice::from_ref(&account_id)) + .await + .map_err(|_| AccountListQueryError::Usage)? + .into_iter() + .next() + .map(|usage| apply_selected_window_usage(usage, &selected, "a_usage)); + let usage_by_account = usage + .iter() + .cloned() + .map(|usage| (usage.account_id.clone(), usage)) + .collect::>(); + let models = self + .model_usage(&usage_by_account) + .await? + .remove(&account_id) + .unwrap_or_default(); + let refreshing = self + .refresh_activity + .refreshing_account_ids(std::slice::from_ref(&account_id), Utc::now()) + .await?; + Ok(AccountListItem { + token_refreshing: token_refresh_status_eligible(account.status) + && refreshing.contains(&account_id), + account, + usage: usage.take(), + quota: quota_by_account.remove(&account_id), + models, + }) + } + + async fn list_all_accounts(&self) -> Result, AccountListQueryError> { + let mut page_number = 1; + let mut accounts = Vec::new(); + loop { + let page = self + .accounts + .list_page(page_number, ACCOUNT_STATS_PAGE_LIMIT, None, None, None) + .await + .map_err(|_| AccountListQueryError::Accounts)?; + let total = page.total; + accounts.extend(page.items); + if accounts.len() as u64 >= total || total == 0 { + return Ok(accounts); + } + page_number = page_number.saturating_add(1); + } + } + + async fn quota_by_account( + &self, + ) -> Result, AccountListQueryError> { + let snapshots = self + .accounts + .quota_snapshots() + .await + .map_err(|_| AccountListQueryError::Accounts)?; + Ok(snapshots + .into_iter() + .filter_map(|snapshot| { + let quota = QuotaSnapshot::from_json(&snapshot.quota_json).ok()?; + Some(( + snapshot.account_id, + AccountQuotaReadModel::from_snapshot(quota, snapshot.quota_fetched_at), + )) + }) + .collect()) + } + + async fn quota_local_usage( + &self, + windows: &[AccountQuotaWindowSelection], + ) -> Result>, AccountListQueryError> + { + let queries = windows + .iter() + .map(|selection| UsageBucketWindow { + account_id: selection.account_id.clone(), + key: selection.window.key.clone(), + start: selection.window.start, + end: selection.window.end, + }) + .collect::>(); + self.usage + .usage_by_windows(&queries) + .await + .map_err(|_| AccountListQueryError::QuotaUsage) + .map(|usage| { + usage + .into_iter() + .map(|(account_id, windows)| { + let windows = windows + .into_iter() + .map(|(key, usage)| { + ( + key, + AccountQuotaWindowLocalUsage { + request_count: usage.request_count, + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + cached_tokens: usage.cached_tokens, + }, + ) + }) + .collect(); + (account_id, windows) + }) + .collect() + }) + } + + async fn model_usage( + &self, + usage_by_account: &HashMap, + ) -> Result>, AccountListQueryError> { + let now = Utc::now(); + let windows = usage_by_account + .values() + .filter_map(|usage| { + current_usage_window(usage, now).map(|(start, end)| ModelUsageWindow { + account_id: usage.account_id.clone(), + start, + end, + }) + }) + .collect::>(); + if windows.is_empty() { + return Ok(HashMap::new()); + } + let rows = self + .usage + .model_usage_by_windows(&windows) + .await + .map_err(|_| AccountListQueryError::ModelUsage)?; + let mut records = HashMap::<(String, String), AccountModelUsage>::new(); + for row in rows { + let billing_amount_usd = billing::calculate_billing_amount( + nonnegative_i64_to_u64(row.input_tokens), + nonnegative_i64_to_u64(row.output_tokens), + nonnegative_i64_to_u64(row.cached_tokens), + nonnegative_i64_to_u64(row.cache_write_tokens), + &row.model, + row.service_tier.as_deref(), + ); + let record = records + .entry((row.account_id.clone(), row.model.clone())) + .or_insert_with(|| AccountModelUsage { + model: row.model, + request_count: 0, + error_count: 0, + input_tokens: 0, + output_tokens: 0, + cached_tokens: 0, + billing_amount_usd: 0.0, + last_used_at: None, + }); + record.request_count += row.request_count; + record.error_count += row.error_count; + record.input_tokens += row.input_tokens; + record.output_tokens += row.output_tokens; + record.cached_tokens += row.cached_tokens; + record.billing_amount_usd += billing_amount_usd; + record.last_used_at = record.last_used_at.max(row.last_used_at); + } + let mut by_account = HashMap::>::new(); + for ((account_id, _), record) in records { + by_account.entry(account_id).or_default().push(record); + } + for records in by_account.values_mut() { + records.sort_by(|left, right| { + right + .request_count + .cmp(&left.request_count) + .then_with(|| right.last_used_at.cmp(&left.last_used_at)) + .then_with(|| left.model.cmp(&right.model)) + }); + } + Ok(by_account) + } +} + +#[derive(Debug, Clone)] +struct AccountQuotaWindowSelection { + account_id: String, + window: AccountQuotaUsageWindow, +} + +fn account_summary(accounts: &[ManagedAccount]) -> AccountSummary { + AccountSummary { + total: accounts.len() as u64, + active: accounts + .iter() + .filter(|account| account.status == AccountStatus::Active) + .count() as u64, + quota_exhausted: accounts + .iter() + .filter(|account| account.status == AccountStatus::QuotaExhausted) + .count() as u64, + attention: accounts + .iter() + .filter(|account| { + matches!( + account.status, + AccountStatus::Expired | AccountStatus::Disabled | AccountStatus::Banned + ) + }) + .count() as u64, + } +} + +fn quota_usage_windows( + accounts: &[ManagedAccount], + quota_by_account: &HashMap, +) -> Vec { + accounts + .iter() + .filter_map(|account| { + quota_by_account + .get(&account.id) + .map(|quota| (account.id.as_str(), quota)) + }) + .flat_map(|(account_id, quota)| { + quota + .usage_windows() + .into_iter() + .map(move |window| AccountQuotaWindowSelection { + account_id: account_id.to_string(), + window, + }) + }) + .collect() +} + +fn selected_quota_windows( + accounts: &[ManagedAccount], + quota_by_account: &HashMap, +) -> HashMap { + accounts + .iter() + .filter_map(|account| { + let window = quota_by_account + .get(&account.id)? + .usage_windows() + .into_iter() + .max_by(|left, right| { + left.window_seconds + .cmp(&right.window_seconds) + .then_with(|| left.end.cmp(&right.end)) + .then_with(|| left.key.cmp(&right.key)) + })?; + Some((account.id.clone(), window)) + }) + .collect() +} + +fn apply_selected_window_usage( + mut usage: AccountUsageRecord, + selected: &HashMap, + quota_usage: &HashMap>, +) -> AccountUsageRecord { + let Some(window) = selected.get(&usage.account_id) else { + return usage; + }; + let stats = quota_usage + .get(&usage.account_id) + .and_then(|windows| windows.get(&window.key)) + .copied() + .unwrap_or_default(); + usage.window_request_count = stats.request_count; + usage.window_input_tokens = stats.input_tokens; + usage.window_output_tokens = stats.output_tokens; + usage.window_cached_tokens = stats.cached_tokens; + usage.window_started_at = Some(window.start); + usage.window_reset_at = Some(window.end); + usage.limit_window_seconds = Some(window.window_seconds); + usage +} + +fn current_usage_window( + usage: &AccountUsageRecord, + now: DateTime, +) -> Option<(DateTime, DateTime)> { + let start = usage.window_started_at?; + let end = usage.window_reset_at.unwrap_or(now); + (start <= end).then_some((start, end)) +} pub(super) fn account_display_status( status: AccountStatus, diff --git a/backend/src/api/admin/accounts_routes/quota_presenter.rs b/backend/src/api/admin/accounts_routes/quota_presenter.rs index 060ab8a7b..c914b1576 100644 --- a/backend/src/api/admin/accounts_routes/quota_presenter.rs +++ b/backend/src/api/admin/accounts_routes/quota_presenter.rs @@ -4,7 +4,7 @@ use chrono::{DateTime, Utc}; use serde::Serialize; use crate::{ - admin_queries::accounts::{AccountQuotaReadModel, AccountQuotaWindowLocalUsage}, + api::admin::accounts_routes::query::{AccountQuotaReadModel, AccountQuotaWindowLocalUsage}, fleet::quota::{QuotaWindowKind, QuotaWindowRole}, infra::{ format::{ diff --git a/backend/src/api/admin/accounts_routes/quota_routes.rs b/backend/src/api/admin/accounts_routes/quota_routes.rs index a98531d96..b38e96565 100644 --- a/backend/src/api/admin/accounts_routes/quota_routes.rs +++ b/backend/src/api/admin/accounts_routes/quota_routes.rs @@ -17,10 +17,7 @@ pub async fn account_quota( let quota = serde_json::to_value(&data.quota) .map_err(|error| AdminError::internal(error.to_string()))?; let raw = data.raw; - let quota_read = crate::admin_queries::accounts::AccountQuotaReadModel::from_snapshot( - data.quota, - Some(Utc::now()), - ); + let quota_read = AccountQuotaReadModel::from_snapshot(data.quota, Some(Utc::now())); let account = state .services .admin_accounts @@ -30,8 +27,7 @@ pub async fn account_quota( .ok_or_else(account_not_found)?; let item = state .services - .admin_queries - .accounts + .account_list .enrich_account(account, quota_read) .await .map_err(|error| AdminError::internal(error.to_string()))?; diff --git a/backend/src/api/admin/dashboard_routes.rs b/backend/src/api/admin/dashboard_routes.rs index e407c1149..777671be1 100644 --- a/backend/src/api/admin/dashboard_routes.rs +++ b/backend/src/api/admin/dashboard_routes.rs @@ -1,23 +1,21 @@ //! 管理端 Dashboard 聚合视图。 -use std::collections::HashMap; +use std::{collections::HashMap, sync::Arc}; use axum::{ extract::{Query, State}, http::StatusCode, response::IntoResponse, }; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Duration, Utc}; use serde::{Deserialize, Serialize}; use serde_json::Value; +use thiserror::Error; use crate::{ - admin_queries::dashboard::{ - DashboardAccountUsage, DashboardDesktopReleaseSnapshot, DashboardPoolSummary, - DashboardQueryError, DashboardWireProfile, - }, api::AppState, api::admin::{ + accounts_routes::query::RefreshActivityQuery, dashboard_presenter::{ DashboardAccountCounts, DashboardCardsData, DashboardHealthTimelineData, DashboardTrendData, DashboardTrendKind, dashboard_cards, @@ -30,12 +28,28 @@ use crate::{ usage_record_models, usage_token_details, }, }, - fleet::pool::AccountCapacitySummary, + fleet::{ + account::{Account, AccountStatus}, + pool::{AccountCapacitySummary, AccountPoolService}, + quota::QuotaSnapshot, + refresh::token_refresh_status_eligible, + store::AccountStore, + }, infra::{ format::{format_duration_ms, format_tokens}, - time::{china_datetime, china_relative_time}, + time::{china_datetime, china_day_start, china_quarter_hour_start, china_relative_time}, + }, + settings::service::SettingsService, + telemetry::{ + account_usage::query::{ + AccountUsageQueryService, AccountUsageTimeBucket, RetainedUsageSummary, + }, + usage::{ + insights::RequestHealthTimeBucket, + query::{UsageQueryFilter, UsageQueryService}, + types::{UsageRecord, metadata_i64, metadata_string}, + }, }, - telemetry::usage::types::{UsageRecord, metadata_i64, metadata_string}, }; #[derive(Debug, Deserialize)] @@ -212,7 +226,6 @@ pub(crate) async fn dashboard_summary( let now = Utc::now(); let dashboard = state .services - .admin_queries .dashboard .summary(now) .await @@ -253,7 +266,6 @@ pub(crate) async fn dashboard_trend( ) -> Result { let time_buckets = state .services - .admin_queries .dashboard .time_buckets(Utc::now()) .await @@ -420,3 +432,319 @@ fn wire_profile_data( }, } } + +const ACCOUNT_USAGE_LIMIT: u32 = 4; +const USAGE_RECORD_LIMIT: u32 = 10; +const TIME_BUCKET_MINUTES: i64 = 15; +const TIME_BUCKET_SLOTS: i64 = 7 * 24 * 4; + +/// Dashboard 消费的请求画像快照端口。 +pub(crate) trait DashboardWireProfileQuery: Send + Sync { + fn snapshot(&self) -> DashboardWireProfile; +} + +/// Dashboard 消费的 Desktop 发布观测端口。 +pub(crate) trait DashboardDesktopReleaseQuery: Send + Sync { + fn snapshot(&self) -> DashboardDesktopReleaseSnapshot; +} + +#[derive(Debug, Clone)] +pub(crate) struct DashboardWireProfile { + pub(crate) originator: String, + pub(crate) codex_version: String, + pub(crate) desktop_version: String, + pub(crate) desktop_build: String, + pub(crate) os_type: String, + pub(crate) os_version: String, + pub(crate) arch: String, + pub(crate) terminal: String, + pub(crate) user_agent: String, + pub(crate) verified_at: DateTime, +} + +#[derive(Debug, Clone, Default)] +pub(crate) struct DashboardDesktopReleaseSnapshot { + pub(crate) checked_at: Option>, + pub(crate) latest: Option, + pub(crate) last_error: Option, +} + +#[derive(Debug, Clone)] +pub(crate) struct DashboardDesktopRelease { + pub(crate) version: String, + pub(crate) build: String, + pub(crate) published_at: Option>, + pub(crate) minimum_system_version: Option, + pub(crate) hardware_requirements: Option, + pub(crate) download_url: Option, + pub(crate) download_size: Option, + pub(crate) signature_present: bool, +} + +#[derive(Debug, Clone)] +pub(crate) struct DashboardReadModel { + pub(crate) account_counts: DashboardQueryAccountCounts, + pub(crate) retained_usage: RetainedUsageSummary, + pub(crate) time_buckets: Vec, + pub(crate) health_buckets: Vec, + pub(crate) account_usage: Vec, + pub(crate) usage_records: Vec, + pub(crate) account_emails: HashMap, + pub(crate) pool: DashboardPoolSummary, + pub(crate) capacity: AccountCapacitySummary, + pub(crate) rotation_strategy: String, + pub(crate) wire_profile: DashboardWireProfile, + pub(crate) desktop_release: DashboardDesktopReleaseSnapshot, +} + +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct DashboardQueryAccountCounts { + pub(crate) total: u64, + pub(crate) enabled: u64, + pub(crate) abnormal: u64, +} + +#[derive(Debug, Clone)] +pub(crate) struct DashboardAccountUsage { + pub(crate) id: String, + pub(crate) email: String, + pub(crate) plan_type: Option, + pub(crate) total_tokens: u64, + pub(crate) quota_used_percent: Option, + pub(crate) last_used_at: DateTime, +} + +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct DashboardPoolSummary { + pub(crate) total: usize, + pub(crate) active: usize, + pub(crate) expired: usize, + pub(crate) quota_exhausted: usize, + pub(crate) refreshing: usize, + pub(crate) disabled: usize, + pub(crate) banned: usize, +} + +#[derive(Debug, Error)] +pub(crate) enum DashboardQueryError { + #[error("failed to load dashboard {0}")] + Data(&'static str), +} + +#[derive(Clone)] +pub(crate) struct DashboardQueryService { + accounts: Arc, + usage: Arc, + usage_records: Arc, + account_pool: Arc, + refresh_activity: Arc, + settings: Arc, + wire_profile: Arc, + desktop_release: Arc, +} + +pub(crate) struct DashboardQueryServiceParts { + pub(crate) accounts: Arc, + pub(crate) usage: Arc, + pub(crate) usage_records: Arc, + pub(crate) account_pool: Arc, + pub(crate) refresh_activity: Arc, + pub(crate) settings: Arc, + pub(crate) wire_profile: Arc, + pub(crate) desktop_release: Arc, +} + +impl DashboardQueryService { + pub(crate) fn new(parts: DashboardQueryServiceParts) -> Self { + Self { + accounts: parts.accounts, + usage: parts.usage, + usage_records: parts.usage_records, + account_pool: parts.account_pool, + refresh_activity: parts.refresh_activity, + settings: parts.settings, + wire_profile: parts.wire_profile, + desktop_release: parts.desktop_release, + } + } + + pub(crate) async fn summary( + &self, + now: DateTime, + ) -> Result { + let accounts = self + .accounts + .list_pool_accounts() + .await + .map_err(|_| DashboardQueryError::Data("accounts"))?; + let capacity = self.account_pool.capacity_summary_now().await; + let filter = UsageQueryFilter { + start_time: Some(china_day_start(now)), + end_time: Some(now), + ..UsageQueryFilter::default() + }; + let retained_usage = self + .usage + .retained_summary() + .await + .map_err(|_| DashboardQueryError::Data("retained usage summary"))?; + let account_usage_records = self + .usage_records + .account_usage(filter.clone(), ACCOUNT_USAGE_LIMIT) + .await + .map_err(|_| DashboardQueryError::Data("account usage ranking"))?; + let usage_records = self + .usage_records + .list_recent(USAGE_RECORD_LIMIT, filter) + .await + .map_err(|_| DashboardQueryError::Data("recent usage records"))?; + let time_buckets = self.time_buckets(now).await?; + let health_buckets = self.health_buckets(now).await?; + let account_ids = accounts + .iter() + .map(|account| account.id.clone()) + .collect::>(); + let refreshing = self + .refresh_activity + .refreshing_account_ids(&account_ids, now) + .await + .map_err(|_| DashboardQueryError::Data("refreshing accounts"))?; + let quota_used = self.quota_used_by_account(&account_usage_records).await?; + let account_emails = self + .usage_records + .account_email_map(&usage_records) + .await + .map_err(|_| DashboardQueryError::Data("usage record accounts"))?; + + Ok(DashboardReadModel { + account_counts: account_counts(&accounts), + retained_usage, + time_buckets, + health_buckets, + account_usage: account_usage(&accounts, &account_usage_records, "a_used), + usage_records, + account_emails, + pool: pool_summary(&accounts, &refreshing), + capacity, + rotation_strategy: self.settings.current().rotation_strategy, + wire_profile: self.wire_profile.snapshot(), + desktop_release: self.desktop_release.snapshot(), + }) + } + + pub(crate) async fn time_buckets( + &self, + now: DateTime, + ) -> Result, DashboardQueryError> { + let current_slot = china_quarter_hour_start(now); + let start = current_slot - Duration::minutes(TIME_BUCKET_MINUTES * (TIME_BUCKET_SLOTS - 1)); + self.usage + .time_buckets(start, now) + .await + .map_err(|_| DashboardQueryError::Data("time buckets")) + } + + async fn health_buckets( + &self, + now: DateTime, + ) -> Result, DashboardQueryError> { + self.usage_records + .health_timeline(china_day_start(now), now) + .await + .map_err(|_| DashboardQueryError::Data("health timeline")) + } + + async fn quota_used_by_account( + &self, + usage: &[crate::telemetry::usage::query::UsageRecordAccountUsage], + ) -> Result, DashboardQueryError> { + let mut quota_used = HashMap::with_capacity(usage.len()); + for usage in usage { + let quota_json = self + .accounts + .get_quota_json(&usage.account_id) + .await + .map_err(|_| DashboardQueryError::Data("account quota"))?; + let Some(quota) = quota_json + .as_deref() + .and_then(|value| QuotaSnapshot::from_json(value).ok()) + else { + continue; + }; + if let Some(used_percent) = quota.representative_used_percent() { + quota_used.insert(usage.account_id.clone(), used_percent); + } + } + Ok(quota_used) + } +} + +fn account_counts(accounts: &[Account]) -> DashboardQueryAccountCounts { + DashboardQueryAccountCounts { + total: accounts.len() as u64, + enabled: accounts + .iter() + .filter(|account| account.status == AccountStatus::Active) + .count() as u64, + abnormal: accounts + .iter() + .filter(|account| account.status != AccountStatus::Active) + .count() as u64, + } +} + +fn pool_summary( + accounts: &[Account], + refreshing: &std::collections::HashSet, +) -> DashboardPoolSummary { + let mut summary = DashboardPoolSummary { + total: accounts.len(), + ..DashboardPoolSummary::default() + }; + for account in accounts { + match account.status { + status if token_refresh_status_eligible(status) && refreshing.contains(&account.id) => { + summary.refreshing += 1; + } + AccountStatus::Active => summary.active += 1, + AccountStatus::Expired => summary.expired += 1, + AccountStatus::QuotaExhausted => summary.quota_exhausted += 1, + AccountStatus::Disabled => summary.disabled += 1, + AccountStatus::Banned => summary.banned += 1, + } + } + summary +} + +fn account_usage( + accounts: &[Account], + usage_records: &[crate::telemetry::usage::query::UsageRecordAccountUsage], + quota_used: &HashMap, +) -> Vec { + let accounts = accounts + .iter() + .map(|account| (account.id.as_str(), account)) + .collect::>(); + usage_records + .iter() + .map(|usage| { + let account = accounts.get(usage.account_id.as_str()).copied(); + DashboardAccountUsage { + id: usage.account_id.clone(), + email: account + .and_then(|account| account.email.clone()) + .unwrap_or_else(|| usage.account_id.clone()), + plan_type: account.and_then(|account| account.plan_type.clone()), + total_tokens: usage.total_tokens, + quota_used_percent: quota_used.get(&usage.account_id).copied().or_else(|| { + matches!( + account.map(|account| account.status), + Some(AccountStatus::QuotaExhausted) + ) + .then_some(100.0) + }), + last_used_at: usage.last_used_at, + } + }) + .collect() +} diff --git a/backend/src/api/router.rs b/backend/src/api/router.rs index 02bd755ed..fb45aa13c 100644 --- a/backend/src/api/router.rs +++ b/backend/src/api/router.rs @@ -9,9 +9,12 @@ use async_trait::async_trait; use axum::{Router, extract::State, http::StatusCode, middleware, routing::get}; use crate::{ - admin_queries::{accounts::AccountListQueryService, dashboard::DashboardQueryService}, api::{ - admin, assets, client, + admin::{ + self, accounts_routes::query::AccountListQueryService, + dashboard_routes::DashboardQueryService, + }, + assets, client, middleware::{request_id::attach_request_id, trace::http_trace_layer}, }, auth::service::SessionService, @@ -24,16 +27,11 @@ use crate::{ update::service::SystemUpdateService, }; -#[derive(Clone)] -pub struct AdminQueries { - pub accounts: Arc, - pub dashboard: Arc, -} - /// HTTP API 所需的领域服务集合。 #[derive(Clone)] pub struct ApiServices { - pub admin_queries: AdminQueries, + pub(crate) account_list: Arc, + pub(crate) dashboard: Arc, pub health_probe: Arc, pub models: Arc, pub client_keys: Arc, diff --git a/backend/src/bootstrap/services.rs b/backend/src/bootstrap/services.rs index fd0a6fdb1..d73957923 100644 --- a/backend/src/bootstrap/services.rs +++ b/backend/src/bootstrap/services.rs @@ -6,6 +6,14 @@ use axum::Router; use secrecy::ExposeSecret; use crate::{ + api::admin::{ + accounts_routes::query::{AccountListQueryService, RefreshActivityQuery}, + dashboard_routes::{ + DashboardDesktopRelease, DashboardDesktopReleaseQuery, DashboardDesktopReleaseSnapshot, + DashboardQueryService, DashboardQueryServiceParts, DashboardWireProfile, + DashboardWireProfileQuery, + }, + }, auth::{ service::SessionService, store::{PgAdminUserStore, RedisAdminSessionStore}, @@ -147,11 +155,11 @@ impl UsageRecordOptions { } } -impl crate::admin_queries::dashboard::DashboardWireProfileQuery for CodexWireProfileState { - fn snapshot(&self) -> crate::admin_queries::dashboard::DashboardWireProfile { +impl DashboardWireProfileQuery for CodexWireProfileState { + fn snapshot(&self) -> DashboardWireProfile { let profile = CodexWireProfileState::snapshot(self); let user_agent = profile.user_agent(); - crate::admin_queries::dashboard::DashboardWireProfile { + DashboardWireProfile { originator: profile.originator, codex_version: profile.codex_version, desktop_version: profile.desktop_version, @@ -166,22 +174,20 @@ impl crate::admin_queries::dashboard::DashboardWireProfileQuery for CodexWirePro } } -impl crate::admin_queries::dashboard::DashboardDesktopReleaseQuery for DesktopReleaseStatus { - fn snapshot(&self) -> crate::admin_queries::dashboard::DashboardDesktopReleaseSnapshot { +impl DashboardDesktopReleaseQuery for DesktopReleaseStatus { + fn snapshot(&self) -> DashboardDesktopReleaseSnapshot { let snapshot = DesktopReleaseStatus::snapshot(self); - crate::admin_queries::dashboard::DashboardDesktopReleaseSnapshot { + DashboardDesktopReleaseSnapshot { checked_at: snapshot.checked_at, - latest: snapshot.latest.map(|release| { - crate::admin_queries::dashboard::DashboardDesktopRelease { - version: release.version, - build: release.build, - published_at: release.published_at, - minimum_system_version: release.minimum_system_version, - hardware_requirements: release.hardware_requirements, - download_url: release.download_url, - download_size: release.download_size, - signature_present: release.signature_present, - } + latest: snapshot.latest.map(|release| DashboardDesktopRelease { + version: release.version, + build: release.build, + published_at: release.published_at, + minimum_system_version: release.minimum_system_version, + hardware_requirements: release.hardware_requirements, + download_url: release.download_url, + download_size: release.download_size, + signature_present: release.signature_present, }), last_error: snapshot.last_error, } @@ -403,32 +409,25 @@ impl Services { impl From<&Services> for crate::api::router::ApiServices { fn from(services: &Services) -> Self { - let refresh_activity = services.token_refresh.clone() - as Arc; - let account_list = Arc::new( - crate::admin_queries::accounts::AccountListQueryService::new( - services.admin_accounts.clone(), - services.usage.clone(), - refresh_activity.clone(), - ), - ); - let dashboard = Arc::new(crate::admin_queries::dashboard::DashboardQueryService::new( - crate::admin_queries::dashboard::DashboardQueryServiceParts { - accounts: services.accounts.clone(), - usage: services.usage.clone(), - usage_records: services.usage_records.clone(), - account_pool: services.account_pool.clone(), - refresh_activity, - settings: services.settings.clone(), - wire_profile: Arc::new(services.wire_profile.clone()), - desktop_release: Arc::new(services.desktop_release.clone()), - }, + let refresh_activity = services.token_refresh.clone() as Arc; + let account_list = Arc::new(AccountListQueryService::new( + services.admin_accounts.clone(), + services.usage.clone(), + refresh_activity.clone(), )); + let dashboard = Arc::new(DashboardQueryService::new(DashboardQueryServiceParts { + accounts: services.accounts.clone(), + usage: services.usage.clone(), + usage_records: services.usage_records.clone(), + account_pool: services.account_pool.clone(), + refresh_activity, + settings: services.settings.clone(), + wire_profile: Arc::new(services.wire_profile.clone()), + desktop_release: Arc::new(services.desktop_release.clone()), + })); Self { - admin_queries: crate::api::router::AdminQueries { - accounts: account_list, - dashboard, - }, + account_list, + dashboard, health_probe: Arc::new(crate::bootstrap::state::RuntimeHealthProbe::new( services.database.clone(), services.redis.clone(), diff --git a/backend/src/lib.rs b/backend/src/lib.rs index 28f9cbd84..dac846127 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -1,4 +1,3 @@ -pub mod admin_queries; pub mod api; pub mod auth; pub mod bootstrap; diff --git a/backend/tests/architecture/dependencies.rs b/backend/tests/architecture/dependencies.rs index 9d41c6141..473a29e55 100644 --- a/backend/tests/architecture/dependencies.rs +++ b/backend/tests/architecture/dependencies.rs @@ -158,15 +158,8 @@ fn forbidden_dependency(source: &[String], dependency: &[String]) -> Option<&'st return Some("models owns ModelCatalogSource; upstream is only its adapter"); } - if source_owner == "admin_queries" && matches!(dependency_owner, Some("api" | "upstream")) { - return Some("admin query services cannot depend on transport or API presenters"); - } - - if source_owner == "api" - && (matches!(dependency_owner, Some("sqlx" | "redis")) - || dependency.iter().any(|segment| segment == "store")) - { - return Some("API routes and presenters cannot access persistence adapters directly"); + if source_owner == "api" && matches!(dependency_owner, Some("sqlx" | "redis")) { + return Some("API code cannot access database clients directly"); } None diff --git a/docs/architecture-audit.md b/docs/architecture-audit.md index df86f145f..9affa8b67 100644 --- a/docs/architecture-audit.md +++ b/docs/architecture-audit.md @@ -221,7 +221,7 @@ dispatch error 同时包装 transport/SSE failure、决定 HTTP status 和 camel 建议按入口组合： - `ProxyRuntime` -- `AdminQueries` +- 管理端查询依赖按功能模块就地组织 - `AdminCommands` - `BackgroundTasks` diff --git a/docs/architecture.md b/docs/architecture.md index 97806bd3c..0d03795b4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -64,7 +64,6 @@ backend/src/ ├── main.rs ├── lib.rs ├── infra/ -├── admin_queries/ ├── upstream/ ├── fleet/ ├── models/ @@ -84,8 +83,7 @@ backend/src/ | --- | --- | | `main` | 解析子命令，只进入 `bootstrap` 或一次性任务 | | `bootstrap` | 进程 composition root，可装配所有领域模块，不拥有业务规则 | -| `api` | HTTP/WebSocket 契约、鉴权提取和最终响应编码，不写 SQL/Redis | -| `admin_queries` | 管理端按功能组织的只读用例与消费方端口，不依赖 API presenter 或 upstream adapter | +| `api` | HTTP/WebSocket 契约、鉴权提取、管理端功能查询和最终响应编码，不直接使用 SQL/Redis 客户端 | | `dispatch` | 编排一次 `v1/*` 请求，调用 `fleet`、`models`、`upstream`、`telemetry` | | `fleet` | 账号、调度、quota、token、Cookie 和管理操作 | | `upstream` | Codex 协议与 HTTP/WebSocket transport，不选择账号 | @@ -197,7 +195,7 @@ upstream/openai/ `bootstrap` 负责配置、依赖装配、启动任务和关闭协调；它不解释账号错误、history 或 transport outcome。 -Dashboard 和账号列表由 `admin_queries` 完成跨域读取，route 只解析参数，presenter 只负责 HTTP casing、显示文本和格式化。上游请求画像与 Desktop 发布观测通过 `admin_queries` 定义的快照端口注入，query service 不引用 OpenAI adapter 类型。 +Dashboard 查询归入 `api/admin/dashboard_routes.rs`，账号列表查询归入 `api/admin/accounts_routes/query.rs`。HTTP handler 负责参数、鉴权和响应，文件内 query service 负责跨域读取，presenter 负责 casing、显示文本和格式化。上游请求画像与 Desktop 发布观测通过 Dashboard 模块定义的快照端口注入，不直接引用 OpenAI adapter 类型。 ## v1 请求生命周期