systemprompt_analytics/repository/
requests.rs

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

systemprompt_analytics/repository/
requests.rs
