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

systemprompt_logging/repository/analytics/
mod.rs
