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::{DateTime, 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) -> Self {
29        let write_pool = db.write_pool();
30        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    pub async fn purge_before(
39        &self,
40        cutoff: DateTime<Utc>,
41        batch: i64,
42    ) -> Result<u64, LoggingError> {
43        let result = sqlx::query!(
44            "DELETE FROM analytics_events WHERE ctid = ANY(ARRAY(SELECT ctid FROM analytics_events WHERE timestamp < $1 LIMIT $2))",
45            cutoff,
46            batch
47        )
48        .execute(&*self.write_pool)
49        .await?;
50        Ok(result.rows_affected())
51    }
52}
53
54async fn execute_insert(pool: &PgPool, event: &AnalyticsEvent) -> Result<u64, LoggingError> {
55    let params = EventParams::from(event);
56    run_insert_query(pool, params).await
57}
58
59async fn run_insert_query(pool: &PgPool, p: EventParams<'_>) -> Result<u64, LoggingError> {
60    let result = sqlx::query!(
61        r"
62        INSERT INTO analytics_events
63        (user_id, session_id, context_id, event_type, event_category, severity,
64         endpoint, error_code, response_time_ms, agent_id, task_id, message, metadata, timestamp)
65        VALUES
66        ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
67        ",
68        p.user_id.as_str(),
69        p.session_id.as_str(),
70        p.context_id.as_str(),
71        p.event_type,
72        p.event_category,
73        p.severity,
74        p.endpoint,
75        p.error_code,
76        p.response_time_ms,
77        p.agent_id.map(AgentId::as_str),
78        p.task_id.map(TaskId::as_str),
79        p.message,
80        p.metadata,
81        p.timestamp
82    )
83    .execute(pool)
84    .await?;
85    Ok(result.rows_affected())
86}
87
88struct EventParams<'a> {
89    user_id: &'a UserId,
90    session_id: &'a SessionId,
91    context_id: &'a ContextId,
92    event_type: &'a str,
93    event_category: &'a str,
94    severity: &'a str,
95    agent_id: Option<&'a AgentId>,
96    task_id: Option<&'a TaskId>,
97    endpoint: Option<&'a str>,
98    message: Option<&'a str>,
99    error_code: Option<i32>,
100    response_time_ms: Option<i32>,
101    metadata: String,
102    timestamp: DateTime<Utc>,
103}
104
105impl<'a> From<&'a AnalyticsEvent> for EventParams<'a> {
106    fn from(event: &'a AnalyticsEvent) -> Self {
107        Self {
108            user_id: &event.user_id,
109            session_id: &event.session_id,
110            context_id: &event.context_id,
111            event_type: &event.event_type,
112            event_category: &event.event_category,
113            severity: &event.severity,
114            agent_id: event.agent_id.as_ref(),
115            task_id: event.task_id.as_ref(),
116            endpoint: event.endpoint.as_deref(),
117            message: event.message.as_deref(),
118            error_code: event.error_code,
119            response_time_ms: event.response_time_ms,
120            metadata: event.metadata.to_string(),
121            timestamp: Utc::now(),
122        }
123    }
124}
125
126#[derive(Debug, Clone)]
127pub struct AnalyticsEvent {
128    pub user_id: UserId,
129    pub session_id: SessionId,
130    pub context_id: ContextId,
131    pub event_type: String,
132    pub event_category: String,
133    pub severity: String,
134    pub endpoint: Option<String>,
135    pub error_code: Option<i32>,
136    pub response_time_ms: Option<i32>,
137    pub agent_id: Option<AgentId>,
138    pub task_id: Option<TaskId>,
139    pub message: Option<String>,
140    // JSON: analytics event `metadata` — open per-event detail, stored as JSONB.
141    pub metadata: Value,
142}
systemprompt_logging/repository/analytics/mod.rs

systemprompt_logging/repository/analytics/
mod.rs
