Merge pull request #834 from dalamudx/feat/user-group-stats

feat(stats): add user group usage views
2026-10-04 00:17:45 +08:00 · 2026-09-19 20:43:45 +08:00
parent 37e3a36680 a95f0d2488
commit ba7c9f8b27
24 changed files with 1198 additions and 346 deletions
@@ -596,6 +596,20 @@ pub(super) fn classify_admin_observability_family_route(
            "admin:stats",
            false,
        ))
+    } else if method == http::Method::GET
+        && matches!(
+            normalized_path,
+            "/api/admin/stats/leaderboard/user-groups"
+                | "/api/admin/stats/leaderboard/user-groups/"
+        )
+    {
+        Some(classified(
+            "admin_proxy",
+            "stats_manage",
+            "leaderboard_user_groups",
+            "admin:stats",
+            false,
+        ))
    } else if method == http::Method::GET
        && matches!(
            normalized_path,
@@ -157,6 +157,28 @@ fn classifies_admin_stats_leaderboard_models_as_admin_proxy_route() {
    assert!(!decision.is_execution_runtime_candidate());
 }

+#[test]
+fn classifies_admin_stats_leaderboard_user_groups_as_admin_proxy_route() {
+    let headers = headers(&[]);
+    let uri: Uri = "/api/admin/stats/leaderboard/user-groups"
+        .parse()
+        .expect("uri should parse");
+    let decision =
+        classify_control_route(&http::Method::GET, &uri, &headers).expect("route should classify");
+
+    assert_eq!(decision.route_class.as_deref(), Some("admin_proxy"));
+    assert_eq!(decision.route_family.as_deref(), Some("stats_manage"));
+    assert_eq!(
+        decision.route_kind.as_deref(),
+        Some("leaderboard_user_groups")
+    );
+    assert_eq!(
+        decision.auth_endpoint_signature.as_deref(),
+        Some("admin:stats")
+    );
+    assert!(!decision.is_execution_runtime_candidate());
+}
+
 #[test]
 fn classifies_admin_stats_leaderboard_users_as_admin_proxy_route() {
    let headers = headers(&[]);
@@ -11,3 +11,31 @@ pub(crate) use self::stats::{
 };
 pub(crate) use self::stats::{AdminStatsTimeRange, AdminStatsUsageFilter};
 pub(crate) use self::usage::maybe_build_local_admin_usage_response;
+
+pub(crate) async fn resolve_usage_user_group_scope(
+    state: &crate::handlers::admin::request::AdminAppState<'_>,
+    query: Option<&str>,
+    include_inactive: bool,
+    exclude_admin: bool,
+) -> Result<Result<Option<Vec<String>>, String>, crate::GatewayError> {
+    let group_id = crate::handlers::admin::shared::query_param_value(query, "user_group_id");
+    let Some(group_id) = group_id else {
+        return Ok(Ok(None));
+    };
+    if crate::handlers::admin::shared::query_param_value(query, "user_id").is_some() {
+        return Ok(Err(
+            "user_id and user_group_id cannot be used together".to_string()
+        ));
+    }
+    if !state.has_user_data_reader() {
+        return Ok(Err("user group data is unavailable".to_string()));
+    }
+
+    match state
+        .resolve_usage_user_group_member_ids(&group_id, include_inactive, exclude_admin)
+        .await?
+    {
+        Some(user_ids) => Ok(Ok(Some(user_ids))),
+        None => Ok(Err("user_group_id does not exist".to_string())),
+    }
+}
@@ -172,6 +172,7 @@ pub(super) async fn build_admin_monitoring_system_status_response(
            created_from_unix_secs: today_start.timestamp().max(0) as u64,
            created_until_unix_secs: now_unix_secs.saturating_add(1),
            user_id: None,
+            user_ids: None,
            provider_name: None,
            model: None,
        })
@@ -1,3 +1,4 @@
+use super::super::resolve_usage_user_group_scope;
 use super::range::{build_comparison_range, parse_bounded_u32};
 use super::resolve_admin_usage_time_range;
 use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
@@ -294,6 +295,17 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response(
        }

        let filters = AdminStatsUsageFilter::from_query(request_context.query_string());
+        let user_ids = match resolve_usage_user_group_scope(
+            state,
+            request_context.query_string(),
+            false,
+            false,
+        )
+        .await?
+        {
+            Ok(value) => value,
+            Err(detail) => return Ok(Some(admin_stats_bad_request_response(detail))),
+        };
        let query_granularity = match granularity {
            AdminStatsGranularity::Hour => UsageTimeSeriesGranularity::Hour,
            AdminStatsGranularity::Day
@@ -311,6 +323,7 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response(
                granularity: query_granularity,
                tz_offset_minutes: time_range.tz_offset_minutes,
                user_id: filters.user_id,
+                user_ids,
                provider_name: filters.provider_name,
                model: filters.model,
            })
@@ -77,6 +77,7 @@ pub(super) async fn maybe_build_local_admin_stats_cost_response(
                granularity: UsageTimeSeriesGranularity::Day,
                tz_offset_minutes: time_range.tz_offset_minutes,
                user_id: None,
+                user_ids: None,
                provider_name: None,
                model: None,
            })
@@ -3,12 +3,13 @@ use crate::GatewayError;
 use aether_data_contracts::repository::usage::StoredRequestUsageAudit;

 pub(super) use aether_admin::observability::stats::{
-    build_admin_stats_leaderboard_response, build_api_key_leaderboard_items,
-    build_api_key_leaderboard_items_from_summaries, build_model_leaderboard_items,
-    build_model_leaderboard_items_from_summaries, build_user_leaderboard_items,
-    build_user_leaderboard_items_from_summaries, compare_leaderboard_items, compute_dense_rank,
-    AdminStatsLeaderboardItem, AdminStatsLeaderboardMetric, AdminStatsLeaderboardNameMode,
-    AdminStatsSortOrder, AdminStatsUserMetadata,
+    build_admin_stats_leaderboard_response, build_admin_stats_user_group_leaderboard_response,
+    build_api_key_leaderboard_items, build_api_key_leaderboard_items_from_summaries,
+    build_model_leaderboard_items, build_model_leaderboard_items_from_summaries,
+    build_user_leaderboard_items, build_user_leaderboard_items_from_summaries,
+    compare_leaderboard_items, compute_dense_rank, AdminStatsLeaderboardItem,
+    AdminStatsLeaderboardMetric, AdminStatsLeaderboardNameMode, AdminStatsSortOrder,
+    AdminStatsUserMetadata,
 };

 pub(super) async fn load_user_leaderboard_metadata(
@@ -1,7 +1,9 @@
+use super::super::resolve_usage_user_group_scope;
 use super::leaderboard::{
-    build_admin_stats_leaderboard_response, build_api_key_leaderboard_items_from_summaries,
-    build_model_leaderboard_items_from_summaries, build_user_leaderboard_items_from_summaries,
-    compare_leaderboard_items, load_user_leaderboard_metadata, AdminStatsLeaderboardNameMode,
+    build_admin_stats_leaderboard_response, build_admin_stats_user_group_leaderboard_response,
+    build_api_key_leaderboard_items_from_summaries, build_model_leaderboard_items_from_summaries,
+    build_user_leaderboard_items_from_summaries, compare_leaderboard_items,
+    load_user_leaderboard_metadata, AdminStatsLeaderboardItem, AdminStatsLeaderboardNameMode,
 };
 use super::range::{parse_bounded_u32, parse_nonnegative_usize};
 use super::resolve_admin_usage_time_range;
@@ -14,6 +16,7 @@ use aether_admin::observability::stats::{
 };
 use aether_data_contracts::repository::usage::{UsageLeaderboardGroupBy, UsageLeaderboardQuery};
 use axum::{body::Body, http, response::Response};
+use std::collections::{BTreeMap, BTreeSet};

 pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
    state: &AdminAppState<'_>,
@@ -79,6 +82,7 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
                created_until_unix_secs,
                group_by: UsageLeaderboardGroupBy::Model,
                user_id: filters.user_id,
+                user_ids: None,
                provider_name: filters.provider_name,
                model: filters.model,
            })
@@ -156,6 +160,7 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
                created_until_unix_secs,
                group_by: UsageLeaderboardGroupBy::ApiKey,
                user_id: filters.user_id,
+                user_ids: None,
                provider_name: filters.provider_name,
                model: filters.model,
            })
@@ -206,6 +211,154 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
        )));
    }

+    if request_context
+        .decision()
+        .and_then(|decision| decision.route_kind.as_deref())
+        == Some("leaderboard_user_groups")
+        && request_context.method() == http::Method::GET
+        && matches!(
+            request_context.path(),
+            "/api/admin/stats/leaderboard/user-groups"
+                | "/api/admin/stats/leaderboard/user-groups/"
+        )
+    {
+        let time_range = match resolve_admin_usage_time_range(query) {
+            Ok(value) => value,
+            Err(detail) => return Ok(Some(admin_stats_bad_request_response(detail))),
+        };
+        let metric = match AdminStatsLeaderboardMetric::parse(query) {
+            Ok(value) => value,
+            Err(detail) => return Ok(Some(admin_stats_bad_request_response(detail))),
+        };
+        let order = match AdminStatsSortOrder::parse(query) {
+            Ok(value) => value,
+            Err(detail) => return Ok(Some(admin_stats_bad_request_response(detail))),
+        };
+        let limit = match query_param_value(query, "limit")
+            .map(|value| parse_bounded_u32("limit", &value, 1, 100))
+            .transpose()
+        {
+            Ok(Some(value)) => value as usize,
+            Ok(None) => 10,
+            Err(detail) => return Ok(Some(admin_stats_bad_request_response(detail))),
+        };
+        let offset = match query_param_value(query, "offset")
+            .map(|value| parse_nonnegative_usize("offset", &value))
+            .transpose()
+        {
+            Ok(Some(value)) => value,
+            Ok(None) => 0,
+            Err(detail) => return Ok(Some(admin_stats_bad_request_response(detail))),
+        };
+        let empty_counts = BTreeMap::new();
+        if !state.has_usage_data_reader() || !state.has_user_data_reader() {
+            return Ok(Some(build_admin_stats_user_group_leaderboard_response(
+                metric,
+                Some(&time_range),
+                &[],
+                &empty_counts,
+                &empty_counts,
+                offset,
+                limit,
+            )));
+        }
+        let include_inactive = query_param_bool(query, "include_inactive", false);
+        let exclude_admin = query_param_bool(query, "exclude_admin", false);
+        let filters = AdminStatsUsageFilter::from_query(query);
+        if filters.user_id.is_some() {
+            return Ok(Some(admin_stats_bad_request_response(
+                "user_id is not supported for the user group leaderboard".to_string(),
+            )));
+        }
+        let Some((created_from_unix_secs, created_until_unix_secs)) = time_range.to_unix_bounds()
+        else {
+            return Ok(Some(build_admin_stats_user_group_leaderboard_response(
+                metric,
+                Some(&time_range),
+                &[],
+                &empty_counts,
+                &empty_counts,
+                offset,
+                limit,
+            )));
+        };
+
+        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::<Vec<_>>();
+        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::<BTreeMap<_, _>>();
+
+        let mut leaderboard = Vec::new();
+        let mut member_counts = BTreeMap::new();
+        let mut active_member_counts = BTreeMap::new();
+        for group in state.list_user_groups().await? {
+            let members = state.list_user_group_members(&group.id).await?;
+            let member_count = members.iter().filter(|member| !member.is_deleted).count();
+            let active_member_count = members
+                .iter()
+                .filter(|member| !member.is_deleted && member.is_active)
+                .count();
+            let scoped_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::<BTreeSet<_>>();
+            let mut item = AdminStatsLeaderboardItem {
+                id: group.id.clone(),
+                name: group.name,
+                requests: 0,
+                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;
+                }
+            }
+            member_counts.insert(group.id.clone(), member_count);
+            active_member_counts.insert(group.id, active_member_count);
+            leaderboard.push(item);
+        }
+        leaderboard.sort_by(|left, right| compare_leaderboard_items(metric, order, left, right));
+
+        return Ok(Some(build_admin_stats_user_group_leaderboard_response(
+            metric,
+            Some(&time_range),
+            &leaderboard,
+            &member_counts,
+            &active_member_counts,
+            offset,
+            limit,
+        )));
+    }
+
    if request_context
        .decision()
        .and_then(|decision| decision.route_kind.as_deref())
@@ -253,6 +406,13 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
        let include_inactive = query_param_bool(query, "include_inactive", false);
        let exclude_admin = query_param_bool(query, "exclude_admin", false);
        let filters = AdminStatsUsageFilter::from_query(query);
+        let scoped_user_ids =
+            match resolve_usage_user_group_scope(state, query, include_inactive, exclude_admin)
+                .await?
+            {
+                Ok(value) => value,
+                Err(detail) => return Ok(Some(admin_stats_bad_request_response(detail))),
+            };
        let Some((created_from_unix_secs, created_until_unix_secs)) = time_range.to_unix_bounds()
        else {
            return Ok(Some(admin_stats_leaderboard_empty_response(
@@ -266,6 +426,7 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
                created_until_unix_secs,
                group_by: UsageLeaderboardGroupBy::User,
                user_id: filters.user_id,
+                user_ids: scoped_user_ids,
                provider_name: filters.provider_name,
                model: filters.model,
            })
@@ -1,3 +1,4 @@
+use super::super::resolve_usage_user_group_scope;
 use super::super::stats::resolve_admin_usage_time_range;
 use super::analytics::admin_usage_api_key_names;
 use super::analytics::admin_usage_provider_key_names;
@@ -710,11 +711,16 @@ pub(super) async fn maybe_build_local_admin_usage_summary_response(
                    &Default::default(),
                )));
            };
+            let user_ids = match resolve_usage_user_group_scope(state, query, false, false).await? {
+                Ok(value) => value,
+                Err(detail) => return Ok(Some(admin_usage_bad_request_response(detail))),
+            };
            let summary = state
                .summarize_usage_audits(&UsageAuditSummaryQuery {
                    created_from_unix_secs,
                    created_until_unix_secs,
                    user_id: query_param_value(query, "user_id"),
+                    user_ids,
                    provider_name: query_param_value(query, "provider"),
                    model: query_param_value(query, "model"),
                })
@@ -126,6 +126,30 @@ impl<'a> AdminAppState<'a> {
        self.app.list_user_group_members(group_id).await
    }

+    pub(crate) async fn resolve_usage_user_group_member_ids(
+        &self,
+        group_id: &str,
+        include_inactive: bool,
+        exclude_admin: bool,
+    ) -> Result<Option<Vec<String>>, GatewayError> {
+        if self.find_user_group_by_id(group_id).await?.is_none() {
+            return Ok(None);
+        }
+
+        let mut user_ids = self
+            .list_user_group_members(group_id)
+            .await?
+            .into_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)
+            .collect::<Vec<_>>();
+        user_ids.sort();
+        user_ids.dedup();
+        Ok(Some(user_ids))
+    }
+
    pub(crate) async fn replace_user_group_members(
        &self,
        group_id: &str,
@@ -9,7 +9,8 @@ use aether_data::repository::auth::{
 use aether_data::repository::provider_catalog::InMemoryProviderCatalogReadRepository;
 use aether_data::repository::usage::InMemoryUsageReadRepository;
 use aether_data::repository::users::{
-    InMemoryUserReadRepository, StoredUserAuthRecord, StoredUserSummary,
+    InMemoryUserReadRepository, StoredUserAuthRecord, StoredUserSummary, UpsertUserGroupRecord,
+    UserReadRepository,
 };
 use aether_data_contracts::repository::usage::StoredRequestUsageAudit;
 use async_trait::async_trait;
@@ -1937,6 +1938,135 @@ async fn gateway_handles_admin_stats_leaderboard_users_without_legacy_username_f
    upstream_handle.abort();
 }

+#[tokio::test]
+async fn gateway_aggregates_admin_stats_by_current_user_group_membership() {
+    let (_upstream_url, upstream_hits, upstream_handle) =
+        start_stats_upstream("/api/admin/stats/leaderboard/user-groups").await;
+    let usage_repository = Arc::new(InMemoryUsageReadRepository::seed(vec![
+        sample_usage_row(
+            "usage-group-a",
+            "req-group-a",
+            Some("user-1"),
+            Some("key-1"),
+            Some("key-1"),
+            "OpenAI",
+            "gpt-5",
+            60,
+            20,
+            0.4,
+            0.4,
+            DAY_1_UNIX_SECS,
+        ),
+        sample_usage_row(
+            "usage-group-b",
+            "req-group-b",
+            Some("user-2"),
+            Some("key-2"),
+            Some("key-2"),
+            "OpenAI",
+            "gpt-5",
+            40,
+            10,
+            0.35,
+            0.35,
+            DAY_1_UNIX_SECS + 10,
+        ),
+        sample_usage_row(
+            "usage-group-outside",
+            "req-group-outside",
+            Some("user-3"),
+            Some("key-3"),
+            Some("key-3"),
+            "OpenAI",
+            "gpt-5",
+            100,
+            50,
+            1.5,
+            1.5,
+            DAY_1_UNIX_SECS + 20,
+        ),
+    ]));
+    let user_repository = InMemoryUserReadRepository::seed_auth_users([
+        sample_auth_user("user-1", "alice", "user", true),
+        sample_auth_user("user-2", "bob", "user", true),
+        sample_auth_user("user-3", "carol", "user", true),
+    ]);
+    let group = user_repository
+        .create_user_group(UpsertUserGroupRecord {
+            name: "Engineering".to_string(),
+            description: None,
+            priority: 0,
+            allowed_providers: None,
+            allowed_providers_mode: "inherit".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
+        .expect("group creation should succeed")
+        .expect("group should be created");
+    user_repository
+        .replace_user_group_members(&group.id, &["user-1".to_string(), "user-2".to_string()])
+        .await
+        .expect("group members should be replaced");
+    let data_state = GatewayDataState::with_usage_reader_for_tests(usage_repository)
+        .with_user_reader(Arc::new(user_repository));
+    let gateway = build_router_with_state(
+        AppState::new()
+            .expect("gateway should build")
+            .with_data_state_for_tests(data_state),
+    );
+    let (gateway_url, gateway_handle) = start_server(gateway).await;
+    let client = reqwest::Client::new();
+
+    let response = admin_request(client.get(format!(
+        "{gateway_url}/api/admin/stats/leaderboard/user-groups?start_date=2024-03-21&end_date=2024-03-21&metric=cost&tz_offset_minutes=0"
+    )))
+    .send()
+    .await
+    .expect("group leaderboard 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["attribution"], "current_membership");
+    assert_eq!(payload["total"], 1);
+    assert_eq!(payload["items"][0]["id"], group.id);
+    assert_eq!(payload["items"][0]["name"], "Engineering");
+    assert_eq!(payload["items"][0]["requests"], 2);
+    assert_eq!(payload["items"][0]["cost"], 0.75);
+    assert_eq!(payload["items"][0]["member_count"], 2);
+    assert_eq!(payload["items"][0]["active_member_count"], 2);
+
+    let response = admin_request(client.get(format!(
+        "{gateway_url}/api/admin/stats/leaderboard/users?start_date=2024-03-21&end_date=2024-03-21&metric=cost&tz_offset_minutes=0&user_group_id={}",
+        group.id
+    )))
+    .send()
+    .await
+    .expect("group member leaderboard 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["total"], 2);
+    assert!(payload["items"]
+        .as_array()
+        .is_some_and(|items| { items.iter().all(|item| item["id"] != "user-3") }));
+
+    let response = admin_request(client.get(format!(
+        "{gateway_url}/api/admin/stats/leaderboard/users?start_date=2024-03-21&end_date=2024-03-21&user_id=user-1&user_group_id={}",
+        group.id
+    )))
+    .send()
+    .await
+    .expect("conflicting scope request should complete");
+    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
+
+    gateway_handle.abort();
+    upstream_handle.abort();
+}
+
 #[tokio::test]
 async fn gateway_handles_admin_stats_leaderboard_users_locally_without_usage_reader() {
    let (upstream_url, upstream_hits, upstream_handle) =