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

systemprompt_analytics/repository/
conversations.rs
