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