systemprompt_analytics/repository/
events.rs

1//! Persistence for raw analytics events.
2//!
3//! [`AnalyticsEventsRepository`] writes individual and batched
4//! `analytics_events` rows (the batch path uses `UNNEST` for a single
5//! round-trip) and reads them back per session or content as
6//! [`StoredAnalyticsEvent`]. Writes target the write pool; reads the read
7//! pool.
8//!
9//! Copyright (c) systemprompt.io — Business Source License 1.1.
10//! See <https://systemprompt.io> for licensing details.
11
12use std::sync::Arc;
13
14use crate::Result;
15use sqlx::PgPool;
16use systemprompt_database::DbPool;
17use systemprompt_identifiers::{ContentId, SessionId, UserId};
18
19use crate::models::{AnalyticsEventCreated, AnalyticsEventType, CreateAnalyticsEventInput};
20
21#[derive(Clone, Debug)]
22pub struct AnalyticsEventsRepository {
23    pool: Arc<PgPool>,
24    write_pool: Arc<PgPool>,
25}
26
27impl AnalyticsEventsRepository {
28    pub fn new(db: &DbPool) -> Result<Self> {
29        let pool = db.pool_arc()?;
30        let write_pool = db.write_pool_arc()?;
31        Ok(Self { pool, write_pool })
32    }
33
34    pub async fn create_event(
35        &self,
36        session_id: &SessionId,
37        user_id: &UserId,
38        input: &CreateAnalyticsEventInput,
39    ) -> Result<AnalyticsEventCreated> {
40        let id = format!("evt_{}", uuid::Uuid::new_v4());
41        let event_type = input.event_type.as_str();
42        let event_category = input.event_type.category();
43
44        let event_data = Self::build_event_data(input);
45
46        sqlx::query!(
47            r#"
48            INSERT INTO analytics_events (
49                id, user_id, session_id, event_type, event_category,
50                severity, endpoint, event_data
51            )
52            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
53            "#,
54            id,
55            user_id.as_str(),
56            session_id.as_str(),
57            event_type,
58            event_category,
59            "info",
60            input.page_url,
61            event_data
62        )
63        .execute(&*self.write_pool)
64        .await?;
65
66        Ok(AnalyticsEventCreated {
67            id,
68            event_type: event_type.to_owned(),
69        })
70    }
71
72    pub async fn create_events_batch(
73        &self,
74        session_id: &SessionId,
75        user_id: &UserId,
76        inputs: &[CreateAnalyticsEventInput],
77    ) -> Result<Vec<AnalyticsEventCreated>> {
78        if inputs.is_empty() {
79            return Ok(Vec::new());
80        }
81
82        let mut ids = Vec::with_capacity(inputs.len());
83        let mut user_ids = Vec::with_capacity(inputs.len());
84        let mut session_ids = Vec::with_capacity(inputs.len());
85        let mut event_types = Vec::with_capacity(inputs.len());
86        let mut event_categories = Vec::with_capacity(inputs.len());
87        let mut severities = Vec::with_capacity(inputs.len());
88        let mut endpoints: Vec<String> = Vec::with_capacity(inputs.len());
89        let mut event_datas = Vec::with_capacity(inputs.len());
90
91        for input in inputs {
92            let id = format!("evt_{}", uuid::Uuid::new_v4());
93            ids.push(id);
94            user_ids.push(user_id.as_str().to_owned());
95            session_ids.push(session_id.as_str().to_owned());
96            event_types.push(input.event_type.as_str().to_owned());
97            event_categories.push(input.event_type.category().to_owned());
98            severities.push("info".to_owned());
99            endpoints.push(input.page_url.clone());
100            event_datas.push(Self::build_event_data(input));
101        }
102
103        sqlx::query!(
104            r#"
105            INSERT INTO analytics_events (id, user_id, session_id, event_type, event_category, severity, endpoint, event_data)
106            SELECT * FROM UNNEST($1::text[], $2::text[], $3::text[], $4::text[], $5::text[], $6::text[], $7::text[], $8::jsonb[])
107            "#,
108            &ids,
109            &user_ids,
110            &session_ids,
111            &event_types,
112            &event_categories,
113            &severities,
114            &endpoints,
115            &event_datas
116        )
117        .execute(&*self.write_pool)
118        .await?;
119
120        Ok(ids
121            .into_iter()
122            .zip(event_types)
123            .map(|(id, event_type)| AnalyticsEventCreated { id, event_type })
124            .collect())
125    }
126
127    pub async fn count_events_by_type(
128        &self,
129        session_id: &SessionId,
130        event_type: &AnalyticsEventType,
131    ) -> Result<i64> {
132        let count = sqlx::query_scalar!(
133            r#"
134            SELECT COUNT(*) as "count!"
135            FROM analytics_events
136            WHERE session_id = $1 AND event_type = $2
137            "#,
138            session_id.as_str(),
139            event_type.as_str()
140        )
141        .fetch_one(&*self.pool)
142        .await?;
143
144        Ok(count)
145    }
146
147    pub async fn find_by_session(
148        &self,
149        session_id: &SessionId,
150        limit: i64,
151    ) -> Result<Vec<StoredAnalyticsEvent>> {
152        let events = sqlx::query_as!(
153            StoredAnalyticsEvent,
154            r#"
155            SELECT
156                id,
157                user_id as "user_id: UserId",
158                session_id as "session_id: SessionId",
159                event_type,
160                event_category,
161                endpoint as page_url,
162                event_data,
163                timestamp
164            FROM analytics_events
165            WHERE session_id = $1
166            ORDER BY timestamp DESC
167            LIMIT $2
168            "#,
169            session_id.as_str(),
170            limit
171        )
172        .fetch_all(&*self.pool)
173        .await?;
174
175        Ok(events)
176    }
177
178    pub async fn find_by_content(
179        &self,
180        content_id: &ContentId,
181        limit: i64,
182    ) -> Result<Vec<StoredAnalyticsEvent>> {
183        let events = sqlx::query_as!(
184            StoredAnalyticsEvent,
185            r#"
186            SELECT
187                id,
188                user_id as "user_id: UserId",
189                session_id as "session_id: SessionId",
190                event_type,
191                event_category,
192                endpoint as page_url,
193                event_data,
194                timestamp
195            FROM analytics_events
196            WHERE event_data->>'content_id' = $1
197            ORDER BY timestamp DESC
198            LIMIT $2
199            "#,
200            content_id.as_str(),
201            limit
202        )
203        .fetch_all(&*self.pool)
204        .await?;
205
206        Ok(events)
207    }
208
209    fn build_event_data(input: &CreateAnalyticsEventInput) -> serde_json::Value {
210        let mut data = input.data.clone().unwrap_or(serde_json::json!({}));
211
212        if let Some(obj) = data.as_object_mut() {
213            if let Some(content_id) = &input.content_id {
214                obj.insert(
215                    "content_id".to_owned(),
216                    serde_json::json!(content_id.as_str()),
217                );
218            }
219            if let Some(slug) = &input.slug {
220                obj.insert("slug".to_owned(), serde_json::json!(slug));
221            }
222            if let Some(referrer) = &input.referrer {
223                obj.insert("referrer".to_owned(), serde_json::json!(referrer));
224            }
225        }
226
227        data
228    }
229}
230
231#[derive(Debug, Clone, sqlx::FromRow)]
232pub struct StoredAnalyticsEvent {
233    pub id: String,
234    pub user_id: UserId,
235    pub session_id: Option<SessionId>,
236    pub event_type: String,
237    pub event_category: String,
238    pub page_url: Option<String>,
239    pub event_data: Option<serde_json::Value>,
240    pub timestamp: chrono::DateTime<chrono::Utc>,
241}
systemprompt_analytics/repository/events.rs

systemprompt_analytics/repository/
events.rs
