systemprompt_analytics/repository/
events.rs

1//! Raw analytics ingestion through the logging sink.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use crate::Result;
7use systemprompt_identifiers::{SessionId, UserId};
8use systemprompt_traits::analytics_events::{AnalyticsEventRecord, DynAnalyticsEventStore};
9
10use crate::models::{AnalyticsEventCreated, CreateAnalyticsEventInput};
11
12#[derive(Clone, Debug)]
13pub struct AnalyticsEventsRepository {
14    event_sink: DynAnalyticsEventStore,
15}
16
17impl AnalyticsEventsRepository {
18    pub const fn new(event_sink: DynAnalyticsEventStore) -> Self {
19        Self { event_sink }
20    }
21
22    pub async fn create_event(
23        &self,
24        session_id: &SessionId,
25        user_id: &UserId,
26        input: &CreateAnalyticsEventInput,
27    ) -> Result<AnalyticsEventCreated> {
28        let event = Self::build_record(session_id, user_id, input);
29        self.event_sink
30            .persist_events(std::slice::from_ref(&event))
31            .await?;
32        Ok(AnalyticsEventCreated {
33            id: event.id,
34            event_type: event.event_type,
35        })
36    }
37
38    pub async fn create_events_batch(
39        &self,
40        session_id: &SessionId,
41        user_id: &UserId,
42        inputs: &[CreateAnalyticsEventInput],
43    ) -> Result<Vec<AnalyticsEventCreated>> {
44        if inputs.is_empty() {
45            return Ok(Vec::new());
46        }
47
48        let events: Vec<_> = inputs
49            .iter()
50            .map(|input| Self::build_record(session_id, user_id, input))
51            .collect();
52        self.event_sink.persist_events(&events).await?;
53        Ok(events
54            .into_iter()
55            .map(|event| AnalyticsEventCreated {
56                id: event.id,
57                event_type: event.event_type,
58            })
59            .collect())
60    }
61
62    fn build_record(
63        session_id: &SessionId,
64        user_id: &UserId,
65        input: &CreateAnalyticsEventInput,
66    ) -> AnalyticsEventRecord {
67        AnalyticsEventRecord {
68            id: format!("evt_{}", uuid::Uuid::new_v4()),
69            user_id: user_id.clone(),
70            session_id: session_id.clone(),
71            event_type: input.event_type.as_str().to_owned(),
72            event_category: input.event_type.category().to_owned(),
73            page_url: input.page_url.clone(),
74            event_data: Self::build_event_data(input),
75        }
76    }
77
78    fn build_event_data(input: &CreateAnalyticsEventInput) -> serde_json::Value {
79        let mut data = input.data.clone().unwrap_or(serde_json::json!({}));
80
81        if let Some(obj) = data.as_object_mut() {
82            if let Some(content_id) = &input.content_id {
83                obj.insert(
84                    "content_id".to_owned(),
85                    serde_json::json!(content_id.as_str()),
86                );
87            }
88            if let Some(slug) = &input.slug {
89                obj.insert("slug".to_owned(), serde_json::json!(slug));
90            }
91            if let Some(referrer) = &input.referrer {
92                obj.insert("referrer".to_owned(), serde_json::json!(referrer));
93            }
94        }
95
96        data
97    }
98}
systemprompt_analytics/repository/events.rs

systemprompt_analytics/repository/
events.rs
