diff --git a/apps/aether-gateway/src/handlers/admin/observability/mod.rs b/apps/aether-gateway/src/handlers/admin/observability/mod.rs index 5ca53be1b..0d81bd6c0 100644 --- a/apps/aether-gateway/src/handlers/admin/observability/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/observability/mod.rs @@ -31,6 +31,16 @@ pub(crate) async fn resolve_usage_user_group_scope( return Ok(Err("user group data is unavailable".to_string())); } + if group_id == UNGROUPED_USAGE_ID { + let ids = ungrouped_usage_users(state) + .await? + .into_iter() + .filter(|user| include_inactive || user.is_active) + .filter(|user| !exclude_admin || !user.role.eq_ignore_ascii_case("admin")) + .map(|user| user.id) + .collect(); + return Ok(Ok(Some(ids))); + } match state .resolve_usage_user_group_member_ids(&group_id, include_inactive, exclude_admin) .await? @@ -39,3 +49,100 @@ pub(crate) async fn resolve_usage_user_group_scope( None => Ok(Err("user_group_id does not exist".to_string())), } } + +/// Reserved statistics-only scope; never a permission group. +pub(crate) const UNGROUPED_USAGE_ID: &str = "__ungrouped__"; + +pub(crate) async fn ungrouped_usage_users( + state: &crate::handlers::admin::request::AdminAppState<'_>, +) -> Result, crate::GatewayError> { + use aether_data::repository::users::UserExportListQuery; + let mut users = Vec::new(); + let mut skip = 0; + loop { + let page = state + .list_export_users_page(&UserExportListQuery { + skip, + limit: 500, + ..Default::default() + }) + .await?; + let count = page.len(); + if count == 0 { + break; + } + let ids = page.into_iter().map(|user| user.id).collect::>(); + let grouped = state + .list_user_group_memberships_by_user_ids(&ids) + .await? + .into_iter() + .map(|membership| membership.user_id) + .collect::>(); + let ids = ids + .into_iter() + .filter(|id| !grouped.contains(id)) + .collect::>(); + users.extend( + state + .list_users_by_ids(&ids) + .await? + .into_iter() + .filter(|user| !user.is_deleted), + ); + skip += count; + if count < 500 { + break; + } + } + Ok(users) +} + +/// Current group provider policy, resolved to the provider-name dimension used by usage rollups. +/// None is unrestricted; Some(empty) deliberately matches no usage. +pub(crate) async fn usage_group_provider_names( + state: &crate::handlers::admin::request::AdminAppState<'_>, + group: &aether_data::repository::users::StoredUserGroup, +) -> Result>, crate::GatewayError> { + if matches!( + group.allowed_providers_mode.as_str(), + "unrestricted" | "inherit" + ) { + return Ok(None); + } + if group.allowed_providers_mode != "specific" { + return Ok(Some(Vec::new())); + } + let allowed = group.allowed_providers.as_deref().unwrap_or_default(); + let providers = state.list_provider_catalog_providers(false).await?; + let mut names = providers + .into_iter() + .filter(|provider| { + allowed.iter().any(|value| { + let value = value.trim(); + value.eq_ignore_ascii_case(&provider.id) + || value.eq_ignore_ascii_case(&provider.name) + || value.eq_ignore_ascii_case(&provider.provider_type) + }) + }) + .map(|provider| provider.name) + .collect::>(); + names.sort(); + names.dedup(); + Ok(Some(names)) +} + +pub(crate) async fn resolve_usage_group_provider_names( + state: &crate::handlers::admin::request::AdminAppState<'_>, + query: Option<&str>, +) -> Result>, crate::GatewayError> { + let Some(id) = crate::handlers::admin::shared::query_param_value(query, "user_group_id") else { + return Ok(None); + }; + if id == UNGROUPED_USAGE_ID { + return Ok(None); + } + let Some(group) = state.find_user_group_by_id(&id).await? else { + return Ok(Some(Vec::new())); + }; + usage_group_provider_names(state, &group).await +} diff --git a/apps/aether-gateway/src/handlers/admin/observability/monitoring/activity.rs b/apps/aether-gateway/src/handlers/admin/observability/monitoring/activity.rs index a7504cb03..4e590277e 100644 --- a/apps/aether-gateway/src/handlers/admin/observability/monitoring/activity.rs +++ b/apps/aether-gateway/src/handlers/admin/observability/monitoring/activity.rs @@ -169,6 +169,7 @@ pub(super) async fn build_admin_monitoring_system_status_response( let today_usage = state .summarize_usage_audits(&UsageAuditSummaryQuery { + provider_names: None, created_from_unix_secs: today_start.timestamp().max(0) as u64, created_until_unix_secs: now_unix_secs.saturating_add(1), user_id: None, diff --git a/apps/aether-gateway/src/handlers/admin/observability/stats/analytics_routes.rs b/apps/aether-gateway/src/handlers/admin/observability/stats/analytics_routes.rs index 7f0af1fe5..a039afdcd 100644 --- a/apps/aether-gateway/src/handlers/admin/observability/stats/analytics_routes.rs +++ b/apps/aether-gateway/src/handlers/admin/observability/stats/analytics_routes.rs @@ -110,6 +110,7 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response( }; let current_summary = state .summarize_usage_audits(&UsageAuditSummaryQuery { + provider_names: None, created_from_unix_secs: current_from_unix_secs, created_until_unix_secs: current_until_unix_secs, ..Default::default() @@ -117,6 +118,7 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response( .await?; let comparison_summary = state .summarize_usage_audits(&UsageAuditSummaryQuery { + provider_names: None, created_from_unix_secs: comparison_from_unix_secs, created_until_unix_secs: comparison_until_unix_secs, ..Default::default() @@ -318,6 +320,11 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response( }; let buckets = state .summarize_usage_time_series(&UsageTimeSeriesQuery { + provider_names: super::super::resolve_usage_group_provider_names( + state, + request_context.query_string(), + ) + .await?, created_from_unix_secs, created_until_unix_secs, granularity: query_granularity, diff --git a/apps/aether-gateway/src/handlers/admin/observability/stats/cost_routes.rs b/apps/aether-gateway/src/handlers/admin/observability/stats/cost_routes.rs index 6e6c2f34a..7867a1b2f 100644 --- a/apps/aether-gateway/src/handlers/admin/observability/stats/cost_routes.rs +++ b/apps/aether-gateway/src/handlers/admin/observability/stats/cost_routes.rs @@ -72,6 +72,7 @@ pub(super) async fn maybe_build_local_admin_stats_cost_response( }; let buckets = state .summarize_usage_time_series(&UsageTimeSeriesQuery { + provider_names: None, created_from_unix_secs, created_until_unix_secs, granularity: UsageTimeSeriesGranularity::Day, diff --git a/apps/aether-gateway/src/handlers/admin/observability/stats/leaderboard_routes.rs b/apps/aether-gateway/src/handlers/admin/observability/stats/leaderboard_routes.rs index 85b2cf4d3..49cd8a1a8 100644 --- a/apps/aether-gateway/src/handlers/admin/observability/stats/leaderboard_routes.rs +++ b/apps/aether-gateway/src/handlers/admin/observability/stats/leaderboard_routes.rs @@ -78,6 +78,7 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response( }; let summaries = state .summarize_usage_leaderboard(&UsageLeaderboardQuery { + provider_names: None, created_from_unix_secs, created_until_unix_secs, group_by: UsageLeaderboardGroupBy::Model, @@ -156,6 +157,7 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response( }; let summaries = state .summarize_usage_leaderboard(&UsageLeaderboardQuery { + provider_names: None, created_from_unix_secs, created_until_unix_secs, group_by: UsageLeaderboardGroupBy::ApiKey, @@ -283,34 +285,6 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response( ))); }; - let summaries = state - .summarize_usage_leaderboard(&UsageLeaderboardQuery { - created_from_unix_secs, - created_until_unix_secs, - group_by: UsageLeaderboardGroupBy::User, - user_id: None, - user_ids: None, - provider_name: filters.provider_name, - model: filters.model, - }) - .await?; - let user_ids = summaries - .iter() - .map(|item| item.group_key.clone()) - .collect::>(); - let user_metadata = load_user_leaderboard_metadata(state, &user_ids).await?; - let user_usage = build_user_leaderboard_items_from_summaries( - &summaries, - &user_metadata, - state.has_auth_user_data_reader(), - state.has_user_data_reader(), - include_inactive, - exclude_admin, - ) - .into_iter() - .map(|item| (item.id.clone(), item)) - .collect::>(); - let mut leaderboard = Vec::new(); let mut member_counts = BTreeMap::new(); let mut active_member_counts = BTreeMap::new(); @@ -321,13 +295,38 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response( .iter() .filter(|member| !member.is_deleted && member.is_active) .count(); - let scoped_user_ids = members + let user_ids = members .iter() .filter(|member| !member.is_deleted) .filter(|member| include_inactive || member.is_active) .filter(|member| !exclude_admin || !member.role.eq_ignore_ascii_case("admin")) - .map(|member| member.user_id.as_str()) - .collect::>(); + .map(|member| member.user_id.clone()) + .collect::>(); + let summaries = state + .summarize_usage_leaderboard(&UsageLeaderboardQuery { + created_from_unix_secs, + created_until_unix_secs, + group_by: UsageLeaderboardGroupBy::User, + user_id: None, + user_ids: Some(user_ids), + provider_names: super::super::usage_group_provider_names(state, &group).await?, + provider_name: filters.provider_name.clone(), + model: filters.model.clone(), + }) + .await?; + let user_ids = summaries + .iter() + .map(|row| row.group_key.clone()) + .collect::>(); + let metadata = load_user_leaderboard_metadata(state, &user_ids).await?; + let users = build_user_leaderboard_items_from_summaries( + &summaries, + &metadata, + state.has_auth_user_data_reader(), + state.has_user_data_reader(), + include_inactive, + exclude_admin, + ); let mut item = AdminStatsLeaderboardItem { id: group.id.clone(), name: group.name, @@ -335,17 +334,47 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response( tokens: 0, cost: 0.0, }; - for user_id in scoped_user_ids { - if let Some(user) = user_usage.get(user_id) { - item.requests = item.requests.saturating_add(user.requests); - item.tokens = item.tokens.saturating_add(user.tokens); - item.cost += user.cost; - } + for user in users { + item.requests = item.requests.saturating_add(user.requests); + item.tokens = item.tokens.saturating_add(user.tokens); + item.cost += user.cost; } member_counts.insert(group.id.clone(), member_count); active_member_counts.insert(group.id, active_member_count); leaderboard.push(item); } + let ungrouped = super::super::ungrouped_usage_users(state).await?; + let id = super::super::UNGROUPED_USAGE_ID.to_string(); + member_counts.insert(id.clone(), ungrouped.len()); + active_member_counts.insert( + id.clone(), + ungrouped.iter().filter(|user| user.is_active).count(), + ); + let user_ids = ungrouped + .into_iter() + .filter(|user| include_inactive || user.is_active) + .filter(|user| !exclude_admin || !user.role.eq_ignore_ascii_case("admin")) + .map(|user| user.id) + .collect(); + let rows = state + .summarize_usage_leaderboard(&UsageLeaderboardQuery { + created_from_unix_secs, + created_until_unix_secs, + group_by: UsageLeaderboardGroupBy::User, + user_id: None, + user_ids: Some(user_ids), + provider_names: None, + provider_name: filters.provider_name.clone(), + model: filters.model.clone(), + }) + .await?; + leaderboard.push(AdminStatsLeaderboardItem { + id, + name: "Ungrouped".to_string(), + requests: rows.iter().map(|row| row.request_count).sum(), + tokens: rows.iter().map(|row| row.total_tokens).sum(), + cost: rows.iter().map(|row| row.total_cost_usd).sum(), + }); leaderboard.sort_by(|left, right| compare_leaderboard_items(metric, order, left, right)); return Ok(Some(build_admin_stats_user_group_leaderboard_response( @@ -422,6 +451,8 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response( }; let summaries = state .summarize_usage_leaderboard(&UsageLeaderboardQuery { + provider_names: super::super::resolve_usage_group_provider_names(state, query) + .await?, created_from_unix_secs, created_until_unix_secs, group_by: UsageLeaderboardGroupBy::User, diff --git a/apps/aether-gateway/src/handlers/admin/observability/usage/summary_routes.rs b/apps/aether-gateway/src/handlers/admin/observability/usage/summary_routes.rs index e7541e523..cc5402a54 100644 --- a/apps/aether-gateway/src/handlers/admin/observability/usage/summary_routes.rs +++ b/apps/aether-gateway/src/handlers/admin/observability/usage/summary_routes.rs @@ -717,6 +717,8 @@ pub(super) async fn maybe_build_local_admin_usage_summary_response( }; let summary = state .summarize_usage_audits(&UsageAuditSummaryQuery { + provider_names: super::super::resolve_usage_group_provider_names(state, query) + .await?, created_from_unix_secs, created_until_unix_secs, user_id: query_param_value(query, "user_id"), diff --git a/apps/aether-gateway/src/tests/control/admin/stats.rs b/apps/aether-gateway/src/tests/control/admin/stats.rs index 32fdb1e99..82dbf34d3 100644 --- a/apps/aether-gateway/src/tests/control/admin/stats.rs +++ b/apps/aether-gateway/src/tests/control/admin/stats.rs @@ -2031,7 +2031,12 @@ async fn gateway_aggregates_admin_stats_by_current_user_group_membership() { assert_eq!(response.status(), StatusCode::OK); let payload: serde_json::Value = response.json().await.expect("json body should parse"); assert_eq!(payload["attribution"], "current_membership"); - assert_eq!(payload["total"], 1); + assert_eq!(payload["total"], 2); + let mut payload = payload; + payload["items"] + .as_array_mut() + .unwrap() + .retain(|item| item["id"] == group.id); assert_eq!(payload["items"][0]["id"], group.id); assert_eq!(payload["items"][0]["name"], "Engineering"); assert_eq!(payload["items"][0]["requests"], 2); @@ -2067,6 +2072,222 @@ async fn gateway_aggregates_admin_stats_by_current_user_group_membership() { upstream_handle.abort(); } +#[tokio::test] +async fn gateway_group_usage_intersects_members_and_providers_with_shared_overlap() { + let rows = vec![ + sample_usage_row( + "g", + "g", + Some("user-1"), + None, + None, + "Gemini", + "model", + 10, + 2, + 0.4, + 0.4, + DAY_1_UNIX_SECS, + ), + sample_usage_row( + "s", + "s", + Some("user-1"), + None, + None, + "Shared", + "model", + 10, + 2, + 0.35, + 0.35, + DAY_1_UNIX_SECS, + ), + sample_usage_row( + "o", + "o", + Some("user-1"), + None, + None, + "Other", + "model", + 10, + 2, + 1.5, + 1.5, + DAY_1_UNIX_SECS, + ), + sample_usage_row( + "x", + "x", + Some("user-2"), + None, + None, + "Gemini", + "model", + 10, + 2, + 10.0, + 10.0, + DAY_1_UNIX_SECS, + ), + ]; + let users = InMemoryUserReadRepository::seed_auth_users([ + sample_auth_user("user-1", "alice", "user", true), + sample_auth_user("user-2", "bob", "user", true), + ]); + let export_users = users.list_export_users().await.unwrap(); + let users = users.with_export_users(export_users); + let mut group_ids = Vec::new(); + for (name, allowed, mode) in [ + ( + "Gemini group", + vec!["provider-gemini", "Shared", "provider-shared"], + "specific", + ), + ("Other group", vec!["Other", "shared-type"], "specific"), + ("Denied", vec!["Gemini"], "deny_all"), + ("Empty", vec![], "specific"), + ("Inherited", vec![], "inherit"), + ("Unrestricted", vec![], "unrestricted"), + ] { + let group = users + .create_user_group(UpsertUserGroupRecord { + name: name.to_string(), + description: None, + priority: 0, + allowed_providers: Some(allowed.into_iter().map(str::to_string).collect()), + allowed_providers_mode: mode.to_string(), + allowed_api_formats: None, + allowed_api_formats_mode: "inherit".to_string(), + allowed_models: None, + allowed_models_mode: "inherit".to_string(), + rate_limit: None, + rate_limit_mode: "inherit".to_string(), + }) + .await + .unwrap() + .unwrap(); + users + .replace_user_group_members(&group.id, &["user-1".to_string()]) + .await + .unwrap(); + group_ids.push(group.id); + } + let mut shared = sample_provider("provider-shared", "Shared", 0); + shared.provider_type = "shared-type".to_string(); + let providers = InMemoryProviderCatalogReadRepository::seed( + vec![ + sample_provider("provider-gemini", "Gemini", 0), + shared, + sample_provider("provider-other", "Other", 0), + ], + vec![], + vec![], + ); + let data = GatewayDataState::with_usage_reader_for_tests(Arc::new( + InMemoryUsageReadRepository::seed(rows), + )) + .with_user_reader(Arc::new(users)) + .with_provider_catalog_reader(Arc::new(providers)); + let gateway = build_router_with_state(AppState::new().unwrap().with_data_state_for_tests(data)); + let (url, handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + let range = "start_date=2024-03-21&end_date=2024-03-21&tz_offset_minutes=0"; + let paths = [ + ( + format!("usage/stats?{range}&user_group_id=__ungrouped__"), + Some(10.0), + ), + ( + format!("stats/time-series?{range}&granularity=day&user_group_id=__ungrouped__"), + Some(10.0), + ), + ( + format!("stats/leaderboard/users?{range}&metric=cost&user_group_id=__ungrouped__"), + Some(10.0), + ), + ( + format!("stats/leaderboard/user-groups?{range}&metric=cost"), + None, + ), + ( + format!("usage/stats?{range}&user_group_id={}", group_ids[0]), + Some(0.75), + ), + ( + format!( + "stats/time-series?{range}&granularity=day&user_group_id={}", + group_ids[1] + ), + Some(1.85), + ), + ( + format!( + "stats/leaderboard/users?{range}&metric=cost&user_group_id={}", + group_ids[0] + ), + Some(0.75), + ), + (format!("usage/stats?{range}&user_id=user-1"), Some(2.25)), + ( + format!( + "usage/stats?{range}&user_group_id={}&provider=Other", + group_ids[0] + ), + Some(0.0), + ), + ]; + for (path, expected) in paths { + let response = admin_request(client.get(format!("{url}/api/admin/{path}"))) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK, "{path}"); + let body: serde_json::Value = response.json().await.unwrap(); + if let Some(expected) = expected { + let value = if path.starts_with("stats/time-series") { + &body[0]["total_cost"] + } else if path.starts_with("stats/leaderboard") { + &body["items"][0]["cost"] + } else { + &body["total_cost"] + }; + assert!( + (value.as_f64().unwrap() - expected).abs() < 1e-9, + "{path}: {body}" + ); + } else { + let items = body["items"].as_array().unwrap(); + assert_eq!(items.len(), 7); + let ungrouped = items + .iter() + .find(|item| item["id"] == "__ungrouped__") + .unwrap(); + assert_eq!(ungrouped["cost"], 10.0); + assert_eq!(ungrouped["member_count"], 1); + assert_eq!(ungrouped["active_member_count"], 1); + + for (id, cost, requests) in [ + (&group_ids[0], 0.75, 2), + (&group_ids[1], 1.85, 2), + (&group_ids[2], 0.0, 0), + (&group_ids[3], 0.0, 0), + (&group_ids[4], 2.25, 3), + (&group_ids[5], 2.25, 3), + ] { + let row = items + .iter() + .find(|item| item["id"].as_str() == Some(id.as_str())) + .unwrap(); + assert!((row["cost"].as_f64().unwrap() - cost).abs() < 1e-9); + assert_eq!(row["requests"], requests); + } + } + } + handle.abort(); +} + #[tokio::test] async fn gateway_handles_admin_stats_leaderboard_users_locally_without_usage_reader() { let (upstream_url, upstream_hits, upstream_handle) = diff --git a/crates/aether-data/adapters/postgres/src/usage/mod.rs b/crates/aether-data/adapters/postgres/src/usage/mod.rs index e9f81ea60..759363c20 100644 --- a/crates/aether-data/adapters/postgres/src/usage/mod.rs +++ b/crates/aether-data/adapters/postgres/src/usage/mod.rs @@ -3715,6 +3715,7 @@ FROM "#, created_until_unix_secs: u64, user_id: Option<&str>, user_ids: Option<&[String]>, + provider_names: Option<&[String]>, provider_name: Option<&str>, model: Option<&str>, ) -> Result { @@ -3770,6 +3771,12 @@ FROM usage_billing_facts AS "usage" .push_bind(created_until_unix_secs as f64) .push("::double precision)"); push_usage_user_scope(&mut builder, "\"usage\".user_id", user_id, user_ids); + if let Some(names) = provider_names { + builder + .push(" AND \"usage\".provider_name = ANY(") + .push_bind(names.to_vec()) + .push("::text[])"); + } if let Some(provider_name) = provider_name { builder.push(if has_where { " AND " } else { " WHERE " }); has_where = true; @@ -3796,13 +3803,17 @@ FROM usage_billing_facts AS "usage" &self, query: &UsageAuditSummaryQuery, ) -> Result { - if query.provider_name.is_some() || query.model.is_some() { + // Provider rollups lack cache-cost/error detail required by this response. + // Keep these scopes on canonical facts rather than inventing missing daily totals. + if query.provider_names.is_some() || query.provider_name.is_some() || query.model.is_some() + { return self .summarize_usage_audits_raw( query.created_from_unix_secs, query.created_until_unix_secs, query.user_id.as_deref(), query.user_ids.as_deref(), + query.provider_names.as_deref(), query.provider_name.as_deref(), query.model.as_deref(), ) @@ -3815,6 +3826,7 @@ FROM usage_billing_facts AS "usage" query.created_until_unix_secs, query.user_id.as_deref(), query.user_ids.as_deref(), + query.provider_names.as_deref(), None, None, ) @@ -3831,6 +3843,7 @@ FROM usage_billing_facts AS "usage" query.created_until_unix_secs, query.user_id.as_deref(), query.user_ids.as_deref(), + query.provider_names.as_deref(), None, None, ) @@ -3846,6 +3859,7 @@ FROM usage_billing_facts AS "usage" dashboard_utc_to_unix_secs(raw_end), query.user_id.as_deref(), query.user_ids.as_deref(), + query.provider_names.as_deref(), None, None, ) @@ -3872,6 +3886,7 @@ FROM usage_billing_facts AS "usage" dashboard_utc_to_unix_secs(raw_end), query.user_id.as_deref(), query.user_ids.as_deref(), + query.provider_names.as_deref(), None, None, ) @@ -6821,6 +6836,12 @@ FROM usage_billing_facts AS "usage" query.user_id.as_deref(), query.user_ids.as_deref(), ); + if let Some(names) = query.provider_names.as_ref() { + builder + .push(" AND \"usage\".provider_name = ANY(") + .push_bind(names.clone()) + .push("::text[])"); + } if let Some(provider_name) = query.provider_name.as_deref() { builder.push(if has_where { " AND " } else { " WHERE " }); has_where = true; @@ -6861,6 +6882,7 @@ FROM usage_billing_facts AS "usage" end_day_utc: DateTime, user_id: Option<&str>, user_ids: Option<&[String]>, + provider_names: Option<&[String]>, ) -> Result, DataLayerError> { if start_day_utc >= end_day_utc { return Ok(Vec::new()); @@ -6881,7 +6903,9 @@ SELECT AS total_response_time_ms FROM "#, ); - builder.push(if scoped_to_users { + builder.push(if provider_names.is_some() { + "stats_user_daily_provider" + } else if scoped_to_users { "stats_user_daily" } else { "stats_daily" @@ -6894,6 +6918,12 @@ FROM "#, if scoped_to_users { push_usage_user_scope(&mut builder, "user_id", user_id, user_ids); } + if let Some(names) = provider_names { + builder + .push(" AND provider_name = ANY(") + .push_bind(names.to_vec()) + .push("::text[])"); + } builder.push(" GROUP BY date ORDER BY date ASC"); let mut rows = builder.build().fetch(&self.pool); @@ -6979,6 +7009,7 @@ WHERE is_complete IS TRUE &mut grouped, self.summarize_usage_time_series_raw( &UsageTimeSeriesQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start), created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end), granularity: UsageTimeSeriesGranularity::Day, @@ -7000,6 +7031,7 @@ WHERE is_complete IS TRUE aggregate_end, query.user_id.as_deref(), query.user_ids.as_deref(), + query.provider_names.as_deref(), ) .await?, ); @@ -7007,6 +7039,7 @@ WHERE is_complete IS TRUE &mut grouped, self.summarize_usage_time_series_raw( &UsageTimeSeriesQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(aggregate_start), created_until_unix_secs: dashboard_utc_to_unix_secs(aggregate_end), granularity: UsageTimeSeriesGranularity::Day, @@ -7025,6 +7058,7 @@ WHERE is_complete IS TRUE &mut grouped, self.summarize_usage_time_series_raw( &UsageTimeSeriesQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start), created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end), granularity: UsageTimeSeriesGranularity::Day, @@ -7044,7 +7078,10 @@ WHERE is_complete IS TRUE } } - if query.user_id.is_none() && query.user_ids.is_none() && query.tz_offset_minutes % 60 == 0 + if query.provider_names.is_none() + && query.user_id.is_none() + && query.user_ids.is_none() + && query.tz_offset_minutes % 60 == 0 { if let Some(cutoff_utc) = self.read_stats_hourly_cutoff().await? { let start_utc = dashboard_unix_secs_to_utc(query.created_from_unix_secs); @@ -7057,6 +7094,7 @@ WHERE is_complete IS TRUE &mut grouped, self.summarize_usage_time_series_raw( &UsageTimeSeriesQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start), created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end), granularity: query.granularity, @@ -7085,6 +7123,7 @@ WHERE is_complete IS TRUE &mut grouped, self.summarize_usage_time_series_raw( &UsageTimeSeriesQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(aggregate_start), created_until_unix_secs: dashboard_utc_to_unix_secs(aggregate_end), granularity: query.granularity, @@ -7103,6 +7142,7 @@ WHERE is_complete IS TRUE &mut grouped, self.summarize_usage_time_series_raw( &UsageTimeSeriesQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start), created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end), granularity: query.granularity, @@ -7149,6 +7189,7 @@ WHERE "usage".created_at >= TO_TIMESTAMP($1::double precision) AND ($4::varchar IS NULL OR "usage".provider_name = $4) AND ($5::varchar IS NULL OR "usage".model = $5) AND ($6::text[] IS NULL OR "usage".user_id::text = ANY($6)) + AND ($7::text[] IS NULL OR "usage".provider_name = ANY($7)) GROUP BY group_key ORDER BY group_key ASC "#, @@ -7163,6 +7204,7 @@ ORDER BY group_key ASC .bind(query.provider_name.as_deref()) .bind(query.model.as_deref()) .bind(query.user_ids.clone()) + .bind(query.provider_names.clone()) .fetch(&self.pool); let mut items = Vec::new(); while let Some(row) = rows.try_next().await.map_postgres_err()? { @@ -7177,6 +7219,51 @@ ORDER BY group_key ASC end_day_utc: DateTime, query: &UsageLeaderboardQuery, ) -> Result>, DataLayerError> { + if let Some(names) = query.provider_names.as_ref() { + let group_key = match query.group_by { + UsageLeaderboardGroupBy::User => "user_id", + UsageLeaderboardGroupBy::Model => "model", + UsageLeaderboardGroupBy::ApiKey => return Ok(None), + }; + let mut builder = QueryBuilder::::new(format!( + r#" +SELECT + {group_key} AS group_key, + MAX(NULLIF(BTRIM(username), '')) AS legacy_name, + COALESCE(SUM(total_requests), 0)::BIGINT AS request_count, + COALESCE(SUM(total_tokens), 0)::BIGINT AS total_tokens, + COALESCE(SUM(total_cost), 0)::DOUBLE PRECISION AS total_cost_usd +FROM stats_user_daily_model_provider +WHERE date >= +"# + )); + builder + .push_bind(start_day_utc) + .push(" AND date < ") + .push_bind(end_day_utc); + builder + .push(" AND provider_name = ANY(") + .push_bind(names.clone()) + .push("::text[])"); + push_usage_user_scope( + &mut builder, + "user_id", + query.user_id.as_deref(), + query.user_ids.as_deref(), + ); + if let Some(provider) = query.provider_name.as_ref() { + builder + .push(" AND provider_name = ") + .push_bind(provider.clone()); + } + if let Some(model) = query.model.as_ref() { + builder.push(" AND model = ").push_bind(model.clone()); + } + builder.push(format!(" GROUP BY {group_key} ORDER BY {group_key}")); + return fetch_usage_leaderboard_query(builder.build(), &self.pool) + .await + .map(Some); + } let items = match query.group_by { UsageLeaderboardGroupBy::Model => { let mut builder = if let Some(user_id) = query.user_id.as_deref() { @@ -7452,6 +7539,7 @@ WHERE stats_daily_api_key.date >= absorb_usage_leaderboard_rows( &mut grouped, self.summarize_usage_leaderboard_raw(&UsageLeaderboardQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start), created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end), ..query.clone() @@ -7474,6 +7562,7 @@ WHERE stats_daily_api_key.date >= absorb_usage_leaderboard_rows( &mut grouped, self.summarize_usage_leaderboard_raw(&UsageLeaderboardQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(aggregate_start), created_until_unix_secs: dashboard_utc_to_unix_secs(aggregate_end), ..query.clone() @@ -7487,6 +7576,7 @@ WHERE stats_daily_api_key.date >= absorb_usage_leaderboard_rows( &mut grouped, self.summarize_usage_leaderboard_raw(&UsageLeaderboardQuery { + provider_names: query.provider_names.clone(), created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start), created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end), ..query.clone() diff --git a/crates/aether-data/contracts/src/repository/usage/types.rs b/crates/aether-data/contracts/src/repository/usage/types.rs index c266625e8..494727f15 100644 --- a/crates/aether-data/contracts/src/repository/usage/types.rs +++ b/crates/aether-data/contracts/src/repository/usage/types.rs @@ -1060,6 +1060,9 @@ pub struct UsageAuditAggregationQuery { #[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)] pub struct UsageAuditSummaryQuery { + /// Optional provider-name allowlist, intersected with provider_name; empty matches nothing. + #[serde(default)] + pub provider_names: Option>, pub created_from_unix_secs: u64, pub created_until_unix_secs: u64, pub user_id: Option, @@ -1453,6 +1456,9 @@ pub enum UsageTimeSeriesGranularity { #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct UsageTimeSeriesQuery { + /// Optional provider-name allowlist, intersected with provider_name; empty matches nothing. + #[serde(default)] + pub provider_names: Option>, pub created_from_unix_secs: u64, pub created_until_unix_secs: u64, pub granularity: UsageTimeSeriesGranularity, @@ -1487,6 +1493,9 @@ pub enum UsageLeaderboardGroupBy { #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct UsageLeaderboardQuery { + /// Optional provider-name allowlist, intersected with provider_name; empty matches nothing. + #[serde(default)] + pub provider_names: Option>, pub created_from_unix_secs: u64, pub created_until_unix_secs: u64, pub group_by: UsageLeaderboardGroupBy, diff --git a/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs b/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs index 02fa33e24..9453e3578 100644 --- a/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs +++ b/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs @@ -2382,6 +2382,7 @@ INSERT INTO public.stats_daily_api_key ( .expect("API key daily aggregate fixtures should be inserted"); let leaderboard_query = UsageLeaderboardQuery { + provider_names: None, created_from_unix_secs: u64::try_from(stats_day.timestamp()) .expect("historical stats day should be nonnegative"), created_until_unix_secs: u64::try_from((stats_day + chrono::Duration::days(1)).timestamp()) diff --git a/crates/aether-data/runtime/src/repository/usage/memory.rs b/crates/aether-data/runtime/src/repository/usage/memory.rs index da40b1b72..1236c4104 100644 --- a/crates/aether-data/runtime/src/repository/usage/memory.rs +++ b/crates/aether-data/runtime/src/repository/usage/memory.rs @@ -502,6 +502,13 @@ fn usage_matches_summary_query( return false; } } + if query + .provider_names + .as_ref() + .is_some_and(|names| !names.contains(&item.provider_name)) + { + return false; + } if let Some(provider_name) = query.provider_name.as_deref() { if item.provider_name != provider_name { return false; @@ -538,6 +545,13 @@ fn usage_matches_time_series_query( return false; } } + if query + .provider_names + .as_ref() + .is_some_and(|names| !names.contains(&item.provider_name)) + { + return false; + } if let Some(provider_name) = query.provider_name.as_deref() { if item.provider_name != provider_name { return false; @@ -918,6 +932,13 @@ fn usage_matches_leaderboard_query( return false; } } + if query + .provider_names + .as_ref() + .is_some_and(|names| !names.contains(&item.provider_name)) + { + return false; + } if let Some(provider_name) = query.provider_name.as_deref() { if item.provider_name != provider_name { return false; diff --git a/crates/aether-data/runtime/src/repository/usage/memory/tests.rs b/crates/aether-data/runtime/src/repository/usage/memory/tests.rs index d15e6951b..087ffe9a3 100644 --- a/crates/aether-data/runtime/src/repository/usage/memory/tests.rs +++ b/crates/aether-data/runtime/src/repository/usage/memory/tests.rs @@ -457,6 +457,7 @@ async fn unmetered_session_audit_counts_lifecycle_without_token_or_cost_contribu let summary = repository .summarize_usage_audits(&UsageAuditSummaryQuery { + provider_names: None, created_from_unix_secs: 0, created_until_unix_secs: 1_000, ..UsageAuditSummaryQuery::default() @@ -2213,6 +2214,7 @@ async fn dashboard_and_leaderboard_total_tokens_use_effective_cache_aware_tokens let leaderboard = repository .summarize_usage_leaderboard(&UsageLeaderboardQuery { + provider_names: None, created_from_unix_secs: 1_711_000_000, created_until_unix_secs: 1_711_000_001, group_by: UsageLeaderboardGroupBy::User, @@ -2239,6 +2241,7 @@ async fn usage_analytics_filters_by_multiple_user_ids() { let summary = repository .summarize_usage_audits(&UsageAuditSummaryQuery { + provider_names: None, created_from_unix_secs: 1_711_000_000, created_until_unix_secs: 1_711_000_001, user_ids: Some(scoped_user_ids.clone()), @@ -2250,6 +2253,7 @@ async fn usage_analytics_filters_by_multiple_user_ids() { let buckets = repository .summarize_usage_time_series(&UsageTimeSeriesQuery { + provider_names: None, created_from_unix_secs: 1_711_000_000, created_until_unix_secs: 1_711_000_001, granularity: UsageTimeSeriesGranularity::Day, @@ -2271,6 +2275,7 @@ async fn usage_analytics_filters_by_multiple_user_ids() { let leaderboard = repository .summarize_usage_leaderboard(&UsageLeaderboardQuery { + provider_names: None, created_from_unix_secs: 1_711_000_000, created_until_unix_secs: 1_711_000_001, group_by: UsageLeaderboardGroupBy::User, @@ -2285,6 +2290,79 @@ async fn usage_analytics_filters_by_multiple_user_ids() { assert!(leaderboard.iter().all(|item| item.group_key != "user-3")); } +#[tokio::test] +async fn usage_analytics_intersects_provider_allowlist_and_user_scope() { + let mut a = sample_usage("allowed", 1_711_000_000); + a.provider_name = "Gemini".to_string(); + a.user_id = Some("user-1".to_string()); + let mut b = a.clone(); + b.request_id = "other-provider".to_string(); + b.provider_name = "Other".to_string(); + let mut c = a.clone(); + c.request_id = "other-user".to_string(); + c.user_id = Some("user-2".to_string()); + let repository = InMemoryUsageReadRepository::seed(vec![a, b, c]); + for (names, provider, count) in [ + ( + Some(vec!["Gemini".to_string(), "Gemini".to_string()]), + None, + 1, + ), + (Some(vec![]), None, 0), + ( + Some(vec!["Gemini".to_string()]), + Some("Other".to_string()), + 0, + ), + (None, None, 2), + ] { + let summary = repository + .summarize_usage_audits(&UsageAuditSummaryQuery { + created_from_unix_secs: 1_711_000_000, + created_until_unix_secs: 1_711_000_001, + user_ids: Some(vec!["user-1".to_string()]), + provider_names: names.clone(), + provider_name: provider.clone(), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(summary.total_requests, count); + let buckets = repository + .summarize_usage_time_series(&UsageTimeSeriesQuery { + created_from_unix_secs: 1_711_000_000, + created_until_unix_secs: 1_711_000_001, + user_id: None, + user_ids: Some(vec!["user-1".to_string()]), + provider_names: names.clone(), + provider_name: provider.clone(), + model: None, + granularity: UsageTimeSeriesGranularity::Day, + tz_offset_minutes: 0, + }) + .await + .unwrap(); + assert_eq!( + buckets.iter().map(|row| row.total_requests).sum::(), + count + ); + let rows = repository + .summarize_usage_leaderboard(&UsageLeaderboardQuery { + created_from_unix_secs: 1_711_000_000, + created_until_unix_secs: 1_711_000_001, + user_id: None, + user_ids: Some(vec!["user-1".to_string()]), + provider_names: names, + provider_name: provider, + model: None, + group_by: UsageLeaderboardGroupBy::User, + }) + .await + .unwrap(); + assert_eq!(rows.iter().map(|row| row.request_count).sum::(), count); + } +} + #[tokio::test] async fn summarizes_provider_api_key_last_used_at_in_seconds() { let repository = InMemoryUsageReadRepository::seed(vec![ diff --git a/frontend/src/components/stats/LeaderboardTable.vue b/frontend/src/components/stats/LeaderboardTable.vue index d9838acf8..33de9ca4c 100644 --- a/frontend/src/components/stats/LeaderboardTable.vue +++ b/frontend/src/components/stats/LeaderboardTable.vue @@ -77,7 +77,7 @@ {{ item.rank }} - {{ item.name }} + {{ showMemberCount && item.id === '__ungrouped__' ? t('userStats.ungrouped') : item.name }} (() => scope.value === 'user' ? users.value.map(user => ({ id: user.id, name: user.username || user.email || user.id })) - : userGroups.value.map(group => ({ id: group.id, name: group.name }))) + : [...userGroups.value.map(group => ({ id: group.id, name: group.name })), + { id: '__ungrouped__', name: t('userStats.ungrouped') }]) const selectedEntityId = computed({ get: () => scope.value === 'user' ? selectedUserId.value : selectedUserGroupId.value, @@ -417,7 +418,7 @@ async function loadPanels() { limit: PAGE_SIZE }) : Promise.resolve({ items: [] }) - const groupMembersPromise: Promise = scope.value === 'user_group' + const groupMembersPromise: Promise = scope.value === 'user_group' && selectedId !== '__ungrouped__' ? usersApi.listUserGroupMembers(selectedId) : Promise.resolve([]) @@ -433,10 +434,12 @@ async function loadPanels() { series.value = primarySeries comparisonSeries.value = compareSeries memberLeaderboard.value = members.items - groupMemberCount.value = groupMembers.filter(member => !member.is_deleted).length - activeGroupMemberCount.value = groupMembers.filter( - member => !member.is_deleted && member.is_active - ).length + const ungroupedUsers = users.value.filter(user => user.groups?.length === 0) + groupMemberCount.value = selectedId === '__ungrouped__' + ? ungroupedUsers.length : groupMembers.filter(member => !member.is_deleted).length + activeGroupMemberCount.value = selectedId === '__ungrouped__' + ? ungroupedUsers.filter(user => user.is_active).length + : groupMembers.filter(member => !member.is_deleted && member.is_active).length } finally { if (requestId === panelRequestId) { summaryLoading.value = false