systemprompt_analytics/repository/
engagement.rs

1//! Persistence for page-level engagement telemetry.
2//!
3//! [`EngagementRepository`] records [`EngagementEvent`]s (scroll, click,
4//! focus, and reading-pattern metrics) and reads them back per session or
5//! user, plus the aggregated [`SessionEngagementSummary`]. Writes go to the
6//! write pool; reads to the read pool.
7//!
8//! Copyright (c) systemprompt.io — Business Source License 1.1.
9//! See <https://systemprompt.io> for licensing details.
10
11use std::sync::Arc;
12
13use crate::Result;
14use sqlx::PgPool;
15use systemprompt_database::DbPool;
16use systemprompt_identifiers::{ContentId, EngagementEventId, SessionId, UserId};
17
18use crate::models::{CreateEngagementEventInput, EngagementEvent};
19
20#[derive(Clone, Debug)]
21pub struct EngagementRepository {
22    pool: Arc<PgPool>,
23    write_pool: Arc<PgPool>,
24}
25
26impl EngagementRepository {
27    pub fn new(db: &DbPool) -> Result<Self> {
28        let pool = db.pool_arc()?;
29        let write_pool = db.write_pool_arc()?;
30        Ok(Self { pool, write_pool })
31    }
32
33    pub async fn create_engagement(
34        &self,
35        session_id: &SessionId,
36        user_id: &UserId,
37        content_id: Option<&ContentId>,
38        input: &CreateEngagementEventInput,
39    ) -> Result<EngagementEventId> {
40        let id = EngagementEventId::generate();
41
42        sqlx::query!(
43            r#"
44            INSERT INTO engagement_events (
45                id, session_id, user_id, page_url, content_id, event_type,
46                time_on_page_ms, max_scroll_depth, click_count,
47                time_to_first_interaction_ms, time_to_first_scroll_ms,
48                scroll_velocity_avg, scroll_direction_changes,
49                mouse_move_distance_px, keyboard_events, copy_events,
50                focus_time_ms, blur_count, tab_switches, visible_time_ms, hidden_time_ms,
51                is_rage_click, is_dead_click, reading_pattern, event_data
52            )
53            VALUES (
54                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13,
55                $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25
56            )
57            "#,
58            id.as_str(),
59            session_id.as_str(),
60            user_id.as_str(),
61            input.page_url,
62            content_id.map(ContentId::as_str),
63            input.event_type.as_str(),
64            input.time_on_page_ms,
65            input.max_scroll_depth,
66            input.click_count,
67            input.optional_metrics.time_to_first_interaction_ms,
68            input.optional_metrics.time_to_first_scroll_ms,
69            input.optional_metrics.scroll_velocity_avg,
70            input.optional_metrics.scroll_direction_changes,
71            input.optional_metrics.mouse_move_distance_px,
72            input.optional_metrics.keyboard_events,
73            input.optional_metrics.copy_events,
74            input.optional_metrics.focus_time_ms.unwrap_or(0),
75            input.optional_metrics.blur_count.unwrap_or(0),
76            input.optional_metrics.tab_switches.unwrap_or(0),
77            input.optional_metrics.visible_time_ms.unwrap_or(0),
78            input.optional_metrics.hidden_time_ms.unwrap_or(0),
79            input.optional_metrics.is_rage_click,
80            input.optional_metrics.is_dead_click,
81            input.optional_metrics.reading_pattern,
82            input.event_data.clone()
83        )
84        .execute(&*self.write_pool)
85        .await?;
86
87        Ok(id)
88    }
89
90    pub async fn find_by_id(&self, id: &EngagementEventId) -> Result<Option<EngagementEvent>> {
91        let event = sqlx::query_as!(
92            EngagementEvent,
93            r#"
94            SELECT
95                id as "id: EngagementEventId", session_id, user_id, page_url,
96                content_id as "content_id: ContentId",
97                event_type,
98                time_on_page_ms, time_to_first_interaction_ms, time_to_first_scroll_ms,
99                max_scroll_depth, scroll_velocity_avg, scroll_direction_changes,
100                click_count, mouse_move_distance_px, keyboard_events, copy_events,
101                focus_time_ms as "focus_time_ms!",
102                blur_count as "blur_count!",
103                tab_switches as "tab_switches!",
104                visible_time_ms as "visible_time_ms!",
105                hidden_time_ms as "hidden_time_ms!",
106                is_rage_click, is_dead_click, reading_pattern,
107                created_at, updated_at
108            FROM engagement_events
109            WHERE id = $1
110            "#,
111            id.as_str()
112        )
113        .fetch_optional(&*self.pool)
114        .await?;
115
116        Ok(event)
117    }
118
119    pub async fn list_by_session(&self, session_id: &SessionId) -> Result<Vec<EngagementEvent>> {
120        let events = sqlx::query_as!(
121            EngagementEvent,
122            r#"
123            SELECT
124                id as "id: EngagementEventId", session_id, user_id, page_url,
125                content_id as "content_id: ContentId",
126                event_type,
127                time_on_page_ms as "time_on_page_ms!", time_to_first_interaction_ms, time_to_first_scroll_ms,
128                max_scroll_depth as "max_scroll_depth!", scroll_velocity_avg, scroll_direction_changes,
129                click_count as "click_count!", mouse_move_distance_px, keyboard_events, copy_events,
130                focus_time_ms as "focus_time_ms!",
131                blur_count as "blur_count!",
132                tab_switches as "tab_switches!",
133                visible_time_ms as "visible_time_ms!",
134                hidden_time_ms as "hidden_time_ms!",
135                is_rage_click, is_dead_click, reading_pattern,
136                created_at, updated_at
137            FROM engagement_events
138            WHERE session_id = $1
139            ORDER BY created_at ASC
140            "#,
141            session_id.as_str()
142        )
143        .fetch_all(&*self.pool)
144        .await?;
145
146        Ok(events)
147    }
148
149    pub async fn list_by_user(&self, user_id: &UserId, limit: i64) -> Result<Vec<EngagementEvent>> {
150        let events = sqlx::query_as!(
151            EngagementEvent,
152            r#"
153            SELECT
154                id as "id: EngagementEventId", session_id, user_id, page_url,
155                content_id as "content_id: ContentId",
156                event_type,
157                time_on_page_ms as "time_on_page_ms!", time_to_first_interaction_ms, time_to_first_scroll_ms,
158                max_scroll_depth as "max_scroll_depth!", scroll_velocity_avg, scroll_direction_changes,
159                click_count as "click_count!", mouse_move_distance_px, keyboard_events, copy_events,
160                focus_time_ms as "focus_time_ms!",
161                blur_count as "blur_count!",
162                tab_switches as "tab_switches!",
163                visible_time_ms as "visible_time_ms!",
164                hidden_time_ms as "hidden_time_ms!",
165                is_rage_click, is_dead_click, reading_pattern,
166                created_at, updated_at
167            FROM engagement_events
168            WHERE user_id = $1
169            ORDER BY created_at DESC
170            LIMIT $2
171            "#,
172            user_id.as_str(),
173            limit
174        )
175        .fetch_all(&*self.pool)
176        .await?;
177
178        Ok(events)
179    }
180
181    pub async fn get_session_engagement_summary(
182        &self,
183        session_id: &SessionId,
184    ) -> Result<Option<SessionEngagementSummary>> {
185        let summary = sqlx::query_as!(
186            SessionEngagementSummary,
187            r#"
188            SELECT
189                session_id,
190                COUNT(*)::BIGINT as page_count,
191                SUM(time_on_page_ms)::BIGINT as total_time_on_page_ms,
192                AVG(max_scroll_depth)::REAL as avg_scroll_depth,
193                MAX(max_scroll_depth) as max_scroll_depth,
194                SUM(click_count)::BIGINT as total_clicks,
195                COUNT(*) FILTER (WHERE is_rage_click = true)::BIGINT as rage_click_pages,
196                MIN(created_at) as first_engagement,
197                MAX(created_at) as last_engagement
198            FROM engagement_events
199            WHERE session_id = $1
200            GROUP BY session_id
201            "#,
202            session_id.as_str()
203        )
204        .fetch_optional(&*self.pool)
205        .await?;
206
207        Ok(summary)
208    }
209}
210
211#[derive(Debug, Clone, sqlx::FromRow)]
212pub struct SessionEngagementSummary {
213    pub session_id: SessionId,
214    pub page_count: Option<i64>,
215    pub total_time_on_page_ms: Option<i64>,
216    pub avg_scroll_depth: Option<f32>,
217    pub max_scroll_depth: Option<i32>,
218    pub total_clicks: Option<i64>,
219    pub rage_click_pages: Option<i64>,
220    pub first_engagement: Option<chrono::DateTime<chrono::Utc>>,
221    pub last_engagement: Option<chrono::DateTime<chrono::Utc>>,
222}
systemprompt_analytics/repository/engagement.rs

systemprompt_analytics/repository/
engagement.rs
