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_models::ContextKind;
16
17use crate::models::cli::{ConversationListRow, GatewaySessionListRow, TimestampRow};
18
19#[derive(Debug)]
20pub struct ConversationAnalyticsRepository {
21    pool: Arc<PgPool>,
22}
23
24impl ConversationAnalyticsRepository {
25    pub fn new(db: &DbPool) -> Result<Self> {
26        let pool = db.pool_arc()?;
27        Ok(Self { pool })
28    }
29
30    pub async fn list_agent_contexts(
31        &self,
32        start: DateTime<Utc>,
33        end: DateTime<Utc>,
34        limit: i64,
35    ) -> Result<Vec<ConversationListRow>> {
36        sqlx::query_as!(
37            ConversationListRow,
38            r#"
39            SELECT
40                uc.context_id as "context_id!: systemprompt_identifiers::ContextId",
41                uc.name as "name?",
42                (SELECT COUNT(*) FROM agent_tasks at WHERE at.context_id = uc.context_id)::bigint as "task_count!",
43                (SELECT COUNT(*) FROM task_messages tm
44                 JOIN agent_tasks at ON at.task_id = tm.task_id
45                 WHERE at.context_id = uc.context_id)::bigint as "message_count!",
46                uc.created_at as "created_at!",
47                uc.updated_at as "updated_at!"
48            FROM user_contexts uc
49            WHERE uc.created_at >= $1 AND uc.created_at < $2 AND uc.kind = $4
50            ORDER BY uc.updated_at DESC
51            LIMIT $3
52            "#,
53            start,
54            end,
55            limit,
56            ContextKind::User.as_str()
57        )
58        .fetch_all(&*self.pool)
59        .await
60        .map_err(Into::into)
61    }
62
63    pub async fn list_gateway_sessions(
64        &self,
65        start: DateTime<Utc>,
66        end: DateTime<Utc>,
67        limit: i64,
68    ) -> Result<Vec<GatewaySessionListRow>> {
69        sqlx::query_as!(
70            GatewaySessionListRow,
71            r#"
72            SELECT
73                ar.session_id as "session_id!: systemprompt_identifiers::SessionId",
74                COUNT(arm.id)::bigint as "message_count!",
75                MIN(ar.created_at) as "created_at!",
76                MAX(ar.created_at) as "updated_at!"
77            FROM ai_requests ar
78            LEFT JOIN ai_request_messages arm ON arm.request_id = ar.id
79            WHERE ar.task_id IS NULL
80              AND ar.session_id IS NOT NULL
81              AND ar.created_at >= $1 AND ar.created_at < $2
82              AND NOT EXISTS (
83                  SELECT 1 FROM user_contexts uc2 WHERE uc2.context_id::text = ar.session_id
84              )
85            GROUP BY ar.session_id
86            ORDER BY MAX(ar.created_at) DESC
87            LIMIT $3
88            "#,
89            start,
90            end,
91            limit
92        )
93        .fetch_all(&*self.pool)
94        .await
95        .map_err(Into::into)
96    }
97
98    pub async fn get_context_count(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<i64> {
99        let count = sqlx::query_scalar!(
100            r#"SELECT COUNT(*)::bigint as "count!" FROM user_contexts WHERE created_at >= $1 AND created_at < $2 AND kind = $3"#,
101            start,
102            end,
103            ContextKind::User.as_str()
104        )
105        .fetch_one(&*self.pool)
106        .await?;
107        Ok(count)
108    }
109
110    pub async fn get_task_stats(
111        &self,
112        start: DateTime<Utc>,
113        end: DateTime<Utc>,
114    ) -> Result<(i64, Option<f64>)> {
115        let row = sqlx::query!(
116            r#"
117            SELECT COUNT(*)::bigint as "count!", AVG(execution_time_ms)::float8 as avg_time
118            FROM agent_tasks
119            WHERE started_at >= $1 AND started_at < $2
120            "#,
121            start,
122            end
123        )
124        .fetch_one(&*self.pool)
125        .await?;
126        Ok((row.count, row.avg_time))
127    }
128
129    pub async fn get_message_count(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<i64> {
130        let count = sqlx::query_scalar!(
131            r#"SELECT COUNT(*)::bigint as "count!" FROM task_messages WHERE created_at >= $1 AND created_at < $2"#,
132            start,
133            end
134        )
135        .fetch_one(&*self.pool)
136        .await?;
137        Ok(count)
138    }
139
140    pub async fn get_context_timestamps(
141        &self,
142        start: DateTime<Utc>,
143        end: DateTime<Utc>,
144    ) -> Result<Vec<TimestampRow>> {
145        sqlx::query_as!(
146            TimestampRow,
147            r#"
148            SELECT created_at as "timestamp!"
149            FROM user_contexts
150            WHERE created_at >= $1 AND created_at < $2 AND kind = $3
151            "#,
152            start,
153            end,
154            ContextKind::User.as_str()
155        )
156        .fetch_all(&*self.pool)
157        .await
158        .map_err(Into::into)
159    }
160
161    pub async fn get_task_timestamps(
162        &self,
163        start: DateTime<Utc>,
164        end: DateTime<Utc>,
165    ) -> Result<Vec<TimestampRow>> {
166        sqlx::query_as!(
167            TimestampRow,
168            r#"
169            SELECT started_at as "timestamp!"
170            FROM agent_tasks
171            WHERE started_at >= $1 AND started_at < $2
172            "#,
173            start,
174            end
175        )
176        .fetch_all(&*self.pool)
177        .await
178        .map_err(Into::into)
179    }
180
181    pub async fn get_message_timestamps(
182        &self,
183        start: DateTime<Utc>,
184        end: DateTime<Utc>,
185    ) -> Result<Vec<TimestampRow>> {
186        sqlx::query_as!(
187            TimestampRow,
188            r#"
189            SELECT created_at as "timestamp!"
190            FROM task_messages
191            WHERE created_at >= $1 AND created_at < $2
192            "#,
193            start,
194            end
195        )
196        .fetch_all(&*self.pool)
197        .await
198        .map_err(Into::into)
199    }
200}
systemprompt_analytics/repository/conversations.rs

systemprompt_analytics/repository/
conversations.rs
