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                COALESCE(SUM(ar.message_count), 0)::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            WHERE ar.task_id IS NULL
88              AND ar.session_id IS NOT NULL
89              AND ar.created_at >= $1 AND ar.created_at < $2
90              AND ($4::text IS NULL OR ar.user_id = $4)
91              AND NOT EXISTS (
92                  SELECT 1 FROM analytics_report_user_contexts uc2 WHERE uc2.context_id::text = ar.session_id
93              )
94            GROUP BY ar.session_id
95            ORDER BY MAX(ar.created_at) DESC
96            LIMIT $3
97            "#,
98            start,
99            end,
100            limit,
101            user
102        )
103        .fetch_all(&*self.pool)
104        .await
105        .map_err(Into::into)
106    }
107
108    pub async fn get_context_count(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<i64> {
109        let count = sqlx::query_scalar!(
110            r#"SELECT COUNT(*)::bigint as "count!" FROM analytics_report_user_contexts WHERE created_at >= $1 AND created_at < $2 AND kind = $3"#,
111            start,
112            end,
113            ContextKind::User.as_str()
114        )
115        .fetch_one(&*self.pool)
116        .await?;
117        Ok(count)
118    }
119
120    pub async fn get_task_stats(
121        &self,
122        start: DateTime<Utc>,
123        end: DateTime<Utc>,
124    ) -> Result<(i64, Option<f64>)> {
125        let row = sqlx::query!(
126            r#"
127            SELECT COUNT(*)::bigint as "count!", AVG(execution_time_ms)::float8 as avg_time
128            FROM analytics_report_agent_tasks
129            WHERE started_at >= $1 AND started_at < $2
130            "#,
131            start,
132            end
133        )
134        .fetch_one(&*self.pool)
135        .await?;
136        Ok((row.count, row.avg_time))
137    }
138
139    pub async fn get_message_count(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<i64> {
140        let count = sqlx::query_scalar!(
141            r#"SELECT COUNT(*)::bigint as "count!" FROM analytics_report_task_messages WHERE created_at >= $1 AND created_at < $2"#,
142            start,
143            end
144        )
145        .fetch_one(&*self.pool)
146        .await?;
147        Ok(count)
148    }
149
150    pub async fn get_context_timestamps(
151        &self,
152        start: DateTime<Utc>,
153        end: DateTime<Utc>,
154    ) -> Result<Vec<TimestampRow>> {
155        sqlx::query_as!(
156            TimestampRow,
157            r#"
158            SELECT created_at as "timestamp!"
159            FROM analytics_report_user_contexts
160            WHERE created_at >= $1 AND created_at < $2 AND kind = $3
161            "#,
162            start,
163            end,
164            ContextKind::User.as_str()
165        )
166        .fetch_all(&*self.pool)
167        .await
168        .map_err(Into::into)
169    }
170
171    pub async fn get_task_timestamps(
172        &self,
173        start: DateTime<Utc>,
174        end: DateTime<Utc>,
175    ) -> Result<Vec<TimestampRow>> {
176        sqlx::query_as!(
177            TimestampRow,
178            r#"
179            SELECT started_at as "timestamp!"
180            FROM analytics_report_agent_tasks
181            WHERE started_at >= $1 AND started_at < $2
182            "#,
183            start,
184            end
185        )
186        .fetch_all(&*self.pool)
187        .await
188        .map_err(Into::into)
189    }
190
191    pub async fn get_message_timestamps(
192        &self,
193        start: DateTime<Utc>,
194        end: DateTime<Utc>,
195    ) -> Result<Vec<TimestampRow>> {
196        sqlx::query_as!(
197            TimestampRow,
198            r#"
199            SELECT created_at as "timestamp!"
200            FROM analytics_report_task_messages
201            WHERE created_at >= $1 AND created_at < $2
202            "#,
203            start,
204            end
205        )
206        .fetch_all(&*self.pool)
207        .await
208        .map_err(Into::into)
209    }
210}
systemprompt_analytics/repository/conversations.rs

systemprompt_analytics/repository/
conversations.rs
