systemprompt_api/routes/analytics/
events.rs

1//! Analytics event ingestion routes.
2//!
3//! Records single and batched client analytics events, resolving the owning
4//! content id via slug/URL routing and fanning `page_exit` events out into the
5//! engagement store. Shared handler state lives in [`AnalyticsState`].
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use axum::Json;
11use axum::extract::{Extension, State};
12use axum::http::StatusCode;
13use axum::response::IntoResponse;
14use std::sync::Arc;
15
16use systemprompt_analytics::{
17    AnalyticsEventBatchResponse, AnalyticsEventType, AnalyticsEventsRepository,
18    CreateAnalyticsEventBatchInput, CreateAnalyticsEventInput, CreateEngagementEventInput,
19    EngagementOptionalMetrics, EngagementRepository,
20};
21use systemprompt_content::ContentRepository;
22use systemprompt_identifiers::{ContentId, LocaleCode};
23use systemprompt_models::ContentRouting;
24use systemprompt_models::api::ApiError;
25use systemprompt_models::execution::context::RequestContext;
26
27#[derive(Clone)]
28pub struct AnalyticsState {
29    pub events: Arc<AnalyticsEventsRepository>,
30    pub content: Arc<ContentRepository>,
31    pub engagement: Arc<EngagementRepository>,
32    pub content_routing: Option<Arc<dyn ContentRouting>>,
33}
34
35impl std::fmt::Debug for AnalyticsState {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        f.debug_struct("AnalyticsState")
38            .field("content_routing", &self.content_routing.is_some())
39            .finish_non_exhaustive()
40    }
41}
42
43async fn resolve_content_id(
44    content_repo: &ContentRepository,
45    content_routing: Option<&dyn ContentRouting>,
46    page_url: &str,
47    slug: Option<&str>,
48) -> Option<ContentId> {
49    let resolved_slug;
50    let slug_to_use = if let Some(s) = slug {
51        s
52    } else {
53        resolved_slug = content_routing.and_then(|r| r.resolve_slug(page_url))?;
54        &resolved_slug
55    };
56
57    content_repo
58        .get_by_slug(slug_to_use, &LocaleCode::new("en"))
59        .await
60        .map_err(|e| {
61            tracing::warn!(error = %e, slug = %slug_to_use, "Failed to lookup content by slug");
62            e
63        })
64        .ok()
65        .flatten()
66        .map(|c| c.id)
67}
68
69pub(super) async fn record_event(
70    State(state): State<AnalyticsState>,
71    Extension(req_ctx): Extension<RequestContext>,
72    Json(mut input): Json<CreateAnalyticsEventInput>,
73) -> Result<impl IntoResponse, ApiError> {
74    if input.content_id.is_none() {
75        input.content_id = resolve_content_id(
76            &state.content,
77            state.content_routing.as_deref(),
78            &input.page_url,
79            input.slug.as_deref(),
80        )
81        .await;
82    }
83
84    let created = state
85        .events
86        .create_event(req_ctx.session_id(), req_ctx.user_id(), &input)
87        .await
88        .map_err(|e| {
89            tracing::error!(error = %e, "Failed to record analytics event");
90            ApiError::internal_error("Failed to record analytics event")
91        })?;
92
93    if input.event_type == AnalyticsEventType::PageExit {
94        fan_out_engagement(&state, &req_ctx, &input).await;
95    }
96
97    Ok((StatusCode::CREATED, Json(created)))
98}
99
100pub(super) async fn record_events_batch(
101    State(state): State<AnalyticsState>,
102    Extension(req_ctx): Extension<RequestContext>,
103    Json(mut input): Json<CreateAnalyticsEventBatchInput>,
104) -> Result<impl IntoResponse, ApiError> {
105    for event in &mut input.events {
106        if event.content_id.is_none() {
107            event.content_id = resolve_content_id(
108                &state.content,
109                state.content_routing.as_deref(),
110                &event.page_url,
111                event.slug.as_deref(),
112            )
113            .await;
114        }
115    }
116
117    let created = state
118        .events
119        .create_events_batch(req_ctx.session_id(), req_ctx.user_id(), &input.events)
120        .await
121        .map_err(|e| {
122            tracing::error!(error = %e, "Failed to record analytics events batch");
123            ApiError::internal_error("Failed to record analytics events")
124        })?;
125
126    for event in &input.events {
127        if event.event_type == AnalyticsEventType::PageExit {
128            fan_out_engagement(&state, &req_ctx, event).await;
129        }
130    }
131
132    Ok((
133        StatusCode::CREATED,
134        Json(AnalyticsEventBatchResponse {
135            recorded: created.len(),
136            events: created,
137        }),
138    ))
139}
140
141async fn fan_out_engagement(
142    state: &AnalyticsState,
143    req_ctx: &RequestContext,
144    input: &CreateAnalyticsEventInput,
145) {
146    let Some(ref data) = input.data else { return };
147
148    let get_i32 = |key: &str| -> Option<i32> {
149        data.get(key)
150            .and_then(serde_json::Value::as_i64)
151            .map(|v| v as i32)
152    };
153    let get_f32 = |key: &str| -> Option<f32> {
154        data.get(key)
155            .and_then(serde_json::Value::as_f64)
156            .map(|v| v as f32)
157    };
158    let get_bool =
159        |key: &str| -> Option<bool> { data.get(key).and_then(serde_json::Value::as_bool) };
160    let get_string =
161        |key: &str| -> Option<String> { data.get(key).and_then(|v| v.as_str()).map(String::from) };
162
163    let time_on_page = get_i32("time_on_page_ms").unwrap_or(0);
164    if time_on_page == 0 {
165        return;
166    }
167
168    let engagement_input = CreateEngagementEventInput {
169        page_url: input.page_url.clone(),
170        event_type: input.event_type.as_str().to_owned(),
171        time_on_page_ms: time_on_page,
172        max_scroll_depth: get_i32("max_scroll_depth").unwrap_or(0),
173        click_count: get_i32("click_count").unwrap_or(0),
174        event_data: None,
175        optional_metrics: EngagementOptionalMetrics {
176            time_to_first_interaction_ms: get_i32("time_to_first_interaction_ms"),
177            time_to_first_scroll_ms: get_i32("time_to_first_scroll_ms"),
178            scroll_velocity_avg: get_f32("scroll_velocity_avg"),
179            scroll_direction_changes: get_i32("scroll_direction_changes"),
180            mouse_move_distance_px: get_i32("mouse_move_distance_px"),
181            keyboard_events: get_i32("keyboard_events"),
182            copy_events: get_i32("copy_events"),
183            focus_time_ms: get_i32("focus_time_ms"),
184            blur_count: get_i32("blur_count"),
185            tab_switches: get_i32("tab_switches"),
186            visible_time_ms: get_i32("visible_time_ms"),
187            hidden_time_ms: get_i32("hidden_time_ms"),
188            is_rage_click: get_bool("is_rage_click"),
189            is_dead_click: get_bool("is_dead_click"),
190            reading_pattern: get_string("reading_pattern"),
191        },
192    };
193
194    let content_id = resolve_content_id(
195        &state.content,
196        state.content_routing.as_deref(),
197        &input.page_url,
198        input.slug.as_deref(),
199    )
200    .await;
201
202    if let Err(e) = state
203        .engagement
204        .create_engagement(
205            req_ctx.session_id(),
206            req_ctx.user_id(),
207            content_id.as_ref(),
208            &engagement_input,
209        )
210        .await
211    {
212        tracing::warn!(error = %e, "Failed to fan out engagement data from page_exit event");
213    }
214}
systemprompt_api/routes/analytics/events.rs

systemprompt_api/routes/analytics/
events.rs
