systemprompt_analytics/repository/
conversations.rs

1//! Conversation analytics over agent contexts and gateway sessions.
2//!
3//! [`ConversationAnalyticsRepository`] lists agent-task contexts and
4//! task-less gateway AI sessions, and reports task, message, and timestamp
5//! counts used to build conversation activity trends.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use crate::Result;
11use chrono::{DateTime, Utc};
12use sqlx::PgPool;
13use std::sync::Arc;
14use systemprompt_database::DbPool;
15use systemprompt_identifiers::UserId;
16use systemprompt_models::ContextKind;
17
18use crate::models::reporting::{ConversationListRow, GatewaySessionListRow, TimestampRow};
19
20#[derive(Debug)]
21pub struct ConversationAnalyticsRepository {
22    pool: Arc<PgPool>,
23}
24
25impl ConversationAnalyticsRepository {
26    pub fn new(db: &DbPool) -> Result<Self> {
27        let pool = db.pool_arc()?;
28        Ok(Self { pool })
29    }
30
31    pub async fn list_agent_contexts(
32        &self,
33        start: DateTime<Utc>,
34        end: DateTime<Utc>,
35        limit: i64,
36        user: Option<&UserId>,
37    ) -> Result<Vec<ConversationListRow>> {
38        let user = user.map(UserId::as_str);
39        sqlx::query_as!(
40            ConversationListRow,
41            r#"
42            SELECT
43                uc.context_id as "context_id!: systemprompt_identifiers::ContextId",
44                uc.user_id as "user_id!: systemprompt_identifiers::UserId",
45                uc.name as "name?",
46                (SELECT COUNT(*) FROM analytics_report_agent_tasks at WHERE at.context_id = uc.context_id)::bigint as "task_count!",
47                (SELECT COUNT(*) FROM analytics_report_task_messages tm
48                 JOIN analytics_report_agent_tasks at ON at.task_id = tm.task_id
49                 WHERE at.context_id = uc.context_id)::bigint as "message_count!",
50                uc.created_at as "created_at!",
51                uc.updated_at as "updated_at!"
52            FROM analytics_report_user_contexts uc
53            WHERE uc.created_at >= $1 AND uc.created_at < $2 AND uc.kind = $4
54              AND ($5::text IS NULL OR uc.user_id = $5)
55            ORDER BY uc.updated_at DESC
56            LIMIT $3
57            "#,
58            start,
59            end,
60            limit,
61            ContextKind::User.as_str(),
62            user
63        )
64        .fetch_all(&*self.pool)
65        .await
66        .map_err(Into::into)
67    }
68
69    pub async fn list_gateway_sessions(
70        &self,
71        start: DateTime<Utc>,
72        end: DateTime<Utc>,
73        limit: i64,
74        user: Option<&UserId>,
75    ) -> Result<Vec<GatewaySessionListRow>> {
76        let user = user.map(UserId::as_str);
77        sqlx::query_as!(
78            GatewaySessionListRow,
79            r#"
80            SELECT
81                ar.session_id as "session_id!: systemprompt_identifiers::SessionId",
82                MIN(ar.user_id) as "user_id!: systemprompt_identifiers::UserId",
83                COUNT(arm.id)::bigint as "message_count!",
84                MIN(ar.created_at) as "created_at!",
85                MAX(ar.created_at) as "updated_at!"
86            FROM analytics_report_ai_requests ar
87            LEFT JOIN analytics_report_ai_request_messages arm ON arm.request_id = ar.id
88            WHERE ar.task_id IS NULL
89              AND ar.session_id IS NOT NULL
90              AND ar.created_at >= $1 AND ar.created_at < $2
91              AND ($4::text IS NULL OR ar.user_id = $4)
92              AND NOT EXISTS (
93                  SELECT 1 FROM analytics_report_user_contexts uc2 WHERE uc2.context_id::text = ar.session_id
94              )
95            GROUP BY ar.session_id
96            ORDER BY MAX(ar.created_at) DESC
97            LIMIT $3
98            "#,
99            start,
100            end,
101            limit,
102            user
103        )
104        .fetch_all(&*self.pool)
105        .await
106        .map_err(Into::into)
107    }
108
109    pub async fn get_context_count(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<i64> {
110        let count = sqlx::query_scalar!(
111            r#"SELECT COUNT(*)::bigint as "count!" FROM analytics_report_user_contexts WHERE created_at >= $1 AND created_at < $2 AND kind = $3"#,
112            start,
113            end,
114            ContextKind::User.as_str()
115        )
116        .fetch_one(&*self.pool)
117        .await?;
118        Ok(count)
119    }
120
121    pub async fn get_task_stats(
122        &self,
123        start: DateTime<Utc>,
124        end: DateTime<Utc>,
125    ) -> Result<(i64, Option<f64>)> {
126        let row = sqlx::query!(
127            r#"
128            SELECT COUNT(*)::bigint as "count!", AVG(execution_time_ms)::float8 as avg_time
129            FROM analytics_report_agent_tasks
130            WHERE started_at >= $1 AND started_at < $2
131            "#,
132            start,
133            end
134        )
135        .fetch_one(&*self.pool)
136        .await?;
137        Ok((row.count, row.avg_time))
138    }
139
140    pub async fn get_message_count(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<i64> {
141        let count = sqlx::query_scalar!(
142            r#"SELECT COUNT(*)::bigint as "count!" FROM analytics_report_task_messages WHERE created_at >= $1 AND created_at < $2"#,
143            start,
144            end
145        )
146        .fetch_one(&*self.pool)
147        .await?;
148        Ok(count)
149    }
150
151    pub async fn get_context_timestamps(
152        &self,
153        start: DateTime<Utc>,
154        end: DateTime<Utc>,
155    ) -> Result<Vec<TimestampRow>> {
156        sqlx::query_as!(
157            TimestampRow,
158            r#"
159            SELECT created_at as "timestamp!"
160            FROM analytics_report_user_contexts
161            WHERE created_at >= $1 AND created_at < $2 AND kind = $3
162            "#,
163            start,
164            end,
165            ContextKind::User.as_str()
166        )
167        .fetch_all(&*self.pool)
168        .await
169        .map_err(Into::into)
170    }
171
172    pub async fn get_task_timestamps(
173        &self,
174        start: DateTime<Utc>,
175        end: DateTime<Utc>,
176    ) -> Result<Vec<TimestampRow>> {
177        sqlx::query_as!(
178            TimestampRow,
179            r#"
180            SELECT started_at as "timestamp!"
181            FROM analytics_report_agent_tasks
182            WHERE started_at >= $1 AND started_at < $2
183            "#,
184            start,
185            end
186        )
187        .fetch_all(&*self.pool)
188        .await
189        .map_err(Into::into)
190    }
191
192    pub async fn get_message_timestamps(
193        &self,
194        start: DateTime<Utc>,
195        end: DateTime<Utc>,
196    ) -> Result<Vec<TimestampRow>> {
197        sqlx::query_as!(
198            TimestampRow,
199            r#"
200            SELECT created_at as "timestamp!"
201            FROM analytics_report_task_messages
202            WHERE created_at >= $1 AND created_at < $2
203            "#,
204            start,
205            end
206        )
207        .fetch_all(&*self.pool)
208        .await
209        .map_err(Into::into)
210    }
211}
systemprompt_analytics/repository/conversations.rs

systemprompt_analytics/repository/
conversations.rs
