systemprompt_analytics/repository/
cli_sessions.rs

1//! Session analytics over the `user_sessions` table.
2//!
3//! [`CliSessionAnalyticsRepository`] reads session counts, durations, live
4//! activity, and conversion stats for human (non-bot) sessions; every query
5//! filters out bot, behavioural-bot, and scanner traffic.
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;
16
17use crate::models::cli::{LiveSessionRow, SessionStatsRow, SessionTrendRow};
18
19#[derive(Debug)]
20pub struct CliSessionAnalyticsRepository {
21    pool: Arc<PgPool>,
22}
23
24impl CliSessionAnalyticsRepository {
25    pub fn new(db: &DbPool) -> Result<Self> {
26        let pool = db.pool_arc()?;
27        Ok(Self { pool })
28    }
29
30    pub async fn get_stats(
31        &self,
32        start: DateTime<Utc>,
33        end: DateTime<Utc>,
34    ) -> Result<SessionStatsRow> {
35        sqlx::query_as!(
36            SessionStatsRow,
37            r#"
38            SELECT
39                COUNT(*)::bigint as "total_sessions!",
40                COUNT(DISTINCT user_id)::bigint as "unique_users!",
41                AVG(LEAST(
42                    COALESCE(
43                        duration_seconds,
44                        EXTRACT(EPOCH FROM (COALESCE(ended_at, last_activity_at) - started_at))::INTEGER
45                    ),
46                    1800
47                ))::float8 as "avg_duration",
48                AVG(request_count)::float8 as "avg_requests",
49                COUNT(*) FILTER (WHERE converted_at IS NOT NULL)::bigint as "conversions!"
50            FROM user_sessions
51            WHERE started_at >= $1 AND started_at < $2
52              AND is_bot = false AND is_behavioral_bot = false AND is_scanner = false
53            "#,
54            start,
55            end
56        )
57        .fetch_one(&*self.pool)
58        .await
59        .map_err(Into::into)
60    }
61
62    pub async fn get_active_session_count(&self, since: DateTime<Utc>) -> Result<i64> {
63        let count = sqlx::query_scalar!(
64            r#"SELECT COUNT(*)::bigint as "count!" FROM user_sessions WHERE ended_at IS NULL AND last_activity_at >= $1 AND is_bot = false AND is_behavioral_bot = false AND is_scanner = false"#,
65            since
66        )
67        .fetch_one(&*self.pool)
68        .await?;
69        Ok(count)
70    }
71
72    pub async fn get_live_sessions(
73        &self,
74        cutoff: DateTime<Utc>,
75        limit: i64,
76    ) -> Result<Vec<LiveSessionRow>> {
77        sqlx::query_as!(
78            LiveSessionRow,
79            r#"
80            SELECT
81                session_id as "session_id!",
82                COALESCE(user_type, 'unknown') as "user_type",
83                started_at as "started_at!",
84                duration_seconds,
85                request_count,
86                last_activity_at as "last_activity_at!"
87            FROM user_sessions
88            WHERE ended_at IS NULL
89              AND last_activity_at >= $1
90              AND is_bot = false AND is_behavioral_bot = false AND is_scanner = false
91            ORDER BY last_activity_at DESC
92            LIMIT $2
93            "#,
94            cutoff,
95            limit
96        )
97        .fetch_all(&*self.pool)
98        .await
99        .map_err(Into::into)
100    }
101
102    pub async fn get_active_count(&self, cutoff: DateTime<Utc>) -> Result<i64> {
103        let count = sqlx::query_scalar!(
104            r#"SELECT COUNT(*)::bigint as "count!" FROM user_sessions WHERE ended_at IS NULL AND last_activity_at >= $1 AND is_bot = false AND is_behavioral_bot = false AND is_scanner = false"#,
105            cutoff
106        )
107        .fetch_one(&*self.pool)
108        .await?;
109        Ok(count)
110    }
111
112    pub async fn get_sessions_for_trends(
113        &self,
114        start: DateTime<Utc>,
115        end: DateTime<Utc>,
116    ) -> Result<Vec<SessionTrendRow>> {
117        sqlx::query_as!(
118            SessionTrendRow,
119            r#"
120            SELECT
121                started_at as "started_at!",
122                user_id as "user_id: UserId",
123                duration_seconds
124            FROM user_sessions
125            WHERE started_at >= $1 AND started_at < $2
126              AND is_bot = false AND is_behavioral_bot = false AND is_scanner = false
127            ORDER BY started_at
128            "#,
129            start,
130            end
131        )
132        .fetch_all(&*self.pool)
133        .await
134        .map_err(Into::into)
135    }
136
137    pub async fn get_active_count_since(&self, start: DateTime<Utc>) -> Result<i64> {
138        let count = sqlx::query_scalar!(
139            r#"
140            SELECT COUNT(*)::bigint as "count!"
141            FROM user_sessions
142            WHERE ended_at IS NULL
143              AND last_activity_at >= $1
144              AND is_bot = false AND is_behavioral_bot = false AND is_scanner = false
145            "#,
146            start
147        )
148        .fetch_one(&*self.pool)
149        .await?;
150        Ok(count)
151    }
152
153    pub async fn get_total_count(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<i64> {
154        let count = sqlx::query_scalar!(
155            r#"SELECT COUNT(*)::bigint as "count!" FROM user_sessions WHERE started_at >= $1 AND started_at < $2 AND is_bot = false AND is_behavioral_bot = false AND is_scanner = false"#,
156            start,
157            end
158        )
159        .fetch_one(&*self.pool)
160        .await?;
161        Ok(count)
162    }
163}
systemprompt_analytics/repository/cli_sessions.rs

systemprompt_analytics/repository/
cli_sessions.rs
