systemprompt_analytics/repository/
requests.rs

1//! AI-request analytics.
2//!
3//! [`RequestAnalyticsRepository`] reports token, cost, latency, and
4//! cache-hit stats, per-model usage breakdowns, trend series, and a paged
5//! request list, each optionally filtered by a model substring and the list
6//! by user as well. Stats and the model breakdown read the hourly rollup
7//! (`report_ai_usage_hourly`), widening the window to whole UTC hours at its
8//! start; the trend series and the request list are per-request reads of
9//! `report_ai_requests`, which no rollup answers.
10//!
11//! Copyright (c) systemprompt.io — Business Source License 1.1.
12//! See <https://systemprompt.io> for licensing details.
13
14use crate::Result;
15use chrono::{DateTime, Utc};
16use sqlx::PgPool;
17use std::sync::Arc;
18use systemprompt_database::DbPool;
19use systemprompt_identifiers::UserId;
20
21use crate::models::reporting::{
22    ModelUsageRow, RequestListFilter, RequestListRow, RequestStatsRow, RequestTrendRow,
23};
24
25#[derive(Debug)]
26pub struct RequestAnalyticsRepository {
27    pool: Arc<PgPool>,
28}
29
30impl RequestAnalyticsRepository {
31    pub fn new(db: &DbPool) -> Self {
32        let pool = db.pool();
33        Self { pool }
34    }
35
36    pub async fn get_stats(
37        &self,
38        start: DateTime<Utc>,
39        end: DateTime<Utc>,
40        model_filter: Option<&str>,
41    ) -> Result<RequestStatsRow> {
42        if let Some(model) = model_filter {
43            let pattern = format!("%{model}%");
44            sqlx::query_as!(
45                RequestStatsRow,
46                r#"
47                SELECT
48                    COALESCE(SUM(request_count), 0)::bigint as "total!",
49                    SUM(tokens)::bigint as "total_tokens",
50                    SUM(input_tokens)::bigint as "input_tokens",
51                    SUM(output_tokens)::bigint as "output_tokens",
52                    SUM(reasoning_tokens)::bigint as "reasoning_tokens",
53                    SUM(cache_read_tokens)::bigint as "cache_read_tokens",
54                    SUM(cache_creation_tokens)::bigint as "cache_creation_tokens",
55                    SUM(cost_microdollars)::bigint as "cost",
56                    (SUM(latency_sum_ms)::numeric / NULLIF(SUM(latency_n), 0))::float8 as "avg_latency",
57                    COALESCE(SUM(cache_hits), 0)::bigint as "cache_hits!"
58                FROM report_ai_usage_hourly
59                WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
60                  AND model ILIKE $3
61                "#,
62                start,
63                end,
64                pattern
65            )
66            .fetch_one(&*self.pool)
67            .await
68            .map_err(Into::into)
69        } else {
70            sqlx::query_as!(
71                RequestStatsRow,
72                r#"
73                SELECT
74                    COALESCE(SUM(request_count), 0)::bigint as "total!",
75                    SUM(tokens)::bigint as "total_tokens",
76                    SUM(input_tokens)::bigint as "input_tokens",
77                    SUM(output_tokens)::bigint as "output_tokens",
78                    SUM(reasoning_tokens)::bigint as "reasoning_tokens",
79                    SUM(cache_read_tokens)::bigint as "cache_read_tokens",
80                    SUM(cache_creation_tokens)::bigint as "cache_creation_tokens",
81                    SUM(cost_microdollars)::bigint as "cost",
82                    (SUM(latency_sum_ms)::numeric / NULLIF(SUM(latency_n), 0))::float8 as "avg_latency",
83                    COALESCE(SUM(cache_hits), 0)::bigint as "cache_hits!"
84                FROM report_ai_usage_hourly
85                WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
86                "#,
87                start,
88                end
89            )
90            .fetch_one(&*self.pool)
91            .await
92            .map_err(Into::into)
93        }
94    }
95
96    pub async fn list_models(
97        &self,
98        start: DateTime<Utc>,
99        end: DateTime<Utc>,
100        limit: i64,
101    ) -> Result<Vec<ModelUsageRow>> {
102        sqlx::query_as!(
103            ModelUsageRow,
104            r#"
105            SELECT
106                provider as "provider!",
107                model as "model!",
108                SUM(request_count)::bigint as "request_count!",
109                SUM(tokens)::bigint as "total_tokens",
110                SUM(cost_microdollars)::bigint as "total_cost",
111                (SUM(latency_sum_ms)::numeric / NULLIF(SUM(latency_n), 0))::float8 as "avg_latency"
112            FROM report_ai_usage_hourly
113            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
114              AND provider <> '' AND model <> ''
115            GROUP BY provider, model
116            ORDER BY SUM(request_count) DESC
117            LIMIT $3
118            "#,
119            start,
120            end,
121            limit
122        )
123        .fetch_all(&*self.pool)
124        .await
125        .map_err(Into::into)
126    }
127
128    pub async fn get_requests_for_trends(
129        &self,
130        start: DateTime<Utc>,
131        end: DateTime<Utc>,
132    ) -> Result<Vec<RequestTrendRow>> {
133        sqlx::query_as!(
134            RequestTrendRow,
135            r#"
136            SELECT
137                created_at as "created_at!",
138                tokens_used,
139                cost_microdollars,
140                latency_ms
141            FROM report_ai_requests
142            WHERE created_at >= $1 AND created_at < $2
143            ORDER BY created_at
144            "#,
145            start,
146            end
147        )
148        .fetch_all(&*self.pool)
149        .await
150        .map_err(Into::into)
151    }
152
153    pub async fn list_requests(
154        &self,
155        start: DateTime<Utc>,
156        end: DateTime<Utc>,
157        filter: &RequestListFilter,
158    ) -> Result<Vec<RequestListRow>> {
159        let pattern = filter.model.as_deref().map(|model| format!("%{model}%"));
160        let user = filter.user.as_ref().map(UserId::as_str);
161        sqlx::query_as!(
162            RequestListRow,
163            r#"
164            SELECT
165                id as "id!",
166                provider,
167                model,
168                input_tokens,
169                output_tokens,
170                cost_microdollars,
171                latency_ms,
172                cache_hit,
173                created_at as "created_at!",
174                status as "status!",
175                error_message,
176                user_id as "user_id!: UserId"
177            FROM report_ai_requests
178            WHERE created_at >= $1 AND created_at < $2
179              AND ($3::text IS NULL OR model ILIKE $3)
180              AND ($4::text IS NULL OR user_id = $4)
181            ORDER BY created_at DESC, id DESC
182            LIMIT $5 OFFSET $6
183            "#,
184            start,
185            end,
186            pattern,
187            user,
188            filter.limit,
189            filter.offset
190        )
191        .fetch_all(&*self.pool)
192        .await
193        .map_err(Into::into)
194    }
195}
systemprompt_analytics/repository/requests.rs

systemprompt_analytics/repository/
requests.rs
