systemprompt_logging/repository/analytics/
mod.rs

1//! Analytics-event persistence for the logging subsystem.
2//!
3//! [`AnalyticsRepository`] writes structured [`AnalyticsEvent`] rows to the
4//! `analytics_events` table on the database write pool, carrying typed actor,
5//! session, and task identifiers alongside the request-level metrics
6//! (endpoint, error code, response time) the platform aggregates downstream.
7
8use chrono::Utc;
9use serde_json::Value;
10use sqlx::PgPool;
11use std::sync::Arc;
12use systemprompt_database::DbPool;
13use systemprompt_identifiers::{AgentId, ContextId, SessionId, TaskId, UserId};
14
15use crate::models::LoggingError;
16
17#[derive(Debug, Clone)]
18pub struct AnalyticsRepository {
19    write_pool: Arc<PgPool>,
20}
21
22impl AnalyticsRepository {
23    pub fn new(db: &DbPool) -> Result<Self, LoggingError> {
24        let write_pool = db.write_pool_arc()?;
25        Ok(Self { write_pool })
26    }
27
28    pub async fn log_event(&self, event: &AnalyticsEvent) -> Result<i64, LoggingError> {
29        let result = execute_insert(&self.write_pool, event).await?;
30        Ok(i64::try_from(result).unwrap_or(i64::MAX))
31    }
32}
33
34async fn execute_insert(pool: &PgPool, event: &AnalyticsEvent) -> Result<u64, LoggingError> {
35    let params = EventParams::from(event);
36    run_insert_query(pool, params).await
37}
38
39async fn run_insert_query(pool: &PgPool, p: EventParams<'_>) -> Result<u64, LoggingError> {
40    let result = sqlx::query!(
41        r"
42        INSERT INTO analytics_events
43        (user_id, session_id, context_id, event_type, event_category, severity,
44         endpoint, error_code, response_time_ms, agent_id, task_id, message, metadata, timestamp)
45        VALUES
46        ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
47        ",
48        p.user_id,
49        p.session_id,
50        p.context_id,
51        p.event_type,
52        p.event_category,
53        p.severity,
54        p.endpoint,
55        p.error_code,
56        p.response_time_ms,
57        p.agent_id,
58        p.task_id,
59        p.message,
60        p.metadata,
61        p.timestamp
62    )
63    .execute(pool)
64    .await?;
65    Ok(result.rows_affected())
66}
67
68struct EventParams<'a> {
69    user_id: &'a str,
70    session_id: &'a str,
71    context_id: &'a str,
72    event_type: &'a str,
73    event_category: &'a str,
74    severity: &'a str,
75    agent_id: Option<&'a str>,
76    task_id: Option<&'a str>,
77    endpoint: Option<&'a str>,
78    message: Option<&'a str>,
79    error_code: Option<i32>,
80    response_time_ms: Option<i32>,
81    metadata: String,
82    timestamp: chrono::DateTime<Utc>,
83}
84
85impl<'a> From<&'a AnalyticsEvent> for EventParams<'a> {
86    fn from(event: &'a AnalyticsEvent) -> Self {
87        Self {
88            user_id: event.user_id.as_str(),
89            session_id: event.session_id.as_str(),
90            context_id: event.context_id.as_str(),
91            event_type: &event.event_type,
92            event_category: &event.event_category,
93            severity: &event.severity,
94            agent_id: event.agent_id.as_ref().map(AgentId::as_str),
95            task_id: event.task_id.as_ref().map(TaskId::as_str),
96            endpoint: event.endpoint.as_deref(),
97            message: event.message.as_deref(),
98            error_code: event.error_code,
99            response_time_ms: event.response_time_ms,
100            metadata: event.metadata.to_string(),
101            timestamp: Utc::now(),
102        }
103    }
104}
105
106#[derive(Debug, Clone)]
107pub struct AnalyticsEvent {
108    pub user_id: UserId,
109    pub session_id: SessionId,
110    pub context_id: ContextId,
111    pub event_type: String,
112    pub event_category: String,
113    pub severity: String,
114    pub endpoint: Option<String>,
115    pub error_code: Option<i32>,
116    pub response_time_ms: Option<i32>,
117    pub agent_id: Option<AgentId>,
118    pub task_id: Option<TaskId>,
119    pub message: Option<String>,
120    pub metadata: Value,
121}
systemprompt_logging/repository/analytics/mod.rs

systemprompt_logging/repository/analytics/
mod.rs
