starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-07 00:05:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(backend): simplify admin query structure

Browse Source 
Move account and dashboard read orchestration into their existing admin API modules, and remove the top-level admin_queries package and service wrapper.

Update composition wiring, dependency guards, and architecture documentation to reflect feature-local query ownership.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-07-17 13:09:17 +08:00
co-authored by Codex
parent e40bcdc690
commit dc7afb06a2
14 changed files with 985 additions and 1026 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/admin_queries/accounts.rs
-589
View File
@@ -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<String>,
pub status: Option<AccountStatus>,
pub sort: Option<AccountListSort>,
}
#[derive(Debug, Clone)]
pub struct AccountListResult {
pub page: NumberedPage<AccountListItem>,
pub summary: AccountSummary,
}
#[derive(Debug, Clone)]
pub struct AccountListItem {
pub account: ManagedAccount,
pub usage: Option<AccountUsageRecord>,
pub quota: Option<AccountQuotaReadModel>,
pub models: Vec<AccountModelUsage>,
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<DateTime<Utc>>,
pub windows: Vec<AccountQuotaWindowReadModel>,
}
#[derive(Debug, Clone)]
pub struct AccountQuotaWindowReadModel {
pub quota: ClassifiedQuotaWindow,
pub local_usage: Option<AccountQuotaWindowLocalUsage>,
}
#[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<Utc>,
pub end: DateTime<Utc>,
pub window_seconds: u64,
}
impl AccountQuotaReadModel {
pub fn from_snapshot(snapshot: QuotaSnapshot, fetched_at: Option<DateTime<Utc>>) -> Self {
Self {
fetched_at,
windows: snapshot
.classified_windows()
.into_iter()
.map(|quota| AccountQuotaWindowReadModel {
quota,
local_usage: None,
})
.collect(),
}
}
pub fn usage_windows(&self) -> Vec<AccountQuotaUsageWindow> {
self.windows
.iter()
.filter_map(AccountQuotaWindowReadModel::usage_window)
.collect()
}
fn apply_local_usage(
&mut self,
usage_by_window: &HashMap<String, AccountQuotaWindowLocalUsage>,
) {
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<AccountQuotaUsageWindow> {
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<DateTime<Utc>>,
}
#[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<Utc>,
) -> Result<std::collections::HashSet<String>, AccountListQueryError>;
}
#[async_trait]
impl<R> RefreshActivityQuery for TokenRefreshService<R>
where
R: TokenRefresher,
{
async fn refreshing_account_ids(
&self,
account_ids: &[String],
now: DateTime<Utc>,
) -> Result<std::collections::HashSet<String>, AccountListQueryError> {
TokenRefreshService::refreshing_account_ids(self, account_ids, now)
.await
.map_err(|_| AccountListQueryError::RefreshActivity)
}
}
#[derive(Clone)]
pub struct AccountListQueryService {
accounts: Arc<AccountManageService>,
usage: Arc<AccountUsageQueryService>,
refresh_activity: Arc<dyn RefreshActivityQuery>,
}
impl AccountListQueryService {
pub fn new(
accounts: Arc<AccountManageService>,
usage: Arc<AccountUsageQueryService>,
refresh_activity: Arc<dyn RefreshActivityQuery>,
) -> Self {
Self {
accounts,
usage,
refresh_activity,
}
}
pub async fn query(
&self,
query: AccountListQuery,
) -> Result<AccountListResult, AccountListQueryError> {
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::<Vec<_>>();
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, &quota_by_account);
let quota_usage = self.quota_local_usage(&quota_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, &quota_by_account);
let usage_by_account = usage_records
.into_iter()
.map(|usage| {
let usage = apply_selected_window_usage(usage, &selected_windows, &quota_usage);
(usage.account_id.clone(), usage)
})
.collect::<HashMap<_, _>>();
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<AccountListItem, AccountListQueryError> {
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), &quota_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), &quota_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, &quota_usage));
let usage_by_account = usage
.iter()
.cloned()
.map(|usage| (usage.account_id.clone(), usage))
.collect::<HashMap<_, _>>();
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<Vec<ManagedAccount>, 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<HashMap<String, AccountQuotaReadModel>, 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<HashMap<String, HashMap<String, AccountQuotaWindowLocalUsage>>, 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::<Vec<_>>();
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<String, AccountUsageRecord>,
) -> Result<HashMap<String, Vec<AccountModelUsage>>, 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::<Vec<_>>();
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::<String, Vec<AccountModelUsage>>::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<String, AccountQuotaReadModel>,
) -> Vec<AccountQuotaWindowSelection> {
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<String, AccountQuotaReadModel>,
) -> HashMap<String, AccountQuotaUsageWindow> {
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<String, AccountQuotaUsageWindow>,
quota_usage: &HashMap<String, HashMap<String, AccountQuotaWindowLocalUsage>>,
) -> 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<Utc>,
) -> Option<(DateTime<Utc>, DateTime<Utc>)> {
let start = usage.window_started_at?;
let end = usage.window_reset_at.unwrap_or(now);
(start <= end).then_some((start, end))
}
backend/src/admin_queries/dashboard.rs
-345
View File
@@ -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<Utc>,
}
#[derive(Debug, Clone, Default)]
pub struct DashboardDesktopReleaseSnapshot {
pub checked_at: Option<DateTime<Utc>>,
pub latest: Option<DashboardDesktopRelease>,
pub last_error: Option<String>,
}
#[derive(Debug, Clone)]
pub struct DashboardDesktopRelease {
pub version: String,
pub build: String,
pub published_at: Option<DateTime<Utc>>,
pub minimum_system_version: Option<String>,
pub hardware_requirements: Option<String>,
pub download_url: Option<String>,
pub download_size: Option<u64>,
pub signature_present: bool,
}
#[derive(Debug, Clone)]
pub struct DashboardReadModel {
pub account_counts: DashboardAccountCounts,
pub retained_usage: RetainedUsageSummary,
pub time_buckets: Vec<AccountUsageTimeBucket>,
pub health_buckets: Vec<RequestHealthTimeBucket>,
pub account_usage: Vec<DashboardAccountUsage>,
pub usage_records: Vec<UsageRecord>,
pub account_emails: HashMap<String, String>,
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<String>,
pub total_tokens: u64,
pub quota_used_percent: Option<f64>,
pub last_used_at: DateTime<Utc>,
}
#[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<dyn AccountStore>,
usage: Arc<AccountUsageQueryService>,
usage_records: Arc<UsageQueryService>,
account_pool: Arc<AccountPoolService>,
refresh_activity: Arc<dyn RefreshActivityQuery>,
settings: Arc<SettingsService>,
wire_profile: Arc<dyn DashboardWireProfileQuery>,
desktop_release: Arc<dyn DashboardDesktopReleaseQuery>,
}
pub struct DashboardQueryServiceParts {
pub accounts: Arc<dyn AccountStore>,
pub usage: Arc<AccountUsageQueryService>,
pub usage_records: Arc<UsageQueryService>,
pub account_pool: Arc<AccountPoolService>,
pub refresh_activity: Arc<dyn RefreshActivityQuery>,
pub settings: Arc<SettingsService>,
pub wire_profile: Arc<dyn DashboardWireProfileQuery>,
pub desktop_release: Arc<dyn DashboardDesktopReleaseQuery>,
}
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<Utc>,
) -> Result<DashboardReadModel, DashboardQueryError> {
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::<Vec<_>>();
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, &quota_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<Utc>,
) -> Result<Vec<AccountUsageTimeBucket>, 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<Utc>,
) -> Result<Vec<RequestHealthTimeBucket>, 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<HashMap<String, f64>, 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<String>,
) -> 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<String, f64>,
) -> Vec<DashboardAccountUsage> {
let accounts = accounts
.iter()
.map(|account| (account.id.as_str(), account))
.collect::<HashMap<_, _>>();
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()
}
backend/src/admin_queries/mod.rs
-4
View File
@@ -1,4 +0,0 @@
//! 管理端跨域只读用例。
pub mod accounts;
pub mod dashboard;
backend/src/api/admin/accounts_routes/mod.rs
+2 -4
View File
@@ -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;
backend/src/api/admin/accounts_routes/query.rs
+589 -1
View File
@@ -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<String>,
pub(crate) status: Option<AccountStatus>,
pub(crate) sort: Option<AccountListSort>,
}
#[derive(Debug, Clone)]
pub(crate) struct AccountListResult {
pub(crate) page: NumberedPage<AccountListItem>,
pub(crate) summary: AccountSummary,
}
#[derive(Debug, Clone)]
pub(crate) struct AccountListItem {
pub(crate) account: ManagedAccount,
pub(crate) usage: Option<AccountUsageRecord>,
pub(crate) quota: Option<AccountQuotaReadModel>,
pub(crate) models: Vec<AccountModelUsage>,
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<DateTime<Utc>>,
pub(crate) windows: Vec<AccountQuotaWindowReadModel>,
}
#[derive(Debug, Clone)]
pub(crate) struct AccountQuotaWindowReadModel {
pub(crate) quota: ClassifiedQuotaWindow,
pub(crate) local_usage: Option<AccountQuotaWindowLocalUsage>,
}
#[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<Utc>,
pub(crate) end: DateTime<Utc>,
pub(crate) window_seconds: u64,
}
impl AccountQuotaReadModel {
pub(crate) fn from_snapshot(
snapshot: QuotaSnapshot,
fetched_at: Option<DateTime<Utc>>,
) -> 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<AccountQuotaUsageWindow> {
self.windows
.iter()
.filter_map(AccountQuotaWindowReadModel::usage_window)
.collect()
}
fn apply_local_usage(
&mut self,
usage_by_window: &HashMap<String, AccountQuotaWindowLocalUsage>,
) {
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<AccountQuotaUsageWindow> {
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<DateTime<Utc>>,
}
#[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<Utc>,
) -> Result<std::collections::HashSet<String>, AccountListQueryError>;
}
#[async_trait]
impl<R> RefreshActivityQuery for TokenRefreshService<R>
where
R: TokenRefresher,
{
async fn refreshing_account_ids(
&self,
account_ids: &[String],
now: DateTime<Utc>,
) -> Result<std::collections::HashSet<String>, AccountListQueryError> {
TokenRefreshService::refreshing_account_ids(self, account_ids, now)
.await
.map_err(|_| AccountListQueryError::RefreshActivity)
}
}
#[derive(Clone)]
pub(crate) struct AccountListQueryService {
accounts: Arc<AccountManageService>,
usage: Arc<AccountUsageQueryService>,
refresh_activity: Arc<dyn RefreshActivityQuery>,
}
impl AccountListQueryService {
pub(crate) fn new(
accounts: Arc<AccountManageService>,
usage: Arc<AccountUsageQueryService>,
refresh_activity: Arc<dyn RefreshActivityQuery>,
) -> Self {
Self {
accounts,
usage,
refresh_activity,
}
}
pub(crate) async fn query(
&self,
query: AccountListQuery,
) -> Result<AccountListResult, AccountListQueryError> {
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::<Vec<_>>();
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, &quota_by_account);
let quota_usage = self.quota_local_usage(&quota_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, &quota_by_account);
let usage_by_account = usage_records
.into_iter()
.map(|usage| {
let usage = apply_selected_window_usage(usage, &selected_windows, &quota_usage);
(usage.account_id.clone(), usage)
})
.collect::<HashMap<_, _>>();
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<AccountListItem, AccountListQueryError> {
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), &quota_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), &quota_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, &quota_usage));
let usage_by_account = usage
.iter()
.cloned()
.map(|usage| (usage.account_id.clone(), usage))
.collect::<HashMap<_, _>>();
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<Vec<ManagedAccount>, 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<HashMap<String, AccountQuotaReadModel>, 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<HashMap<String, HashMap<String, AccountQuotaWindowLocalUsage>>, 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::<Vec<_>>();
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<String, AccountUsageRecord>,
) -> Result<HashMap<String, Vec<AccountModelUsage>>, 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::<Vec<_>>();
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::<String, Vec<AccountModelUsage>>::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<String, AccountQuotaReadModel>,
) -> Vec<AccountQuotaWindowSelection> {
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<String, AccountQuotaReadModel>,
) -> HashMap<String, AccountQuotaUsageWindow> {
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<String, AccountQuotaUsageWindow>,
quota_usage: &HashMap<String, HashMap<String, AccountQuotaWindowLocalUsage>>,
) -> 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<Utc>,
) -> Option<(DateTime<Utc>, DateTime<Utc>)> {
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,
backend/src/api/admin/accounts_routes/quota_presenter.rs
+1 -1
View File
@@ -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::{
backend/src/api/admin/accounts_routes/quota_routes.rs
+2 -6
View File
@@ -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()))?;
backend/src/api/admin/dashboard_routes.rs
+339 -11
View File
@@ -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<impl IntoResponse, AdminError> {
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<Utc>,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct DashboardDesktopReleaseSnapshot {
pub(crate) checked_at: Option<DateTime<Utc>>,
pub(crate) latest: Option<DashboardDesktopRelease>,
pub(crate) last_error: Option<String>,
}
#[derive(Debug, Clone)]
pub(crate) struct DashboardDesktopRelease {
pub(crate) version: String,
pub(crate) build: String,
pub(crate) published_at: Option<DateTime<Utc>>,
pub(crate) minimum_system_version: Option<String>,
pub(crate) hardware_requirements: Option<String>,
pub(crate) download_url: Option<String>,
pub(crate) download_size: Option<u64>,
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<AccountUsageTimeBucket>,
pub(crate) health_buckets: Vec<RequestHealthTimeBucket>,
pub(crate) account_usage: Vec<DashboardAccountUsage>,
pub(crate) usage_records: Vec<UsageRecord>,
pub(crate) account_emails: HashMap<String, String>,
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<String>,
pub(crate) total_tokens: u64,
pub(crate) quota_used_percent: Option<f64>,
pub(crate) last_used_at: DateTime<Utc>,
}
#[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<dyn AccountStore>,
usage: Arc<AccountUsageQueryService>,
usage_records: Arc<UsageQueryService>,
account_pool: Arc<AccountPoolService>,
refresh_activity: Arc<dyn RefreshActivityQuery>,
settings: Arc<SettingsService>,
wire_profile: Arc<dyn DashboardWireProfileQuery>,
desktop_release: Arc<dyn DashboardDesktopReleaseQuery>,
}
pub(crate) struct DashboardQueryServiceParts {
pub(crate) accounts: Arc<dyn AccountStore>,
pub(crate) usage: Arc<AccountUsageQueryService>,
pub(crate) usage_records: Arc<UsageQueryService>,
pub(crate) account_pool: Arc<AccountPoolService>,
pub(crate) refresh_activity: Arc<dyn RefreshActivityQuery>,
pub(crate) settings: Arc<SettingsService>,
pub(crate) wire_profile: Arc<dyn DashboardWireProfileQuery>,
pub(crate) desktop_release: Arc<dyn DashboardDesktopReleaseQuery>,
}
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<Utc>,
) -> Result<DashboardReadModel, DashboardQueryError> {
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::<Vec<_>>();
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, &quota_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<Utc>,
) -> Result<Vec<AccountUsageTimeBucket>, 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<Utc>,
) -> Result<Vec<RequestHealthTimeBucket>, 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<HashMap<String, f64>, 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<String>,
) -> 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<String, f64>,
) -> Vec<DashboardAccountUsage> {
let accounts = accounts
.iter()
.map(|account| (account.id.as_str(), account))
.collect::<HashMap<_, _>>();
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()
}
backend/src/api/router.rs
+7 -9
View File
@@ -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<AccountListQueryService>,
pub dashboard: Arc<DashboardQueryService>,
}
/// HTTP API 所需的领域服务集合。
#[derive(Clone)]
pub struct ApiServices {
pub admin_queries: AdminQueries,
pub(crate) account_list: Arc<AccountListQueryService>,
pub(crate) dashboard: Arc<DashboardQueryService>,
pub health_probe: Arc<dyn HealthProbe>,
pub models: Arc<ModelService>,
pub client_keys: Arc<KeyVerifier>,
backend/src/bootstrap/services.rs
+40 -41
View File
@@ -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<dyn crate::admin_queries::accounts::RefreshActivityQuery>;
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<dyn RefreshActivityQuery>;
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(),
backend/src/lib.rs
-1
View File
@@ -1,4 +1,3 @@
pub mod admin_queries;
pub mod api;
pub mod auth;
pub mod bootstrap;
backend/tests/architecture/dependencies.rs
+2 -9
View File
@@ -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
docs/architecture-audit.md
+1 -1
View File
@@ -221,7 +221,7 @@ dispatch error 同时包装 transport/SSE failure、决定 HTTP status 和 camel
建议按入口组合：
- `ProxyRuntime`
- `AdminQueries`
- 管理端查询依赖按功能模块就地组织
- `AdminCommands`
- `BackgroundTasks`
docs/architecture.md
+2 -4
View File
@@ -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 请求生命周期