systemprompt_logging/repository/analytics/
ingestion.rs

1//! Batched ingestion into the logging-owned analytics event store.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use async_trait::async_trait;
7use systemprompt_identifiers::SessionId;
8use systemprompt_traits::RepositoryError;
9use systemprompt_traits::analytics_events::{AnalyticsEventRecord, AnalyticsEventStore};
10
11use super::AnalyticsRepository;
12
13#[async_trait]
14impl AnalyticsEventStore for AnalyticsRepository {
15    async fn persist_events(&self, events: &[AnalyticsEventRecord]) -> Result<(), RepositoryError> {
16        if events.is_empty() {
17            return Ok(());
18        }
19
20        let ids: Vec<_> = events
21            .iter()
22            .map(|event| event.id.as_str().to_owned())
23            .collect();
24        let user_ids: Vec<_> = events
25            .iter()
26            .map(|event| event.user_id.to_string())
27            .collect();
28        let session_ids: Vec<_> = events
29            .iter()
30            .map(|event| event.session_id.to_string())
31            .collect();
32        let event_types: Vec<_> = events
33            .iter()
34            .map(|event| event.event_type.clone())
35            .collect();
36        let event_categories: Vec<_> = events
37            .iter()
38            .map(|event| event.event_category.clone())
39            .collect();
40        let severities = vec!["info".to_owned(); events.len()];
41        let endpoints: Vec<_> = events.iter().map(|event| event.page_url.clone()).collect();
42        let event_datas: Vec<_> = events
43            .iter()
44            .map(|event| event.event_data.clone())
45            .collect();
46
47        sqlx::query!(
48            r#"
49            INSERT INTO analytics_events (id, user_id, session_id, event_type, event_category, severity, endpoint, event_data)
50            SELECT * FROM UNNEST($1::text[], $2::text[], $3::text[], $4::text[], $5::text[], $6::text[], $7::text[], $8::jsonb[])
51            "#,
52            &ids,
53            &user_ids,
54            &session_ids,
55            &event_types,
56            &event_categories,
57            &severities,
58            &endpoints,
59            &event_datas
60        )
61        .execute(&*self.write_pool)
62        .await
63        .map_err(RepositoryError::database)?;
64
65        Ok(())
66    }
67
68    async fn get_endpoint_sequence(
69        &self,
70        session_id: &SessionId,
71    ) -> Result<Vec<String>, RepositoryError> {
72        let endpoints = sqlx::query_scalar!(
73            r#"
74        SELECT endpoint
75        FROM analytics_events
76        WHERE session_id = $1
77          AND event_type = 'page_view'
78        ORDER BY timestamp ASC
79        "#,
80            session_id.as_str()
81        )
82        .fetch_all(&*self.write_pool)
83        .await
84        .map_err(RepositoryError::database)?;
85        Ok(endpoints.into_iter().flatten().collect())
86    }
87
88    async fn get_request_timestamps(
89        &self,
90        session_id: &SessionId,
91    ) -> Result<Vec<chrono::DateTime<chrono::Utc>>, RepositoryError> {
92        sqlx::query_scalar!(
93            r#"
94        SELECT timestamp as "timestamp!"
95        FROM analytics_events
96        WHERE session_id = $1
97        ORDER BY timestamp ASC
98        "#,
99            session_id.as_str()
100        )
101        .fetch_all(&*self.write_pool)
102        .await
103        .map_err(RepositoryError::database)
104    }
105
106    async fn has_analytics_events(&self, session_id: &SessionId) -> Result<bool, RepositoryError> {
107        sqlx::query_scalar!(
108            r#"
109        SELECT EXISTS(
110            SELECT 1 FROM analytics_events WHERE session_id = $1
111        ) as "exists!"
112        "#,
113            session_id.as_str()
114        )
115        .fetch_one(&*self.write_pool)
116        .await
117        .map_err(RepositoryError::database)
118    }
119}
systemprompt_logging/repository/analytics/ingestion.rs

systemprompt_logging/repository/analytics/
ingestion.rs
