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

systemprompt_logging/repository/analytics/
ingestion.rs
