i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 09:20:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(gateway/data): 新增 usage 关键词搜索、缓存命中摘要、结算成本摘要及 dashboard 聚合查询能力

Browse Source 
- 新增 UsageAuditKeywordSearchQuery，支持多关键词、用户名、API Key 交叉过滤
- 新增 UsageCacheHitSummaryQuery/StoredUsageCacheHitSummary，统计请求缓存命中率
- 新增 UsageSettledCostSummaryQuery/StoredUsageSettledCostSummary，汇总结算成本
- 新增 UsageDashboardSummaryQuery、UsageBreakdownSummaryQuery 等 dashboard 聚合类型
- SQL 层实现对应查询方法，含 list_by_ids、list_usage_audits_by_keyword_search、count_usage_audits_by_keyword_search
- 将上述能力通过 GatewayDataState / AdminAppState 暴露给 handler 层
- user_me_usage 及 dashboard_filters 切换为新查询接口，移除旧的内存过滤逻辑
- 同步更新 memory 层及测试
This commit is contained in:
fawney19
2026-04-17 00:58:46 +08:00
parent 6e5af5ef70
commit 1e0bc61526
48 changed files with 5241 additions and 1162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
apps/aether-gateway/src/admin_api.rs
View File

@@ -7,7 +7,6 @@ pub(crate) use crate::handlers::admin::{
use crate::handlers::admin::{
admin_stats_bad_request_response as admin_stats_bad_request_response_impl,
list_usage_for_optional_range as list_usage_for_optional_range_impl,
parse_bounded_u32 as parse_bounded_u32_impl, round_to as round_to_impl,
};
use crate::GatewayError;
@@ -61,14 +60,6 @@ pub(crate) fn admin_stats_bad_request_response(detail: String) -> Response<Body>
admin_stats_bad_request_response_impl(detail)
}
pub(crate) async fn list_usage_for_optional_range(
state: &AdminAppState<'_>,
time_range: Option<&AdminStatsTimeRange>,
filters: &AdminStatsUsageFilter,
) -> Result<Vec<aether_data_contracts::repository::usage::StoredRequestUsageAudit>, GatewayError> {
list_usage_for_optional_range_impl(state, time_range, filters).await
}
pub(crate) fn parse_bounded_u32(
field: &str,
value: &str,

14
apps/aether-gateway/src/data/state/auth.rs
View File

@@ -2425,6 +2425,20 @@ impl GatewayDataState {
}
}
pub(crate) async fn list_auth_api_key_export_records_by_name_search(
&self,
name_search: &str,
) -> Result<Vec<StoredAuthApiKeyExportRecord>, DataLayerError> {
match &self.auth_api_key_reader {
Some(repository) => {
repository
.list_export_api_keys_by_name_search(name_search)
.await
}
None => Ok(Vec::new()),
}
}
pub(crate) async fn list_auth_api_key_export_standalone_records_page(
&self,
query: &aether_data::repository::auth::StandaloneApiKeyExportListQuery,

219
apps/aether-gateway/src/data/state/runtime.rs
View File

@@ -646,6 +646,16 @@ impl GatewayDataState {
}
}
pub(crate) async fn list_request_usage_by_ids(
&self,
usage_ids: &[String],
) -> Result<Vec<StoredRequestUsageAudit>, DataLayerError> {
match &self.usage_reader {
Some(repository) => repository.list_by_ids(usage_ids).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn resolve_request_usage_body_ref(
&self,
body_ref: &str,
@@ -676,6 +686,26 @@ impl GatewayDataState {
}
}
pub(crate) async fn list_usage_audits_by_keyword_search(
&self,
query: &aether_data_contracts::repository::usage::UsageAuditKeywordSearchQuery,
) -> Result<Vec<StoredRequestUsageAudit>, DataLayerError> {
match &self.usage_reader {
Some(repository) => repository.list_usage_audits_by_keyword_search(query).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn count_usage_audits_by_keyword_search(
&self,
query: &aether_data_contracts::repository::usage::UsageAuditKeywordSearchQuery,
) -> Result<u64, DataLayerError> {
match &self.usage_reader {
Some(repository) => repository.count_usage_audits_by_keyword_search(query).await,
None => Ok(0),
}
}
pub(crate) async fn aggregate_usage_audits(
&self,
query: &aether_data_contracts::repository::usage::UsageAuditAggregationQuery,
@@ -702,6 +732,181 @@ impl GatewayDataState {
}
}
pub(crate) async fn summarize_usage_cache_hit_summary(
&self,
query: &aether_data_contracts::repository::usage::UsageCacheHitSummaryQuery,
) -> Result<aether_data_contracts::repository::usage::StoredUsageCacheHitSummary, DataLayerError>
{
match &self.usage_reader {
Some(repository) => repository.summarize_usage_cache_hit_summary(query).await,
None => {
Ok(aether_data_contracts::repository::usage::StoredUsageCacheHitSummary::default())
}
}
}
pub(crate) async fn summarize_usage_settled_cost(
&self,
query: &aether_data_contracts::repository::usage::UsageSettledCostSummaryQuery,
) -> Result<
aether_data_contracts::repository::usage::StoredUsageSettledCostSummary,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository.summarize_usage_settled_cost(query).await,
None => Ok(
aether_data_contracts::repository::usage::StoredUsageSettledCostSummary::default(),
),
}
}
pub(crate) async fn summarize_usage_cache_affinity_hit_summary(
&self,
query: &aether_data_contracts::repository::usage::UsageCacheAffinityHitSummaryQuery,
) -> Result<
aether_data_contracts::repository::usage::StoredUsageCacheAffinityHitSummary,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository
.summarize_usage_cache_affinity_hit_summary(query)
.await,
None => Ok(
aether_data_contracts::repository::usage::StoredUsageCacheAffinityHitSummary::default(),
),
}
}
pub(crate) async fn list_usage_cache_affinity_intervals(
&self,
query: &aether_data_contracts::repository::usage::UsageCacheAffinityIntervalQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsageCacheAffinityIntervalRow>,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository.list_usage_cache_affinity_intervals(query).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn summarize_dashboard_usage(
&self,
query: &aether_data_contracts::repository::usage::UsageDashboardSummaryQuery,
) -> Result<aether_data_contracts::repository::usage::StoredUsageDashboardSummary, DataLayerError>
{
match &self.usage_reader {
Some(repository) => repository.summarize_dashboard_usage(query).await,
None => Ok(
aether_data_contracts::repository::usage::StoredUsageDashboardSummary::default(),
),
}
}
pub(crate) async fn list_dashboard_daily_breakdown(
&self,
query: &aether_data_contracts::repository::usage::UsageDashboardDailyBreakdownQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsageDashboardDailyBreakdownRow>,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository.list_dashboard_daily_breakdown(query).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn summarize_dashboard_provider_counts(
&self,
query: &aether_data_contracts::repository::usage::UsageDashboardProviderCountsQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsageDashboardProviderCount>,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository.summarize_dashboard_provider_counts(query).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn summarize_usage_breakdown(
&self,
query: &aether_data_contracts::repository::usage::UsageBreakdownSummaryQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsageBreakdownSummaryRow>,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository.summarize_usage_breakdown(query).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn count_monitoring_usage_errors(
&self,
query: &aether_data_contracts::repository::usage::UsageMonitoringErrorCountQuery,
) -> Result<u64, DataLayerError> {
match &self.usage_reader {
Some(repository) => repository.count_monitoring_usage_errors(query).await,
None => Ok(0),
}
}
pub(crate) async fn list_monitoring_usage_errors(
&self,
query: &aether_data_contracts::repository::usage::UsageMonitoringErrorListQuery,
) -> Result<Vec<StoredRequestUsageAudit>, DataLayerError> {
match &self.usage_reader {
Some(repository) => repository.list_monitoring_usage_errors(query).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn summarize_usage_error_distribution(
&self,
query: &aether_data_contracts::repository::usage::UsageErrorDistributionQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsageErrorDistributionRow>,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository.summarize_usage_error_distribution(query).await,
None => Ok(Vec::new()),
}
}
pub(crate) async fn summarize_usage_performance_percentiles(
&self,
query: &aether_data_contracts::repository::usage::UsagePerformancePercentilesQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsagePerformancePercentilesRow>,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => {
repository
.summarize_usage_performance_percentiles(query)
.await
}
None => Ok(Vec::new()),
}
}
pub(crate) async fn summarize_usage_cost_savings(
&self,
query: &aether_data_contracts::repository::usage::UsageCostSavingsSummaryQuery,
) -> Result<
aether_data_contracts::repository::usage::StoredUsageCostSavingsSummary,
DataLayerError,
> {
match &self.usage_reader {
Some(repository) => repository.summarize_usage_cost_savings(query).await,
None => Ok(
aether_data_contracts::repository::usage::StoredUsageCostSavingsSummary::default(),
),
}
}
pub(crate) async fn summarize_usage_time_series(
&self,
query: &aether_data_contracts::repository::usage::UsageTimeSeriesQuery,
@@ -793,6 +998,20 @@ impl GatewayDataState {
}
}
pub(crate) async fn list_users_by_username_search(
&self,
username_search: &str,
) -> Result<Vec<StoredUserSummary>, DataLayerError> {
match &self.user_reader {
Some(repository) => {
repository
.list_users_by_username_search(username_search)
.await
}
None => Ok(Vec::new()),
}
}
pub(crate) async fn list_export_users(
&self,
) -> Result<Vec<StoredUserExportRow>, DataLayerError> {

5
apps/aether-gateway/src/handlers/admin/mod.rs
View File

@@ -17,9 +17,8 @@ pub(crate) use self::auth::maybe_build_local_admin_security_response;
pub(crate) use self::endpoint::build_admin_endpoint_health_status_payload;
pub(crate) use self::features::maybe_build_local_admin_video_tasks_response;
pub(crate) use self::observability::{
admin_stats_bad_request_response, list_usage_for_optional_range,
maybe_build_local_admin_usage_response, parse_bounded_u32, round_to, AdminStatsTimeRange,
AdminStatsUsageFilter,
admin_stats_bad_request_response, maybe_build_local_admin_usage_response, parse_bounded_u32,
round_to, AdminStatsTimeRange, AdminStatsUsageFilter,
};
pub(crate) use self::provider::oauth::errors::build_internal_control_error_response;
pub(crate) use self::provider::ops::providers::actions::admin_provider_ops_local_action_response;

4
apps/aether-gateway/src/handlers/admin/observability/mod.rs
View File

@@ -6,8 +6,8 @@ mod usage;
pub(super) use self::monitoring::maybe_build_local_admin_monitoring_response;
pub(super) use self::routes::maybe_build_local_admin_observability_response;
pub(crate) use self::stats::{
admin_stats_bad_request_response, list_usage_for_optional_range, list_usage_for_range,
maybe_build_local_admin_stats_response, parse_bounded_u32, round_to,
admin_stats_bad_request_response, maybe_build_local_admin_stats_response, parse_bounded_u32,
round_to,
};
pub(crate) use self::stats::{AdminStatsTimeRange, AdminStatsUsageFilter};
pub(crate) use self::usage::maybe_build_local_admin_usage_response;

44
apps/aether-gateway/src/handlers/admin/observability/monitoring/activity.rs
View File

@@ -4,7 +4,6 @@ use super::route_filters::{
parse_admin_monitoring_limit, parse_admin_monitoring_offset,
parse_admin_monitoring_username_filter,
};
use super::usage_helpers::admin_monitoring_usage_is_error;
use crate::constants::INTERNAL_GATEWAY_PATH_PREFIXES;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::query::monitoring as monitoring_query;
@@ -16,7 +15,9 @@ use aether_admin::observability::monitoring::{
build_admin_monitoring_system_status_payload_response,
build_admin_monitoring_user_behavior_payload_response,
};
use aether_data_contracts::repository::usage::UsageAuditListQuery;
use aether_data_contracts::repository::usage::{
UsageAuditSummaryQuery, UsageMonitoringErrorCountQuery,
};
use axum::{body::Body, response::Response};
pub(super) async fn build_admin_monitoring_audit_logs_response(
@@ -196,30 +197,27 @@ pub(super) async fn build_admin_monitoring_system_status_response(
.saturating_add(standalone_api_key_summary.active);
let today_usage = state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(today_start.timestamp().max(0) as u64),
..Default::default()
.summarize_usage_audits(&UsageAuditSummaryQuery {
created_from_unix_secs: today_start.timestamp().max(0) as u64,
created_until_unix_secs: now_unix_secs.saturating_add(1),
user_id: None,
provider_name: None,
model: None,
})
.await?;
let today_requests = today_usage.len();
let today_tokens = today_usage
.iter()
.map(|item| item.total_tokens)
.sum::<u64>();
let today_cost = today_usage
.iter()
.map(|item| item.total_cost_usd)
.sum::<f64>();
let today_requests = usize::try_from(today_usage.total_requests).unwrap_or(usize::MAX);
let today_tokens = today_usage.recorded_total_tokens;
let today_cost = today_usage.total_cost_usd;
let recent_errors = state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(recent_error_from.timestamp().max(0) as u64),
..Default::default()
})
.await?
.into_iter()
.filter(admin_monitoring_usage_is_error)
.count();
let recent_errors = usize::try_from(
state
.count_monitoring_usage_errors(&UsageMonitoringErrorCountQuery {
created_from_unix_secs: recent_error_from.timestamp().max(0) as u64,
created_until_unix_secs: now_unix_secs.saturating_add(1),
})
.await?,
)
.unwrap_or(usize::MAX);
let tunnel = state.tunnel.stats();
Ok(build_admin_monitoring_system_status_payload_response(

29
apps/aether-gateway/src/handlers/admin/observability/monitoring/cache_store.rs
View File

@@ -8,7 +8,7 @@ use crate::handlers::admin::observability::stats::round_to;
use crate::handlers::admin::request::AdminAppState;
use crate::scheduler::affinity::SCHEDULER_AFFINITY_TTL;
use crate::GatewayError;
use aether_data_contracts::repository::usage::UsageAuditListQuery;
use aether_data_contracts::repository::usage::UsageCacheHitSummaryQuery;
async fn count_admin_monitoring_cache_affinity_entries(state: &AdminAppState<'_>) -> usize {
list_admin_monitoring_cache_affinity_records(state)
@@ -325,27 +325,26 @@ pub(super) async fn build_admin_monitoring_cache_snapshot(
.unwrap_or_else(|| "provider".to_string());
let now = chrono::Utc::now();
let usage = if state.has_usage_data_reader() {
let usage_summary = if state.has_usage_data_reader() {
state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(
(now - chrono::Duration::hours(24)).timestamp().max(0) as u64,
),
..Default::default()
.summarize_usage_cache_hit_summary(&UsageCacheHitSummaryQuery {
created_from_unix_secs: (now - chrono::Duration::hours(24)).timestamp().max(0)
as u64,
created_until_unix_secs: now.timestamp().max(0) as u64,
user_id: None,
})
.await?
} else {
Vec::new()
Default::default()
};
let cache_hits = usage
.iter()
.filter(|item| item.cache_read_input_tokens > 0)
.count();
let cache_misses = usage.len().saturating_sub(cache_hits);
let cache_hit_rate = if usage.is_empty() {
let cache_hits = usage_summary.cache_hit_requests as usize;
let cache_misses = usage_summary
.total_requests
.saturating_sub(usage_summary.cache_hit_requests) as usize;
let cache_hit_rate = if usage_summary.total_requests == 0 {
0.0
} else {
round_to(cache_hits as f64 / usage.len() as f64, 4)
round_to(cache_hits as f64 / usage_summary.total_requests as f64, 4)
};
let total_affinities = count_admin_monitoring_cache_affinity_entries(state).await;
let storage_type = if state.redis_kv_runner().is_some() {

15
apps/aether-gateway/src/handlers/admin/observability/monitoring/resilience/snapshot.rs
View File

@@ -1,9 +1,8 @@
use super::super::usage_helpers::admin_monitoring_usage_is_error;
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::{provider_key_health_summary, unix_secs_to_rfc3339};
use crate::GatewayError;
use aether_data_contracts::repository::{
provider_catalog::StoredProviderCatalogKey, usage::UsageAuditListQuery,
provider_catalog::StoredProviderCatalogKey, usage::UsageMonitoringErrorListQuery,
};
use serde_json::json;
use std::collections::BTreeMap;
@@ -152,14 +151,12 @@ pub(super) async fn build_admin_monitoring_resilience_snapshot(
}
let mut recent_usage_errors = state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(recent_error_from.timestamp().max(0) as u64),
..Default::default()
.list_monitoring_usage_errors(&UsageMonitoringErrorListQuery {
created_from_unix_secs: recent_error_from.timestamp().max(0) as u64,
created_until_unix_secs: (now.timestamp().max(0) as u64).saturating_add(1),
limit: None,
})
.await?
.into_iter()
.filter(admin_monitoring_usage_is_error)
.collect::<Vec<_>>();
.await?;
recent_usage_errors
.sort_by(|left, right| right.created_at_unix_ms.cmp(&left.created_at_unix_ms));

45
apps/aether-gateway/src/handlers/admin/observability/stats/analytics_routes.rs
View File

@@ -8,13 +8,14 @@ use aether_admin::observability::stats::{
admin_stats_error_distribution_empty_response,
admin_stats_performance_percentiles_empty_response, admin_stats_time_series_empty_response,
build_admin_stats_comparison_response_from_aggregates,
build_admin_stats_error_distribution_response,
build_admin_stats_performance_percentiles_response,
build_admin_stats_error_distribution_response_from_summaries,
build_admin_stats_performance_percentiles_response_from_summaries,
build_admin_stats_time_series_response_from_summaries, AdminStatsAggregate,
AdminStatsComparisonType, AdminStatsGranularity, AdminStatsTimeRange, AdminStatsUsageFilter,
};
use aether_data_contracts::repository::usage::{
UsageAuditSummaryQuery, UsageTimeSeriesGranularity, UsageTimeSeriesQuery,
UsageAuditSummaryQuery, UsageErrorDistributionQuery, UsagePerformancePercentilesQuery,
UsageTimeSeriesGranularity, UsageTimeSeriesQuery,
};
use axum::{body::Body, http, response::Response};
@@ -124,13 +125,20 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response(
return Ok(Some(admin_stats_error_distribution_empty_response()));
}
let usage = state
.list_admin_usage_for_range(&time_range, &AdminStatsUsageFilter::default())
let Some((created_from_unix_secs, created_until_unix_secs)) = time_range.to_unix_bounds()
else {
return Ok(Some(admin_stats_error_distribution_empty_response()));
};
let rows = state
.summarize_usage_error_distribution(&UsageErrorDistributionQuery {
created_from_unix_secs,
created_until_unix_secs,
tz_offset_minutes: time_range.tz_offset_minutes,
})
.await?;
return Ok(Some(build_admin_stats_error_distribution_response(
&time_range,
&usage,
)));
return Ok(Some(
build_admin_stats_error_distribution_response_from_summaries(&rows),
));
}
if request_context.route_kind() == Some("performance_percentiles")
@@ -149,13 +157,20 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response(
return Ok(Some(admin_stats_performance_percentiles_empty_response()));
}
let usage = state
.list_admin_usage_for_range(&time_range, &AdminStatsUsageFilter::default())
let Some((created_from_unix_secs, created_until_unix_secs)) = time_range.to_unix_bounds()
else {
return Ok(Some(admin_stats_performance_percentiles_empty_response()));
};
let rows = state
.summarize_usage_performance_percentiles(&UsagePerformancePercentilesQuery {
created_from_unix_secs,
created_until_unix_secs,
tz_offset_minutes: time_range.tz_offset_minutes,
})
.await?;
return Ok(Some(build_admin_stats_performance_percentiles_response(
&time_range,
&usage,
)));
return Ok(Some(
build_admin_stats_performance_percentiles_response_from_summaries(&time_range, &rows),
));
}
if request_context.route_kind() == Some("time_series")

53
apps/aether-gateway/src/handlers/admin/observability/stats/cost_routes.rs
View File

@@ -6,7 +6,12 @@ use crate::GatewayError;
use aether_admin::observability::stats::{
admin_stats_bad_request_response, admin_stats_cost_forecast_empty_response,
admin_stats_cost_savings_empty_response, build_admin_stats_cost_forecast_response,
build_admin_stats_cost_savings_response, AdminStatsGranularity, AdminStatsUsageFilter,
build_admin_stats_cost_forecast_response_from_summaries,
build_admin_stats_cost_savings_response, build_admin_stats_cost_savings_response_from_summary,
AdminStatsGranularity, AdminStatsUsageFilter,
};
use aether_data_contracts::repository::usage::{
UsageCostSavingsSummaryQuery, UsageTimeSeriesGranularity, UsageTimeSeriesQuery,
};
use axum::{body::Body, http, response::Response};
@@ -61,14 +66,28 @@ pub(super) async fn maybe_build_local_admin_stats_cost_response(
return Ok(Some(admin_stats_bad_request_response(detail)));
}
let usage = state
.list_admin_usage_for_range(&time_range, &AdminStatsUsageFilter::default())
let Some((created_from_unix_secs, created_until_unix_secs)) = time_range.to_unix_bounds()
else {
return Ok(Some(admin_stats_cost_forecast_empty_response()));
};
let buckets = state
.summarize_usage_time_series(&UsageTimeSeriesQuery {
created_from_unix_secs,
created_until_unix_secs,
granularity: UsageTimeSeriesGranularity::Day,
tz_offset_minutes: time_range.tz_offset_minutes,
user_id: None,
provider_name: None,
model: None,
})
.await?;
return Ok(Some(build_admin_stats_cost_forecast_response(
&time_range,
forecast_days,
&usage,
)));
return Ok(Some(
build_admin_stats_cost_forecast_response_from_summaries(
&time_range,
forecast_days,
&buckets,
),
));
}
if request_context
@@ -94,11 +113,23 @@ pub(super) async fn maybe_build_local_admin_stats_cost_response(
provider_name: query_param_value(query, "provider_name"),
model: query_param_value(query, "model"),
};
let usage = state
.list_admin_usage_for_range(&time_range, &filters)
let Some((created_from_unix_secs, created_until_unix_secs)) = time_range.to_unix_bounds()
else {
return Ok(Some(admin_stats_cost_savings_empty_response()));
};
let summary = state
.summarize_usage_cost_savings(&UsageCostSavingsSummaryQuery {
created_from_unix_secs,
created_until_unix_secs,
user_id: None,
provider_name: filters.provider_name,
model: filters.model,
})
.await?;
return Ok(Some(build_admin_stats_cost_savings_response(&usage)));
return Ok(Some(build_admin_stats_cost_savings_response_from_summary(
&summary,
)));
}
Ok(None)

5
apps/aether-gateway/src/handlers/admin/observability/stats/mod.rs
View File

@@ -8,10 +8,7 @@ mod leaderboard;
mod leaderboard_routes;
mod provider_quota_routes;
mod range;
pub(crate) use self::range::{
list_usage_for_optional_range, list_usage_for_range, parse_bounded_u32,
resolve_admin_usage_time_range,
};
pub(crate) use self::range::{parse_bounded_u32, resolve_admin_usage_time_range};
pub(crate) use aether_admin::observability::stats::{
admin_stats_bad_request_response, aggregate_usage_stats, round_to, AdminStatsTimeRange,
AdminStatsUsageFilter,

48
apps/aether-gateway/src/handlers/admin/observability/stats/range.rs
View File

@@ -1,12 +1,9 @@
use crate::handlers::admin::request::AdminAppState;
use crate::GatewayError;
pub(crate) use aether_admin::observability::stats::parse_bounded_u32;
use aether_admin::observability::stats::AdminStatsTimeRange;
pub(super) use aether_admin::observability::stats::{
admin_usage_default_days, build_comparison_range, build_time_range_from_days, parse_naive_date,
parse_nonnegative_usize, parse_tz_offset_minutes, resolve_preset_dates, user_today,
};
use aether_admin::observability::stats::{AdminStatsTimeRange, AdminStatsUsageFilter};
use aether_data_contracts::repository::usage::{StoredRequestUsageAudit, UsageAuditListQuery};
pub(crate) fn resolve_admin_usage_time_range(
query: Option<&str>,
@@ -23,46 +20,3 @@ pub(crate) fn resolve_admin_usage_time_range(
}
}
}
pub(crate) async fn list_usage_for_range(
state: &AdminAppState<'_>,
time_range: &AdminStatsTimeRange,
filters: &AdminStatsUsageFilter,
) -> Result<Vec<StoredRequestUsageAudit>, GatewayError> {
let Some((created_from_unix_secs, created_until_unix_secs)) = time_range.to_unix_bounds()
else {
return Ok(Vec::new());
};
state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(created_from_unix_secs),
created_until_unix_secs: Some(created_until_unix_secs),
user_id: filters.user_id.clone(),
provider_name: filters.provider_name.clone(),
model: filters.model.clone(),
api_format: None,
statuses: None,
is_stream: None,
error_only: false,
limit: None,
offset: None,
newest_first: false,
})
.await
}
pub(crate) async fn list_usage_for_optional_range(
state: &AdminAppState<'_>,
time_range: Option<&AdminStatsTimeRange>,
filters: &AdminStatsUsageFilter,
) -> Result<Vec<StoredRequestUsageAudit>, GatewayError> {
match time_range {
Some(time_range) => list_usage_for_range(state, time_range, filters).await,
None => {
let default_time_range = build_time_range_from_days(1, 0)
.map_err(|detail| GatewayError::Internal(detail.to_string()))?;
list_usage_for_range(state, &default_time_range, filters).await
}
}
}

40
apps/aether-gateway/src/handlers/admin/observability/usage/analytics/cache_affinity.rs
View File

@@ -1,35 +1,25 @@
use crate::handlers::admin::request::AdminAppState;
use crate::GatewayError;
use aether_data_contracts::repository::usage::{StoredRequestUsageAudit, UsageAuditListQuery};
use aether_data_contracts::repository::usage::{
StoredUsageCacheAffinityIntervalRow, UsageCacheAffinityIntervalGroupBy,
UsageCacheAffinityIntervalQuery,
};
pub(in super::super) async fn list_recent_completed_usage_for_cache_affinity(
pub(in super::super) async fn list_usage_cache_affinity_intervals(
state: &AdminAppState<'_>,
hours: u32,
group_by: UsageCacheAffinityIntervalGroupBy,
user_id: Option<&str>,
) -> Result<Vec<StoredRequestUsageAudit>, GatewayError> {
api_key_id: Option<&str>,
) -> Result<Vec<StoredUsageCacheAffinityIntervalRow>, GatewayError> {
let now_unix_secs = u64::try_from(chrono::Utc::now().timestamp()).unwrap_or_default();
let created_from_unix_secs = now_unix_secs.saturating_sub(u64::from(hours) * 3600);
let mut items = state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(created_from_unix_secs),
created_until_unix_secs: None,
state
.list_usage_cache_affinity_intervals(&UsageCacheAffinityIntervalQuery {
created_from_unix_secs: now_unix_secs.saturating_sub(u64::from(hours) * 3600),
created_until_unix_secs: now_unix_secs.saturating_add(1),
group_by,
user_id: user_id.map(ToOwned::to_owned),
provider_name: None,
model: None,
api_format: None,
statuses: None,
is_stream: None,
error_only: false,
limit: None,
offset: None,
newest_first: false,
api_key_id: api_key_id.map(ToOwned::to_owned),
})
.await?;
items.retain(|item| item.status == "completed");
items.sort_by(|left, right| {
left.created_at_unix_ms
.cmp(&right.created_at_unix_ms)
.then_with(|| left.id.cmp(&right.id))
});
Ok(items)
.await
}

2
apps/aether-gateway/src/handlers/admin/observability/usage/analytics/mod.rs
View File

@@ -3,6 +3,6 @@ mod cache_affinity;
mod filters;
pub(super) use aggregations::admin_usage_aggregation_by_user_json;
pub(super) use cache_affinity::list_recent_completed_usage_for_cache_affinity;
pub(super) use cache_affinity::list_usage_cache_affinity_intervals;
pub(super) use filters::admin_usage_api_key_names;
pub(super) use filters::admin_usage_provider_key_names;

69
apps/aether-gateway/src/handlers/admin/observability/usage/analytics_routes/cache_affinity_hit_analysis.rs
View File

@@ -1,14 +1,12 @@
use super::super::super::stats::round_to;
use super::super::analytics::list_recent_completed_usage_for_cache_affinity;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::handlers::admin::shared::query_param_value;
use crate::GatewayError;
use aether_admin::observability::usage::{
admin_usage_bad_request_response, admin_usage_cache_creation_tokens,
admin_usage_data_unavailable_response, admin_usage_matches_optional_id,
admin_usage_parse_recent_hours, admin_usage_total_input_context,
ADMIN_USAGE_DATA_UNAVAILABLE_DETAIL,
admin_usage_bad_request_response, admin_usage_data_unavailable_response,
admin_usage_parse_recent_hours, ADMIN_USAGE_DATA_UNAVAILABLE_DETAIL,
};
use aether_data_contracts::repository::usage::UsageCacheAffinityHitSummaryQuery;
use axum::{
body::Body,
response::{IntoResponse, Response},
@@ -33,65 +31,44 @@ pub(super) async fn build_admin_usage_cache_affinity_hit_analysis_response(
};
let user_id = query_param_value(query, "user_id");
let api_key_id = query_param_value(query, "api_key_id");
let usage =
list_recent_completed_usage_for_cache_affinity(state, hours, user_id.as_deref()).await?;
let filtered: Vec<_> = usage
.into_iter()
.filter(|item| {
admin_usage_matches_optional_id(item.user_id.as_deref(), user_id.as_deref())
&& admin_usage_matches_optional_id(
item.api_key_id.as_deref(),
api_key_id.as_deref(),
)
let now_unix_secs = u64::try_from(chrono::Utc::now().timestamp()).unwrap_or_default();
let summary = state
.summarize_usage_cache_affinity_hit_summary(&UsageCacheAffinityHitSummaryQuery {
created_from_unix_secs: now_unix_secs.saturating_sub(u64::from(hours) * 3600),
created_until_unix_secs: now_unix_secs.saturating_add(1),
user_id,
api_key_id,
})
.collect();
let total_requests = filtered.len();
let total_input_tokens: u64 = filtered.iter().map(|item| item.input_tokens).sum();
let total_cache_read_tokens: u64 = filtered
.iter()
.map(|item| item.cache_read_input_tokens)
.sum();
let total_cache_creation_tokens: u64 =
filtered.iter().map(admin_usage_cache_creation_tokens).sum();
let total_input_context: u64 = filtered.iter().map(admin_usage_total_input_context).sum();
let total_cache_read_cost: f64 = filtered.iter().map(|item| item.cache_read_cost_usd).sum();
let total_cache_creation_cost: f64 = filtered
.iter()
.map(|item| item.cache_creation_cost_usd)
.sum();
let requests_with_cache_hit = filtered
.iter()
.filter(|item| item.cache_read_input_tokens > 0)
.count();
let token_cache_hit_rate = if total_input_context == 0 {
.await?;
let token_cache_hit_rate = if summary.total_input_context == 0 {
0.0
} else {
round_to(
total_cache_read_tokens as f64 / total_input_context as f64 * 100.0,
summary.cache_read_tokens as f64 / summary.total_input_context as f64 * 100.0,
2,
)
};
let request_cache_hit_rate = if total_requests == 0 {
let request_cache_hit_rate = if summary.total_requests == 0 {
0.0
} else {
round_to(
requests_with_cache_hit as f64 / total_requests as f64 * 100.0,
summary.requests_with_cache_hit as f64 / summary.total_requests as f64 * 100.0,
2,
)
};
Ok(Json(json!({
"analysis_period_hours": hours,
"total_requests": total_requests,
"requests_with_cache_hit": requests_with_cache_hit,
"total_requests": summary.total_requests,
"requests_with_cache_hit": summary.requests_with_cache_hit,
"request_cache_hit_rate": request_cache_hit_rate,
"total_input_tokens": total_input_tokens,
"total_cache_read_tokens": total_cache_read_tokens,
"total_cache_creation_tokens": total_cache_creation_tokens,
"total_input_tokens": summary.input_tokens,
"total_cache_read_tokens": summary.cache_read_tokens,
"total_cache_creation_tokens": summary.cache_creation_tokens,
"token_cache_hit_rate": token_cache_hit_rate,
"total_cache_read_cost_usd": round_to(total_cache_read_cost, 4),
"total_cache_creation_cost_usd": round_to(total_cache_creation_cost, 4),
"estimated_savings_usd": round_to(total_cache_read_cost * 9.0, 4),
"total_cache_read_cost_usd": round_to(summary.cache_read_cost_usd, 4),
"total_cache_creation_cost_usd": round_to(summary.cache_creation_cost_usd, 4),
"estimated_savings_usd": round_to(summary.cache_read_cost_usd * 9.0, 4),
}))
.into_response())
}

72
apps/aether-gateway/src/handlers/admin/observability/usage/analytics_routes/cache_affinity_interval_timeline.rs
View File

@@ -1,13 +1,13 @@
use super::super::analytics::list_recent_completed_usage_for_cache_affinity;
use super::super::analytics::list_usage_cache_affinity_intervals;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::handlers::admin::shared::{query_param_bool, query_param_value, unix_secs_to_rfc3339};
use crate::GatewayError;
use aether_admin::observability::usage::{
admin_usage_bad_request_response, admin_usage_data_unavailable_response,
admin_usage_group_completed_by_user, admin_usage_parse_recent_hours,
admin_usage_parse_timeline_limit, admin_usage_point_sort_key, admin_usage_proportional_limits,
ADMIN_USAGE_DATA_UNAVAILABLE_DETAIL,
admin_usage_parse_recent_hours, admin_usage_parse_timeline_limit, admin_usage_point_sort_key,
admin_usage_proportional_limits, ADMIN_USAGE_DATA_UNAVAILABLE_DETAIL,
};
use aether_data_contracts::repository::usage::UsageCacheAffinityIntervalGroupBy;
use axum::{
body::Body,
response::{IntoResponse, Response},
@@ -51,50 +51,40 @@ pub(super) async fn build_admin_usage_cache_affinity_interval_timeline_response(
};
let user_id = query_param_value(query, "user_id");
let include_user_info = query_param_bool(query, "include_user_info", false);
let usage =
list_recent_completed_usage_for_cache_affinity(state, hours, user_id.as_deref()).await?;
let intervals = list_usage_cache_affinity_intervals(
state,
hours,
UsageCacheAffinityIntervalGroupBy::User,
user_id.as_deref(),
None,
)
.await?;
let mut grouped: BTreeMap<String, Vec<serde_json::Value>> = BTreeMap::new();
let mut models = BTreeSet::new();
let mut legacy_usernames_by_user_id = BTreeMap::new();
let mut usernames_by_user_id = BTreeMap::new();
for (group_user_id, items) in admin_usage_group_completed_by_user(&usage) {
if let Some(ref requested_user_id) = user_id {
if &group_user_id != requested_user_id {
continue;
for row in intervals {
if row.interval_minutes > 120.0 {
continue;
}
let mut point = json!({
"x": unix_secs_to_rfc3339(row.created_at_unix_secs),
"y": ((row.interval_minutes * 100.0).round()) / 100.0,
});
if !row.model.trim().is_empty() {
point["model"] = json!(row.model.clone());
models.insert(row.model);
}
if include_user_info && user_id.is_none() {
point["user_id"] = json!(row.group_id.clone());
if let Some(username) = row.username {
legacy_usernames_by_user_id
.entry(row.group_id.clone())
.or_insert(username);
}
}
let mut previous_created_at_unix_ms = None;
for item in items {
if let Some(previous) = previous_created_at_unix_ms {
let interval_minutes =
item.created_at_unix_ms.saturating_sub(previous) as f64 / 60.0;
if interval_minutes <= 120.0 {
let mut point = json!({
"x": unix_secs_to_rfc3339(item.created_at_unix_ms),
"y": ((interval_minutes * 100.0).round()) / 100.0,
});
if !item.model.trim().is_empty() {
point["model"] = json!(item.model.clone());
models.insert(item.model.clone());
}
if include_user_info && user_id.is_none() {
point["user_id"] = json!(group_user_id.clone());
if let Some(username) = item.username.clone() {
legacy_usernames_by_user_id
.entry(group_user_id.clone())
.or_insert(username);
}
}
grouped
.entry(group_user_id.clone())
.or_default()
.push(point);
}
}
previous_created_at_unix_ms = Some(item.created_at_unix_ms);
}
grouped.entry(row.group_id).or_default().push(point);
}
if include_user_info && user_id.is_none() {

46
apps/aether-gateway/src/handlers/admin/observability/usage/analytics_routes/cache_affinity_ttl_analysis.rs
View File

@@ -1,15 +1,15 @@
use super::super::super::stats::round_to;
use super::super::analytics::list_recent_completed_usage_for_cache_affinity;
use super::super::analytics::list_usage_cache_affinity_intervals;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::handlers::admin::shared::query_param_value;
use crate::GatewayError;
use aether_admin::observability::usage::{
admin_usage_bad_request_response, admin_usage_calculate_recommended_ttl,
admin_usage_collect_request_intervals_minutes, admin_usage_data_unavailable_response,
admin_usage_group_completed_by_api_key, admin_usage_group_completed_by_user,
admin_usage_matches_optional_id, admin_usage_parse_recent_hours, admin_usage_percentile_cont,
admin_usage_ttl_recommendation_reason, ADMIN_USAGE_DATA_UNAVAILABLE_DETAIL,
admin_usage_data_unavailable_response, admin_usage_parse_recent_hours,
admin_usage_percentile_cont, admin_usage_ttl_recommendation_reason,
ADMIN_USAGE_DATA_UNAVAILABLE_DETAIL,
};
use aether_data_contracts::repository::usage::UsageCacheAffinityIntervalGroupBy;
use axum::{
body::Body,
response::{IntoResponse, Response},
@@ -36,19 +36,28 @@ pub(super) async fn build_admin_usage_cache_affinity_ttl_analysis_response(
let user_id = query_param_value(query, "user_id");
let api_key_id = query_param_value(query, "api_key_id");
let group_by_api_key = api_key_id.is_some();
let usage =
list_recent_completed_usage_for_cache_affinity(state, hours, user_id.as_deref()).await?;
let grouped = if group_by_api_key {
admin_usage_group_completed_by_api_key(&usage, api_key_id.as_deref())
} else {
admin_usage_group_completed_by_user(&usage)
let intervals = list_usage_cache_affinity_intervals(
state,
hours,
if group_by_api_key {
UsageCacheAffinityIntervalGroupBy::ApiKey
} else {
UsageCacheAffinityIntervalGroupBy::User
},
user_id.as_deref(),
api_key_id.as_deref(),
)
.await?;
let grouped =
intervals
.into_iter()
.filter(|(group_user_id, _)| {
admin_usage_matches_optional_id(Some(group_user_id.as_str()), user_id.as_deref())
})
.collect()
};
.fold(BTreeMap::<String, Vec<f64>>::new(), |mut grouped, row| {
grouped
.entry(row.group_id)
.or_default()
.push(row.interval_minutes);
grouped
});
let user_map: BTreeMap<String, aether_data::repository::users::StoredUserSummary> =
if !group_by_api_key && state.has_user_data_reader() {
@@ -71,8 +80,7 @@ pub(super) async fn build_admin_usage_cache_affinity_ttl_analysis_response(
});
let mut users = Vec::new();
for (group_id, items) in grouped {
let intervals = admin_usage_collect_request_intervals_minutes(&items);
for (group_id, intervals) in grouped {
if intervals.len() < 2 {
continue;
}

231
apps/aether-gateway/src/handlers/admin/observability/usage/summary_routes.rs
View File

@@ -5,14 +5,14 @@ use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::handlers::admin::shared::query_param_value;
use crate::GatewayError;
use aether_admin::observability::usage::{
admin_usage_bad_request_response, admin_usage_data_unavailable_response,
admin_usage_matches_search, admin_usage_matches_username, admin_usage_parse_ids,
admin_usage_bad_request_response, admin_usage_data_unavailable_response, admin_usage_parse_ids,
admin_usage_parse_limit, admin_usage_parse_offset, build_admin_usage_active_requests_response,
build_admin_usage_records_response, build_admin_usage_summary_stats_response_from_summary,
ADMIN_USAGE_DATA_UNAVAILABLE_DETAIL,
};
use aether_data_contracts::repository::usage::{
StoredRequestUsageAudit, UsageAuditListQuery, UsageAuditSummaryQuery,
StoredRequestUsageAudit, UsageAuditKeywordSearchQuery, UsageAuditListQuery,
UsageAuditSummaryQuery,
};
use axum::{body::Body, http, response::Response};
use std::collections::{BTreeMap, BTreeSet};
@@ -23,13 +23,8 @@ async fn load_admin_usage_by_ids(
state: &AdminAppState<'_>,
requested_ids: &BTreeSet<String>,
) -> Result<Vec<StoredRequestUsageAudit>, GatewayError> {
let mut items = Vec::with_capacity(requested_ids.len());
for usage_id in requested_ids {
if let Some(item) = state.find_request_usage_by_id(usage_id).await? {
items.push(item);
}
}
Ok(items)
let usage_ids = requested_ids.iter().cloned().collect::<Vec<_>>();
state.list_request_usage_by_ids(&usage_ids).await
}
fn sort_usage_newest_first(items: &mut [StoredRequestUsageAudit]) {
@@ -89,6 +84,134 @@ fn build_admin_usage_records_query(
list_query
}
fn parse_admin_usage_search_keywords(search: &str) -> Vec<String> {
search
.split_whitespace()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase())
.collect()
}
#[derive(Default)]
struct AdminUsageSearchContext {
matched_user_ids_by_keyword: Vec<Vec<String>>,
matched_api_key_ids_by_keyword: Vec<Vec<String>>,
matched_user_ids_for_username: Vec<String>,
}
async fn resolve_admin_usage_search_context(
state: &AdminAppState<'_>,
keywords: &[String],
username_filter: Option<&str>,
) -> Result<AdminUsageSearchContext, GatewayError> {
let auth_user_reader_available = state.has_auth_user_data_reader();
let auth_api_key_reader_available = state.has_auth_api_key_data_reader();
let username_filter = username_filter
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
let mut matched_user_ids_cache = BTreeMap::<String, Vec<String>>::new();
let mut matched_api_key_ids_cache = BTreeMap::<String, Vec<String>>::new();
if auth_user_reader_available {
for keyword in keywords {
matched_user_ids_cache.entry(keyword.clone()).or_insert(
state
.search_auth_user_summaries_by_username(keyword)
.await?
.into_iter()
.map(|user| user.id)
.collect(),
);
}
if let Some(username_keyword) = username_filter.as_ref() {
matched_user_ids_cache
.entry(username_keyword.clone())
.or_insert(
state
.search_auth_user_summaries_by_username(username_keyword)
.await?
.into_iter()
.map(|user| user.id)
.collect(),
);
}
}
if auth_api_key_reader_available {
for keyword in keywords {
matched_api_key_ids_cache.entry(keyword.clone()).or_insert(
state
.list_auth_api_key_export_records_by_name_search(keyword)
.await?
.into_iter()
.map(|record| record.api_key_id)
.collect(),
);
}
}
Ok(AdminUsageSearchContext {
matched_user_ids_by_keyword: keywords
.iter()
.map(|keyword| {
matched_user_ids_cache
.get(keyword)
.cloned()
.unwrap_or_default()
})
.collect(),
matched_api_key_ids_by_keyword: keywords
.iter()
.map(|keyword| {
matched_api_key_ids_cache
.get(keyword)
.cloned()
.unwrap_or_default()
})
.collect(),
matched_user_ids_for_username: username_filter
.as_ref()
.and_then(|keyword| matched_user_ids_cache.get(keyword))
.cloned()
.unwrap_or_default(),
})
}
fn build_admin_usage_keyword_search_query(
base_query: &UsageAuditListQuery,
keywords: Vec<String>,
username_keyword: Option<String>,
search_context: AdminUsageSearchContext,
auth_user_reader_available: bool,
auth_api_key_reader_available: bool,
limit: Option<usize>,
offset: Option<usize>,
) -> UsageAuditKeywordSearchQuery {
UsageAuditKeywordSearchQuery {
created_from_unix_secs: base_query.created_from_unix_secs,
created_until_unix_secs: base_query.created_until_unix_secs,
user_id: base_query.user_id.clone(),
provider_name: base_query.provider_name.clone(),
model: base_query.model.clone(),
api_format: base_query.api_format.clone(),
statuses: base_query.statuses.clone(),
is_stream: base_query.is_stream,
error_only: base_query.error_only,
keywords,
matched_user_ids_by_keyword: search_context.matched_user_ids_by_keyword,
auth_user_reader_available,
matched_api_key_ids_by_keyword: search_context.matched_api_key_ids_by_keyword,
auth_api_key_reader_available,
username_keyword,
matched_user_ids_for_username: search_context.matched_user_ids_for_username,
limit,
offset,
newest_first: true,
}
}
pub(super) async fn maybe_build_local_admin_usage_summary_response(
state: &AdminAppState<'_>,
request_context: &AdminRequestContext<'_>,
@@ -150,8 +273,10 @@ pub(super) async fn maybe_build_local_admin_usage_summary_response(
let query = request_context.request_query_string.as_deref();
let requested_ids = admin_usage_parse_ids(query);
let mut items = if let Some(requested_ids) = requested_ids.as_ref() {
load_admin_usage_by_ids(state, requested_ids).await?
let items = if let Some(requested_ids) = requested_ids.as_ref() {
let mut items = load_admin_usage_by_ids(state, requested_ids).await?;
sort_usage_newest_first(&mut items);
items
} else {
let time_range = match resolve_admin_usage_time_range(query) {
Ok(value) => value,
@@ -178,10 +303,6 @@ pub(super) async fn maybe_build_local_admin_usage_summary_response(
})
.await?
};
sort_usage_newest_first(&mut items);
if requested_ids.is_none() && items.len() > ADMIN_USAGE_ACTIVE_LIMIT {
items.truncate(ADMIN_USAGE_ACTIVE_LIMIT);
}
let api_key_names = admin_usage_api_key_names(state, &items).await?;
let provider_key_names = admin_usage_provider_key_names(state, &items).await?;
@@ -242,44 +363,46 @@ pub(super) async fn maybe_build_local_admin_usage_summary_response(
None,
None,
);
let use_metadata_fallback = search.is_some() || username_filter.is_some();
let (usage, total) = if use_metadata_fallback {
let mut usage = state.list_usage_audits(&base_query).await?;
let user_ids: Vec<String> = usage
.iter()
.filter_map(|item| item.user_id.clone())
.collect::<BTreeSet<_>>()
.into_iter()
.collect();
let users_by_id: BTreeMap<
String,
aether_data::repository::users::StoredUserSummary,
> = state.resolve_auth_user_summaries_by_ids(&user_ids).await?;
let api_key_names = admin_usage_api_key_names(state, &usage).await?;
usage.retain(|item| {
admin_usage_matches_search(
item,
search.as_deref(),
&users_by_id,
&api_key_names,
state.has_auth_user_data_reader(),
state.has_auth_api_key_data_reader(),
) && admin_usage_matches_username(
item,
username_filter.as_deref(),
&users_by_id,
state.has_auth_user_data_reader(),
)
});
sort_usage_newest_first(&mut usage);
let total = usage.len();
let records = usage
.into_iter()
.skip(offset)
.take(limit)
.collect::<Vec<_>>();
(records, total)
let active_search = search.as_deref().filter(|value| !value.trim().is_empty());
let active_username_filter = username_filter
.as_deref()
.filter(|value| !value.trim().is_empty());
let (usage, total) = if active_search.is_some() || active_username_filter.is_some() {
let keywords = active_search
.map(parse_admin_usage_search_keywords)
.unwrap_or_default();
let auth_user_reader_available = state.has_auth_user_data_reader();
let auth_api_key_reader_available = state.has_auth_api_key_data_reader();
let search_context =
resolve_admin_usage_search_context(state, &keywords, active_username_filter)
.await?;
let keyword_query = build_admin_usage_keyword_search_query(
&base_query,
keywords,
active_username_filter.map(str::to_owned),
search_context,
auth_user_reader_available,
auth_api_key_reader_available,
None,
None,
);
let total = usize::try_from(
state
.count_usage_audits_by_keyword_search(&keyword_query)
.await?,
)
.unwrap_or(usize::MAX);
let paged_query = UsageAuditKeywordSearchQuery {
limit: Some(limit),
offset: Some(offset),
..keyword_query
};
(
state
.list_usage_audits_by_keyword_search(&paged_query)
.await?,
total,
)
} else {
let total = usize::try_from(state.count_usage_audits(&base_query).await?)
.unwrap_or(usize::MAX);

120
apps/aether-gateway/src/handlers/admin/request/observability.rs
View File

@@ -47,6 +47,29 @@ impl<'a> AdminAppState<'a> {
self.app.count_usage_audits(query).await
}
pub(crate) async fn list_usage_audits_by_keyword_search(
&self,
query: &aether_data_contracts::repository::usage::UsageAuditKeywordSearchQuery,
) -> Result<Vec<aether_data_contracts::repository::usage::StoredRequestUsageAudit>, GatewayError>
{
self.app.list_usage_audits_by_keyword_search(query).await
}
pub(crate) async fn count_usage_audits_by_keyword_search(
&self,
query: &aether_data_contracts::repository::usage::UsageAuditKeywordSearchQuery,
) -> Result<u64, GatewayError> {
self.app.count_usage_audits_by_keyword_search(query).await
}
pub(crate) async fn list_monitoring_usage_errors(
&self,
query: &aether_data_contracts::repository::usage::UsageMonitoringErrorListQuery,
) -> Result<Vec<aether_data_contracts::repository::usage::StoredRequestUsageAudit>, GatewayError>
{
self.app.list_monitoring_usage_errors(query).await
}
pub(crate) async fn aggregate_usage_audits(
&self,
query: &aether_data_contracts::repository::usage::UsageAuditAggregationQuery,
@@ -65,6 +88,44 @@ impl<'a> AdminAppState<'a> {
self.app.summarize_usage_audits(query).await
}
pub(crate) async fn summarize_usage_cache_hit_summary(
&self,
query: &aether_data_contracts::repository::usage::UsageCacheHitSummaryQuery,
) -> Result<aether_data_contracts::repository::usage::StoredUsageCacheHitSummary, GatewayError>
{
self.app.summarize_usage_cache_hit_summary(query).await
}
pub(crate) async fn summarize_usage_settled_cost(
&self,
query: &aether_data_contracts::repository::usage::UsageSettledCostSummaryQuery,
) -> Result<aether_data_contracts::repository::usage::StoredUsageSettledCostSummary, GatewayError>
{
self.app.summarize_usage_settled_cost(query).await
}
pub(crate) async fn summarize_usage_cache_affinity_hit_summary(
&self,
query: &aether_data_contracts::repository::usage::UsageCacheAffinityHitSummaryQuery,
) -> Result<
aether_data_contracts::repository::usage::StoredUsageCacheAffinityHitSummary,
GatewayError,
> {
self.app
.summarize_usage_cache_affinity_hit_summary(query)
.await
}
pub(crate) async fn list_usage_cache_affinity_intervals(
&self,
query: &aether_data_contracts::repository::usage::UsageCacheAffinityIntervalQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsageCacheAffinityIntervalRow>,
GatewayError,
> {
self.app.list_usage_cache_affinity_intervals(query).await
}
pub(crate) async fn summarize_usage_time_series(
&self,
query: &aether_data_contracts::repository::usage::UsageTimeSeriesQuery,
@@ -85,6 +146,36 @@ impl<'a> AdminAppState<'a> {
self.app.summarize_usage_leaderboard(query).await
}
pub(crate) async fn summarize_usage_error_distribution(
&self,
query: &aether_data_contracts::repository::usage::UsageErrorDistributionQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsageErrorDistributionRow>,
GatewayError,
> {
self.app.summarize_usage_error_distribution(query).await
}
pub(crate) async fn summarize_usage_performance_percentiles(
&self,
query: &aether_data_contracts::repository::usage::UsagePerformancePercentilesQuery,
) -> Result<
Vec<aether_data_contracts::repository::usage::StoredUsagePerformancePercentilesRow>,
GatewayError,
> {
self.app
.summarize_usage_performance_percentiles(query)
.await
}
pub(crate) async fn summarize_usage_cost_savings(
&self,
query: &aether_data_contracts::repository::usage::UsageCostSavingsSummaryQuery,
) -> Result<aether_data_contracts::repository::usage::StoredUsageCostSavingsSummary, GatewayError>
{
self.app.summarize_usage_cost_savings(query).await
}
pub(crate) async fn summarize_usage_daily_heatmap(
&self,
query: &aether_data_contracts::repository::usage::UsageDailyHeatmapQuery,
@@ -107,6 +198,14 @@ impl<'a> AdminAppState<'a> {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_request_usage_by_ids(
&self,
usage_ids: &[String],
) -> Result<Vec<aether_data_contracts::repository::usage::StoredRequestUsageAudit>, GatewayError>
{
self.app.list_request_usage_by_ids(usage_ids).await
}
pub(crate) async fn resolve_request_usage_body_ref(
&self,
body_ref: &str,
@@ -118,27 +217,6 @@ impl<'a> AdminAppState<'a> {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_admin_usage_for_range(
&self,
time_range: &crate::handlers::admin::observability::AdminStatsTimeRange,
filters: &crate::handlers::admin::observability::AdminStatsUsageFilter,
) -> Result<Vec<aether_data_contracts::repository::usage::StoredRequestUsageAudit>, GatewayError>
{
crate::handlers::admin::observability::list_usage_for_range(self, time_range, filters).await
}
pub(crate) async fn list_admin_usage_for_optional_range(
&self,
time_range: Option<&crate::handlers::admin::observability::AdminStatsTimeRange>,
filters: &crate::handlers::admin::observability::AdminStatsUsageFilter,
) -> Result<Vec<aether_data_contracts::repository::usage::StoredRequestUsageAudit>, GatewayError>
{
crate::handlers::admin::observability::list_usage_for_optional_range(
self, time_range, filters,
)
.await
}
pub(crate) async fn build_api_format_health_monitor_payload(
&self,
lookback_hours: u64,

19
apps/aether-gateway/src/handlers/admin/request/users.rs
View File

@@ -26,6 +26,15 @@ impl<'a> AdminAppState<'a> {
self.app.resolve_auth_user_summaries_by_ids(user_ids).await
}
pub(crate) async fn search_auth_user_summaries_by_username(
&self,
username_search: &str,
) -> Result<Vec<aether_data::repository::users::StoredUserSummary>, GatewayError> {
self.app
.search_auth_user_summaries_by_username(username_search)
.await
}
pub(crate) async fn list_export_users_page(
&self,
query: &aether_data::repository::users::UserExportListQuery,
@@ -383,6 +392,16 @@ impl<'a> AdminAppState<'a> {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_auth_api_key_export_records_by_name_search(
&self,
name_search: &str,
) -> Result<Vec<aether_data::repository::auth::StoredAuthApiKeyExportRecord>, GatewayError>
{
self.app
.list_auth_api_key_export_records_by_name_search(name_search)
.await
}
pub(crate) async fn list_auth_api_key_export_standalone_records_page(
&self,
query: &aether_data::repository::auth::StandaloneApiKeyExportListQuery,

316
apps/aether-gateway/src/handlers/public/support/dashboard_filters.rs
View File

@@ -2,10 +2,11 @@ use super::{
build_auth_error_response, query_param_value, resolve_authenticated_local_user, AppState,
GatewayError, GatewayPublicRequestContext,
};
use aether_billing::{
normalize_input_tokens_for_billing, normalize_total_input_context_for_cache_hit_rate,
use aether_data_contracts::repository::usage::{
StoredUsageDashboardDailyBreakdownRow, StoredUsageDashboardSummary,
UsageAuditAggregationGroupBy, UsageAuditAggregationQuery, UsageDashboardDailyBreakdownQuery,
UsageDashboardProviderCountsQuery, UsageDashboardSummaryQuery,
};
use aether_data_contracts::repository::usage::{StoredRequestUsageAudit, UsageAuditListQuery};
use axum::{
body::Body,
http,
@@ -38,7 +39,7 @@ struct DashboardUsageTotals {
total_cost_usd: f64,
actual_total_cost_usd: f64,
error_requests: u64,
response_time_sum_ms: u64,
response_time_sum_ms: f64,
response_time_samples: u64,
}
@@ -47,7 +48,7 @@ struct DashboardModelAggregate {
requests: u64,
tokens: u64,
cost: f64,
response_time_sum_ms: u64,
response_time_sum_ms: f64,
response_time_samples: u64,
}
@@ -73,29 +74,32 @@ pub(super) fn decision_route_kind(request_context: &GatewayPublicRequestContext)
}
impl DashboardUsageTotals {
fn record(&mut self, item: &StoredRequestUsageAudit) {
let cache_creation_tokens = dashboard_cache_creation_tokens(item);
self.requests += 1;
self.input_tokens += item.input_tokens;
self.effective_input_tokens += dashboard_effective_input_tokens(item);
self.output_tokens += item.output_tokens;
self.total_tokens += item.total_tokens;
self.cache_creation_tokens += cache_creation_tokens;
self.cache_read_tokens += item.cache_read_input_tokens;
self.cache_hit_total_input_context += dashboard_total_input_context(item);
self.cache_creation_cost_usd += item.cache_creation_cost_usd;
self.cache_read_cost_usd += item.cache_read_cost_usd;
self.total_cost_usd += item.total_cost_usd;
self.actual_total_cost_usd += item.actual_total_cost_usd;
if item.status_code.is_some_and(|value| value >= 400)
|| item.status.eq_ignore_ascii_case("failed")
{
self.error_requests += 1;
}
if let Some(response_time_ms) = item.response_time_ms {
self.response_time_sum_ms += response_time_ms;
self.response_time_samples += 1;
}
fn absorb_summary(&mut self, summary: &StoredUsageDashboardSummary) {
self.requests = self.requests.saturating_add(summary.total_requests);
self.input_tokens = self.input_tokens.saturating_add(summary.input_tokens);
self.effective_input_tokens = self
.effective_input_tokens
.saturating_add(summary.effective_input_tokens);
self.output_tokens = self.output_tokens.saturating_add(summary.output_tokens);
self.total_tokens = self.total_tokens.saturating_add(summary.total_tokens);
self.cache_creation_tokens = self
.cache_creation_tokens
.saturating_add(summary.cache_creation_tokens);
self.cache_read_tokens = self
.cache_read_tokens
.saturating_add(summary.cache_read_tokens);
self.cache_hit_total_input_context = self
.cache_hit_total_input_context
.saturating_add(summary.total_input_context);
self.cache_creation_cost_usd += summary.cache_creation_cost_usd;
self.cache_read_cost_usd += summary.cache_read_cost_usd;
self.total_cost_usd += summary.total_cost_usd;
self.actual_total_cost_usd += summary.actual_total_cost_usd;
self.error_requests = self.error_requests.saturating_add(summary.error_requests);
self.response_time_sum_ms += summary.response_time_sum_ms;
self.response_time_samples = self
.response_time_samples
.saturating_add(summary.response_time_samples);
}
fn avg_response_time_seconds(&self) -> f64 {
@@ -103,7 +107,7 @@ impl DashboardUsageTotals {
0.0
} else {
dashboard_round_f64(
(self.response_time_sum_ms as f64 / self.response_time_samples as f64) / 1000.0,
(self.response_time_sum_ms / self.response_time_samples as f64) / 1000.0,
4,
)
}
@@ -121,49 +125,6 @@ impl DashboardUsageTotals {
}
}
fn dashboard_usage_should_count_in_summary(item: &StoredRequestUsageAudit) -> bool {
!matches!(item.status.as_str(), "pending" | "streaming")
&& !matches!(item.provider_name.as_str(), "unknown" | "pending")
}
fn dashboard_cache_creation_tokens(item: &StoredRequestUsageAudit) -> u64 {
let classified = item
.cache_creation_ephemeral_5m_input_tokens
.saturating_add(item.cache_creation_ephemeral_1h_input_tokens);
if item.cache_creation_input_tokens == 0 && classified > 0 {
classified
} else {
item.cache_creation_input_tokens
}
}
fn dashboard_total_input_context(item: &StoredRequestUsageAudit) -> u64 {
let api_format = item
.endpoint_api_format
.as_deref()
.or(item.api_format.as_deref());
let input_tokens = i64::try_from(item.input_tokens).unwrap_or(i64::MAX);
let cache_creation_tokens =
i64::try_from(dashboard_cache_creation_tokens(item)).unwrap_or(i64::MAX);
let cache_read_tokens = i64::try_from(item.cache_read_input_tokens).unwrap_or(i64::MAX);
normalize_total_input_context_for_cache_hit_rate(
api_format,
input_tokens,
cache_creation_tokens,
cache_read_tokens,
) as u64
}
fn dashboard_effective_input_tokens(item: &StoredRequestUsageAudit) -> u64 {
let api_format = item
.endpoint_api_format
.as_deref()
.or(item.api_format.as_deref());
let input_tokens = i64::try_from(item.input_tokens).unwrap_or(i64::MAX);
let cache_read_tokens = i64::try_from(item.cache_read_input_tokens).unwrap_or(i64::MAX);
normalize_input_tokens_for_billing(api_format, input_tokens, cache_read_tokens) as u64
}
fn dashboard_round_f64(value: f64, decimals: u32) -> f64 {
let factor = 10_f64.powi(i32::try_from(decimals).unwrap_or_default());
(value * factor).round() / factor
@@ -453,21 +414,12 @@ fn dashboard_range_bounds_unix(range: DashboardDateRange) -> Option<(u64, u64)>
Some((start_utc.max(0) as u64, end_utc.max(0) as u64))
}
fn dashboard_usage_local_date(
item: &StoredRequestUsageAudit,
tz_offset_minutes: i32,
) -> Option<chrono::NaiveDate> {
let timestamp = i64::try_from(item.created_at_unix_ms).ok()?;
let datetime = chrono::DateTime::<chrono::Utc>::from_timestamp(timestamp, 0)?;
Some((datetime + chrono::Duration::minutes(i64::from(tz_offset_minutes))).date_naive())
}
async fn dashboard_list_usage_for_range(
async fn dashboard_summary_for_range(
state: &AppState,
range: DashboardDateRange,
user_id: Option<&str>,
error_context: &str,
) -> Result<Vec<StoredRequestUsageAudit>, Response<Body>> {
) -> Result<StoredUsageDashboardSummary, Response<Body>> {
let Some((created_from_unix_secs, created_until_unix_secs)) =
dashboard_range_bounds_unix(range)
else {
@@ -479,26 +431,14 @@ async fn dashboard_list_usage_for_range(
};
match state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(created_from_unix_secs),
created_until_unix_secs: Some(created_until_unix_secs),
.summarize_dashboard_usage(&UsageDashboardSummaryQuery {
created_from_unix_secs,
created_until_unix_secs,
user_id: user_id.map(ToOwned::to_owned),
provider_name: None,
model: None,
api_format: None,
statuses: None,
is_stream: None,
error_only: false,
limit: None,
offset: None,
newest_first: false,
})
.await
{
Ok(mut value) => {
value.retain(dashboard_usage_should_count_in_summary);
Ok(value)
}
Ok(value) => Ok(value),
Err(err) => Err(build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("{error_context}: {err:?}"),
@@ -507,11 +447,45 @@ async fn dashboard_list_usage_for_range(
}
}
fn dashboard_usage_totals(usage: &[StoredRequestUsageAudit]) -> DashboardUsageTotals {
let mut totals = DashboardUsageTotals::default();
for item in usage {
totals.record(item);
async fn dashboard_daily_breakdown_for_range(
state: &AppState,
range: DashboardDateRange,
user_id: Option<&str>,
error_context: &str,
) -> Result<Vec<StoredUsageDashboardDailyBreakdownRow>, Response<Body>> {
let Some((created_from_unix_secs, created_until_unix_secs)) =
dashboard_range_bounds_unix(range)
else {
return Err(build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("{error_context}: invalid time range"),
false,
));
};
match state
.list_dashboard_daily_breakdown(&UsageDashboardDailyBreakdownQuery {
created_from_unix_secs,
created_until_unix_secs,
tz_offset_minutes: range.tz_offset_minutes,
user_id: user_id.map(ToOwned::to_owned),
})
.await
{
Ok(value) => Ok(value),
Err(err) => Err(build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("{error_context}: {err:?}"),
false,
)),
}
}
fn dashboard_usage_totals_from_summary(
summary: &StoredUsageDashboardSummary,
) -> DashboardUsageTotals {
let mut totals = DashboardUsageTotals::default();
totals.absorb_summary(summary);
totals
}
@@ -545,17 +519,28 @@ async fn dashboard_load_api_key_counts(
async fn dashboard_load_user_counts(
state: &AppState,
fallback_user_ids: &[String],
range: DashboardDateRange,
) -> Result<(u64, u64), GatewayError> {
let summary = state.summarize_export_users().await?;
if summary.total > 0 {
return Ok((summary.total, summary.active));
}
let unique = fallback_user_ids
.iter()
.cloned()
.collect::<std::collections::BTreeSet<_>>();
Ok((unique.len() as u64, unique.len() as u64))
let Some((created_from_unix_secs, created_until_unix_secs)) =
dashboard_range_bounds_unix(range)
else {
return Ok((0, 0));
};
let fallback = state
.aggregate_usage_audits(&UsageAuditAggregationQuery {
created_from_unix_secs,
created_until_unix_secs,
group_by: UsageAuditAggregationGroupBy::User,
limit: 10_000,
})
.await?;
let count = fallback.len() as u64;
Ok((count, count))
}
pub(super) async fn handle_dashboard_stats_get(
@@ -605,7 +590,7 @@ pub(super) async fn handle_dashboard_stats_get(
tz_offset_minutes: summary_range.tz_offset_minutes,
};
let user_filter = (!is_admin).then_some(auth.user.id.as_str());
let period_usage = match dashboard_list_usage_for_range(
let period_summary = match dashboard_summary_for_range(
state,
summary_range,
user_filter,
@@ -616,7 +601,7 @@ pub(super) async fn handle_dashboard_stats_get(
Ok(value) => value,
Err(response) => return response,
};
let today_usage = match dashboard_list_usage_for_range(
let today_summary = match dashboard_summary_for_range(
state,
today_range,
user_filter,
@@ -628,14 +613,8 @@ pub(super) async fn handle_dashboard_stats_get(
Err(response) => return response,
};
let period_totals = dashboard_usage_totals(&period_usage);
let today_totals = dashboard_usage_totals(&today_usage);
let fallback_user_ids = period_usage
.iter()
.filter_map(|item| item.user_id.clone())
.collect::<std::collections::BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
let period_totals = dashboard_usage_totals_from_summary(&period_summary);
let today_totals = dashboard_usage_totals_from_summary(&today_summary);
let api_key_counts = match dashboard_load_api_key_counts(state, is_admin, &auth.user.id).await {
Ok(value) => value,
@@ -673,7 +652,7 @@ pub(super) async fn handle_dashboard_stats_get(
if is_admin {
let (total_users, active_users) =
match dashboard_load_user_counts(state, &fallback_user_ids).await {
match dashboard_load_user_counts(state, summary_range).await {
Ok(value) => value,
Err(err) => {
return build_auth_error_response(
@@ -834,25 +813,33 @@ pub(super) async fn handle_dashboard_stats_get(
fn dashboard_daily_aggregate_record(
aggregate: &mut DashboardDailyAggregate,
item: &StoredRequestUsageAudit,
row: &StoredUsageDashboardDailyBreakdownRow,
) {
aggregate.totals.record(item);
let model = aggregate.models.entry(item.model.clone()).or_default();
model.requests += 1;
model.tokens += item.total_tokens;
model.cost += item.total_cost_usd;
if let Some(response_time_ms) = item.response_time_ms {
model.response_time_sum_ms += response_time_ms;
model.response_time_samples += 1;
}
aggregate.totals.requests = aggregate.totals.requests.saturating_add(row.requests);
aggregate.totals.total_tokens = aggregate
.totals
.total_tokens
.saturating_add(row.total_tokens);
aggregate.totals.total_cost_usd += row.total_cost_usd;
aggregate.totals.response_time_sum_ms += row.response_time_sum_ms;
aggregate.totals.response_time_samples = aggregate
.totals
.response_time_samples
.saturating_add(row.response_time_samples);
let provider = aggregate
.providers
.entry(item.provider_name.clone())
.or_default();
provider.requests += 1;
provider.tokens += item.total_tokens;
provider.cost += item.total_cost_usd;
let model = aggregate.models.entry(row.model.clone()).or_default();
model.requests = model.requests.saturating_add(row.requests);
model.tokens = model.tokens.saturating_add(row.total_tokens);
model.cost += row.total_cost_usd;
model.response_time_sum_ms += row.response_time_sum_ms;
model.response_time_samples = model
.response_time_samples
.saturating_add(row.response_time_samples);
let provider = aggregate.providers.entry(row.provider.clone()).or_default();
provider.requests = provider.requests.saturating_add(row.requests);
provider.tokens = provider.tokens.saturating_add(row.total_tokens);
provider.cost += row.total_cost_usd;
}
pub(super) async fn handle_dashboard_daily_stats_get(
@@ -896,7 +883,7 @@ pub(super) async fn handle_dashboard_daily_stats_get(
Err(detail) => return dashboard_bad_request_response(detail),
};
let user_filter = (!is_admin).then_some(auth.user.id.as_str());
let usage = match dashboard_list_usage_for_range(
let usage = match dashboard_daily_breakdown_for_range(
state,
range,
user_filter,
@@ -914,28 +901,26 @@ pub(super) async fn handle_dashboard_daily_stats_get(
let mut provider_summary =
std::collections::BTreeMap::<String, DashboardProviderAggregate>::new();
for item in &usage {
let Some(date) = dashboard_usage_local_date(item, range.tz_offset_minutes) else {
for row in &usage {
let Ok(date) = chrono::NaiveDate::parse_from_str(&row.date, "%Y-%m-%d") else {
continue;
};
let aggregate = by_date.entry(date).or_default();
dashboard_daily_aggregate_record(aggregate, item);
dashboard_daily_aggregate_record(aggregate, row);
let model = model_summary.entry(item.model.clone()).or_default();
model.requests += 1;
model.tokens += item.total_tokens;
model.cost += item.total_cost_usd;
if let Some(response_time_ms) = item.response_time_ms {
model.response_time_sum_ms += response_time_ms;
model.response_time_samples += 1;
}
let model = model_summary.entry(row.model.clone()).or_default();
model.requests = model.requests.saturating_add(row.requests);
model.tokens = model.tokens.saturating_add(row.total_tokens);
model.cost += row.total_cost_usd;
model.response_time_sum_ms += row.response_time_sum_ms;
model.response_time_samples = model
.response_time_samples
.saturating_add(row.response_time_samples);
let provider = provider_summary
.entry(item.provider_name.clone())
.or_default();
provider.requests += 1;
provider.tokens += item.total_tokens;
provider.cost += item.total_cost_usd;
let provider = provider_summary.entry(row.provider.clone()).or_default();
provider.requests = provider.requests.saturating_add(row.requests);
provider.tokens = provider.tokens.saturating_add(row.total_tokens);
provider.cost += row.total_cost_usd;
}
let mut daily_stats = Vec::new();
@@ -1009,8 +994,7 @@ pub(super) async fn handle_dashboard_daily_stats_get(
0.0
} else {
dashboard_round_f64(
(value.response_time_sum_ms as f64 / value.response_time_samples as f64)
/ 1000.0,
(value.response_time_sum_ms / value.response_time_samples as f64) / 1000.0,
4,
)
};
@@ -1213,20 +1197,12 @@ pub(super) async fn handle_dashboard_provider_status_get(
let since_unix_secs = u64::try_from(chrono::Utc::now().timestamp())
.unwrap_or_default()
.saturating_sub(24 * 3600);
let now_unix_secs = u64::try_from(chrono::Utc::now().timestamp()).unwrap_or_default();
let usage = match state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(since_unix_secs),
created_until_unix_secs: None,
.summarize_dashboard_provider_counts(&UsageDashboardProviderCountsQuery {
created_from_unix_secs: since_unix_secs,
created_until_unix_secs: now_unix_secs,
user_id: None,
provider_name: None,
model: None,
api_format: None,
statuses: None,
is_stream: None,
error_only: false,
limit: None,
offset: None,
newest_first: false,
})
.await
{
@@ -1244,7 +1220,7 @@ pub(super) async fn handle_dashboard_provider_status_get(
for item in usage {
*request_counts
.entry(item.provider_name.to_ascii_lowercase())
.or_default() += 1;
.or_default() += item.request_count;
}
let mut entries = providers

4
apps/aether-gateway/src/handlers/public/support/user_me.rs
View File

@@ -8,8 +8,8 @@ use super::{
use crate::admin_api::build_admin_endpoint_health_status_payload;
use crate::handlers::internal::build_management_token_payload;
use crate::handlers::shared::{
admin_stats_bad_request_response, list_usage_for_optional_range, parse_bounded_u32, round_to,
AdminStatsTimeRange, AdminStatsUsageFilter,
admin_stats_bad_request_response, parse_bounded_u32, round_to, AdminStatsTimeRange,
AdminStatsUsageFilter,
};
const USERS_ME_AVAILABLE_MODELS_FETCH_LIMIT: usize = 1000;

1015
apps/aether-gateway/src/handlers/public/support/user_me_usage.rs
View File

File diff suppressed because it is too large Load Diff

59
apps/aether-gateway/src/handlers/public/support/wallet/reads.rs
View File

@@ -4,7 +4,7 @@ use super::{
GatewayError, GatewayPublicRequestContext, Response, WALLET_LEGACY_TIMEZONE,
};
use crate::handlers::shared::round_to;
use aether_data_contracts::repository::usage::UsageAuditListQuery;
use aether_data_contracts::repository::usage::UsageSettledCostSummaryQuery;
use chrono::Utc;
use serde_json::json;
@@ -210,20 +210,11 @@ pub(super) async fn handle_wallet_today_cost(
.unwrap_or_default();
let end_unix_secs = start_unix_secs.saturating_add(24 * 3600);
let items = match state
.list_usage_audits(&UsageAuditListQuery {
created_from_unix_secs: Some(start_unix_secs),
created_until_unix_secs: Some(end_unix_secs),
let summary = match state
.summarize_usage_settled_cost(&UsageSettledCostSummaryQuery {
created_from_unix_secs: start_unix_secs,
created_until_unix_secs: end_unix_secs,
user_id: Some(auth.user.id.clone()),
provider_name: None,
model: None,
api_format: None,
statuses: None,
is_stream: None,
error_only: false,
limit: None,
offset: None,
newest_first: false,
})
.await
{
@@ -237,31 +228,11 @@ pub(super) async fn handle_wallet_today_cost(
}
};
let settled = items
.into_iter()
.filter(|item| item.billing_status == "settled" && item.total_cost_usd > 0.0)
.collect::<Vec<_>>();
let total_cost = settled.iter().map(|item| item.total_cost_usd).sum::<f64>();
let total_requests = settled.len() as u64;
let input_tokens = settled.iter().map(|item| item.input_tokens).sum::<u64>();
let output_tokens = settled.iter().map(|item| item.output_tokens).sum::<u64>();
let cache_creation_tokens = settled
.iter()
.map(|item| item.cache_creation_input_tokens)
.sum::<u64>();
let cache_read_tokens = settled
.iter()
.map(|item| item.cache_read_input_tokens)
.sum::<u64>();
let first_finalized_at = settled
.iter()
.filter_map(|item| item.finalized_at_unix_secs)
.min()
let first_finalized_at = summary
.first_finalized_at_unix_secs
.and_then(unix_secs_to_rfc3339);
let last_finalized_at = settled
.iter()
.filter_map(|item| item.finalized_at_unix_secs)
.max()
let last_finalized_at = summary
.last_finalized_at_unix_secs
.and_then(unix_secs_to_rfc3339);
build_auth_json_response(
@@ -270,12 +241,12 @@ pub(super) async fn handle_wallet_today_cost(
"id": serde_json::Value::Null,
"date": today.to_string(),
"timezone": "UTC",
"total_cost": round_to(total_cost, 6),
"total_requests": total_requests,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cache_creation_tokens": cache_creation_tokens,
"cache_read_tokens": cache_read_tokens,
"total_cost": round_to(summary.total_cost_usd, 6),
"total_requests": summary.total_requests,
"input_tokens": summary.input_tokens,
"output_tokens": summary.output_tokens,
"cache_creation_tokens": summary.cache_creation_tokens,
"cache_read_tokens": summary.cache_read_tokens,
"first_finalized_at": first_finalized_at,
"last_finalized_at": last_finalized_at,
"aggregated_at": Utc::now().to_rfc3339(),

4
apps/aether-gateway/src/handlers/shared/mod.rs
View File

@@ -51,6 +51,6 @@ pub(crate) use self::system_config_values::{
module_available_from_env, system_config_bool, system_config_string,
};
pub(crate) use self::usage_stats::{
admin_stats_bad_request_response, list_usage_for_optional_range, parse_bounded_u32, round_to,
AdminStatsTimeRange, AdminStatsUsageFilter,
admin_stats_bad_request_response, parse_bounded_u32, round_to, AdminStatsTimeRange,
AdminStatsUsageFilter,
};

4
apps/aether-gateway/src/handlers/shared/usage_stats.rs
View File

@@ -1,4 +1,4 @@
pub(crate) use crate::admin_api::{
admin_stats_bad_request_response, list_usage_for_optional_range, parse_bounded_u32, round_to,
AdminStatsTimeRange, AdminStatsUsageFilter,
admin_stats_bad_request_response, parse_bounded_u32, round_to, AdminStatsTimeRange,
AdminStatsUsageFilter,
};

11
apps/aether-gateway/src/state/runtime/api_key_exports.rs
View File

@@ -36,6 +36,17 @@ impl AppState {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_auth_api_key_export_records_by_name_search(
&self,
name_search: &str,
) -> Result<Vec<aether_data::repository::auth::StoredAuthApiKeyExportRecord>, GatewayError>
{
self.data
.list_auth_api_key_export_records_by_name_search(name_search)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_auth_api_key_export_standalone_records_page(
&self,
query: &aether_data::repository::auth::StandaloneApiKeyExportListQuery,

9
apps/aether-gateway/src/state/runtime/auth/api_keys.rs
View File

@@ -174,6 +174,15 @@ mod tests {
self.lookup.list_export_api_keys_by_ids(api_key_ids).await
}
async fn list_export_api_keys_by_name_search(
&self,
name_search: &str,
) -> Result<Vec<StoredAuthApiKeyExportRecord>, aether_data::DataLayerError> {
self.lookup
.list_export_api_keys_by_name_search(name_search)
.await
}
async fn list_export_standalone_api_keys_page(
&self,
query: &StandaloneApiKeyExportListQuery,

41
apps/aether-gateway/src/state/runtime/auth/user_lifecycle.rs
View File

@@ -43,6 +43,47 @@ impl AppState {
Ok(users)
}
pub(crate) async fn search_auth_user_summaries_by_username(
&self,
username_search: &str,
) -> Result<Vec<aether_data::repository::users::StoredUserSummary>, GatewayError> {
let username_search = username_search.trim();
if username_search.is_empty() {
return Ok(Vec::new());
}
let mut users = BTreeMap::new();
if self.has_user_data_reader() {
for user in self
.data
.list_users_by_username_search(username_search)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?
{
users.insert(user.id.clone(), user);
}
}
#[cfg(test)]
if let Some(store) = self.auth_user_store.as_ref() {
let username_search = username_search.to_ascii_lowercase();
for user in store.lock().expect("auth user store should lock").values() {
if user
.username
.to_ascii_lowercase()
.contains(&username_search)
{
let summary = user
.to_summary()
.map_err(|err| GatewayError::Internal(err.to_string()))?;
users.entry(summary.id.clone()).or_insert(summary);
}
}
}
Ok(users.into_values().collect())
}
pub(crate) async fn find_user_auth_by_id(
&self,
user_id: &str,

170
apps/aether-gateway/src/state/runtime/usage_queries.rs
View File

@@ -35,6 +35,26 @@ impl AppState {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn find_request_usage_by_id(
&self,
usage_id: &str,
) -> Result<Option<usage::StoredRequestUsageAudit>, GatewayError> {
self.data
.find_request_usage_by_id(usage_id)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_request_usage_by_ids(
&self,
usage_ids: &[String],
) -> Result<Vec<usage::StoredRequestUsageAudit>, GatewayError> {
self.data
.list_request_usage_by_ids(usage_ids)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_usage_audits(
&self,
query: &usage::UsageAuditListQuery,
@@ -55,6 +75,26 @@ impl AppState {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_usage_audits_by_keyword_search(
&self,
query: &usage::UsageAuditKeywordSearchQuery,
) -> Result<Vec<usage::StoredRequestUsageAudit>, GatewayError> {
self.data
.list_usage_audits_by_keyword_search(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn count_usage_audits_by_keyword_search(
&self,
query: &usage::UsageAuditKeywordSearchQuery,
) -> Result<u64, GatewayError> {
self.data
.count_usage_audits_by_keyword_search(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn aggregate_usage_audits(
&self,
query: &usage::UsageAuditAggregationQuery,
@@ -75,6 +115,136 @@ impl AppState {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_cache_hit_summary(
&self,
query: &usage::UsageCacheHitSummaryQuery,
) -> Result<usage::StoredUsageCacheHitSummary, GatewayError> {
self.data
.summarize_usage_cache_hit_summary(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_settled_cost(
&self,
query: &usage::UsageSettledCostSummaryQuery,
) -> Result<usage::StoredUsageSettledCostSummary, GatewayError> {
self.data
.summarize_usage_settled_cost(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_cache_affinity_hit_summary(
&self,
query: &usage::UsageCacheAffinityHitSummaryQuery,
) -> Result<usage::StoredUsageCacheAffinityHitSummary, GatewayError> {
self.data
.summarize_usage_cache_affinity_hit_summary(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_usage_cache_affinity_intervals(
&self,
query: &usage::UsageCacheAffinityIntervalQuery,
) -> Result<Vec<usage::StoredUsageCacheAffinityIntervalRow>, GatewayError> {
self.data
.list_usage_cache_affinity_intervals(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_dashboard_usage(
&self,
query: &usage::UsageDashboardSummaryQuery,
) -> Result<usage::StoredUsageDashboardSummary, GatewayError> {
self.data
.summarize_dashboard_usage(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_dashboard_daily_breakdown(
&self,
query: &usage::UsageDashboardDailyBreakdownQuery,
) -> Result<Vec<usage::StoredUsageDashboardDailyBreakdownRow>, GatewayError> {
self.data
.list_dashboard_daily_breakdown(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_dashboard_provider_counts(
&self,
query: &usage::UsageDashboardProviderCountsQuery,
) -> Result<Vec<usage::StoredUsageDashboardProviderCount>, GatewayError> {
self.data
.summarize_dashboard_provider_counts(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_breakdown(
&self,
query: &usage::UsageBreakdownSummaryQuery,
) -> Result<Vec<usage::StoredUsageBreakdownSummaryRow>, GatewayError> {
self.data
.summarize_usage_breakdown(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn count_monitoring_usage_errors(
&self,
query: &usage::UsageMonitoringErrorCountQuery,
) -> Result<u64, GatewayError> {
self.data
.count_monitoring_usage_errors(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_monitoring_usage_errors(
&self,
query: &usage::UsageMonitoringErrorListQuery,
) -> Result<Vec<usage::StoredRequestUsageAudit>, GatewayError> {
self.data
.list_monitoring_usage_errors(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_error_distribution(
&self,
query: &usage::UsageErrorDistributionQuery,
) -> Result<Vec<usage::StoredUsageErrorDistributionRow>, GatewayError> {
self.data
.summarize_usage_error_distribution(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_performance_percentiles(
&self,
query: &usage::UsagePerformancePercentilesQuery,
) -> Result<Vec<usage::StoredUsagePerformancePercentilesRow>, GatewayError> {
self.data
.summarize_usage_performance_percentiles(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_cost_savings(
&self,
query: &usage::UsageCostSavingsSummaryQuery,
) -> Result<usage::StoredUsageCostSavingsSummary, GatewayError> {
self.data
.summarize_usage_cost_savings(query)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_time_series(
&self,
query: &usage::UsageTimeSeriesQuery,

34
apps/aether-gateway/src/tests/architecture/admin_observability.rs
View File

@@ -29,7 +29,6 @@ fn non_admin_handlers_do_not_depend_on_admin_stats_module() {
let shared_mod = read_workspace_file("apps/aether-gateway/src/handlers/shared/mod.rs");
for pattern in [
"admin_stats_bad_request_response",
"list_usage_for_optional_range",
"parse_bounded_u32",
"round_to",
"AdminStatsTimeRange",
@@ -40,6 +39,10 @@ fn non_admin_handlers_do_not_depend_on_admin_stats_module() {
"handlers/shared/mod.rs should expose shared usage stats helper {pattern}"
);
}
assert!(
!shared_mod.contains("list_usage_for_optional_range"),
"handlers/shared/mod.rs should not expose unbounded usage range helpers"
);
let admin_observability_mod =
read_workspace_file("apps/aether-gateway/src/handlers/admin/observability/mod.rs");
@@ -57,7 +60,6 @@ fn non_admin_handlers_do_not_depend_on_admin_stats_module() {
}
for pattern in [
"admin_stats_bad_request_response",
"list_usage_for_optional_range",
"parse_bounded_u32",
"round_to",
"AdminStatsTimeRange",
@@ -68,6 +70,12 @@ fn non_admin_handlers_do_not_depend_on_admin_stats_module() {
"handlers/admin/observability/mod.rs should expose admin stats facade helper {pattern}"
);
}
for pattern in ["list_usage_for_optional_range", "list_usage_for_range"] {
assert!(
!admin_observability_mod.contains(pattern),
"handlers/admin/observability/mod.rs should not re-export deprecated usage helper {pattern}"
);
}
let shared_usage_stats =
read_workspace_file("apps/aether-gateway/src/handlers/shared/usage_stats.rs");
@@ -193,14 +201,15 @@ fn admin_stats_root_stays_thin() {
stats_mod.contains("pub(crate) use self::range::{"),
"handlers/admin/observability/stats/mod.rs should re-export the split range seam"
);
for pattern in [
"list_usage_for_optional_range",
"list_usage_for_range",
"parse_bounded_u32",
] {
let pattern = "parse_bounded_u32";
assert!(
stats_mod.contains(pattern),
"handlers/admin/observability/stats/mod.rs should keep range re-export {pattern}"
);
for pattern in ["list_usage_for_optional_range", "list_usage_for_range"] {
assert!(
stats_mod.contains(pattern),
"handlers/admin/observability/stats/mod.rs should keep range re-export {pattern}"
!stats_mod.contains(pattern),
"handlers/admin/observability/stats/mod.rs should not re-export deprecated range helper {pattern}"
);
}
@@ -471,7 +480,7 @@ fn admin_usage_root_stays_thin() {
"mod cache_affinity;",
"mod filters;",
"pub(super) use aggregations::admin_usage_aggregation_by_user_json;",
"pub(super) use cache_affinity::list_recent_completed_usage_for_cache_affinity;",
"pub(super) use cache_affinity::list_usage_cache_affinity_intervals;",
"pub(super) use filters::admin_usage_provider_key_names;",
] {
assert!(
@@ -574,9 +583,8 @@ fn admin_usage_root_stays_thin() {
"apps/aether-gateway/src/handlers/admin/observability/usage/analytics/cache_affinity.rs",
);
assert!(
analytics_cache_affinity.contains(
"pub(in super::super) async fn list_recent_completed_usage_for_cache_affinity("
),
analytics_cache_affinity
.contains("pub(in super::super) async fn list_usage_cache_affinity_intervals("),
"usage/analytics/cache_affinity.rs should own cache-affinity analytics helpers"
);
let analytics_filters = read_workspace_file(

11
apps/aether-gateway/src/tests/architecture/admin_shared.rs
View File

@@ -161,8 +161,6 @@ fn admin_wrapped_state_owns_observability_capabilities() {
"pub(crate) fn has_auth_api_key_data_reader(&self) -> bool",
"pub(crate) fn has_user_data_reader(&self) -> bool",
"pub(crate) async fn list_provider_catalog_providers(",
"pub(crate) async fn list_admin_usage_for_range(",
"pub(crate) async fn list_admin_usage_for_optional_range(",
"pub(crate) async fn aggregate_finalized_request_candidate_timeline_by_endpoint_ids_since(",
"pub(crate) async fn read_recent_request_candidates(",
"pub(crate) fn provider_key_rpm_reset_at(",
@@ -175,6 +173,15 @@ fn admin_wrapped_state_owns_observability_capabilities() {
"handlers/admin/request/mod.rs should expose observability capability {pattern}"
);
}
for pattern in [
"pub(crate) async fn list_admin_usage_for_range(",
"pub(crate) async fn list_admin_usage_for_optional_range(",
] {
assert!(
!admin_request.contains(pattern),
"handlers/admin/request/mod.rs should not expose deprecated unbounded usage helper {pattern}"
);
}
}
#[test]

9
apps/aether-gateway/src/tests/control/admin/stats.rs
View File

@@ -196,6 +196,15 @@ impl AuthApiKeyReadRepository for PartialListAuthApiKeyRepository {
self.lookup.list_export_api_keys_by_ids(api_key_ids).await
}
async fn list_export_api_keys_by_name_search(
&self,
name_search: &str,
) -> Result<Vec<StoredAuthApiKeyExportRecord>, aether_data::DataLayerError> {
self.lookup
.list_export_api_keys_by_name_search(name_search)
.await
}
async fn list_export_standalone_api_keys_page(
&self,
query: &StandaloneApiKeyExportListQuery,

108
apps/aether-gateway/src/tests/frontdoor/public_support.rs
View File

@@ -4916,6 +4916,114 @@ async fn gateway_handles_users_me_usage_without_legacy_api_key_name_fallback_whe
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_handles_users_me_usage_search_with_model_and_api_key_keywords() {
let now = Utc::now();
let user = sample_auth_user(now);
let access_token = build_test_auth_token(
"access",
serde_json::Map::from_iter([
("user_id".to_string(), json!(user.id)),
("role".to_string(), json!(user.role)),
(
"created_at".to_string(),
json!(user.created_at.map(|value| value.to_rfc3339())),
),
(
"session_id".to_string(),
json!("session-users-me-usage-multi-keyword-search"),
),
]),
now + chrono::Duration::hours(1),
);
let usage_repository = Arc::new(InMemoryUsageReadRepository::seed(vec![
sample_user_usage_audit(
"usage-users-me-completed-multi-keyword-search",
"req-users-me-completed-multi-keyword-search",
"user-auth-1",
"gpt-4.1",
"OpenAI",
"completed",
now - chrono::Duration::minutes(20),
),
sample_user_usage_audit(
"usage-users-me-failed-multi-keyword-search",
"req-users-me-failed-multi-keyword-search",
"user-auth-1",
"claude-3.5-sonnet",
"Anthropic",
"failed",
now - chrono::Duration::minutes(10),
),
sample_user_usage_audit(
"usage-users-me-streaming-multi-keyword-search",
"req-users-me-streaming-multi-keyword-search",
"user-auth-1",
"gpt-4.1-mini",
"OpenAI",
"streaming",
now - chrono::Duration::minutes(5),
),
]));
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some("hash-api-key-user-1".to_string()),
sample_usage_auth_snapshot("api-key-user-1", "user-auth-1", "renamed-key"),
)]));
let (gateway_url, upstream_hits, gateway_handle, upstream_handle) =
start_auth_gateway_with_builder(|| {
let data_state = crate::data::GatewayDataState::with_user_wallet_and_usage_for_tests(
Arc::new(InMemoryUserReadRepository::seed_auth_users(vec![
user.clone()
])),
Arc::new(InMemoryWalletRepository::seed(vec![sample_auth_wallet(
"user-auth-1",
now,
)])),
Arc::clone(&usage_repository),
)
.with_auth_api_key_reader(auth_repository);
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(data_state)
.with_auth_sessions_for_tests([sample_auth_session(
"user-auth-1",
"session-users-me-usage-multi-keyword-search",
"device-users-me-usage-multi-keyword-search",
"refresh-token-users-me-usage-multi-keyword-search",
now,
)])
})
.await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/users/me/usage?limit=10&offset=0&search=gpt%20renamed-key"
))
.header("authorization", format!("Bearer {access_token}"))
.header(
"x-client-device-id",
"device-users-me-usage-multi-keyword-search",
)
.header("user-agent", "AetherTest/1.0")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["pagination"]["total"], 2);
assert_eq!(
payload["records"].as_array().expect("records array").len(),
2
);
assert_eq!(payload["records"][0]["model"], "gpt-4.1-mini");
assert_eq!(payload["records"][1]["model"], "gpt-4.1");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_handles_users_me_usage_active_locally_without_proxying_upstream() {
let now = Utc::now();