systemprompt_logging/repository/
mod.rs

1//! Log persistence repository.
2//!
3//! [`LoggingRepository`] writes entries to the configured sinks (terminal
4//! and/or database) and serves paginated reads, lookups, and age-based cleanup;
5//! [`AnalyticsRepository`] records analytics events. Read and write pools are
6//! held separately so reads never contend with the write path.
7//!
8//! Copyright (c) systemprompt.io — Business Source License 1.1.
9//! See <https://systemprompt.io> for licensing details.
10
11use std::io::Write;
12use std::sync::Arc;
13
14use chrono::{DateTime, Utc};
15use sqlx::PgPool;
16use systemprompt_database::DbPool;
17use systemprompt_identifiers::LogId;
18
19use crate::models::{LogEntry, LogFilter, LoggingError};
20
21pub mod analytics;
22mod operations;
23
24pub use analytics::{AnalyticsEvent, AnalyticsRepository};
25
26#[derive(Clone, Debug)]
27pub struct LoggingRepository {
28    pool: Arc<PgPool>,
29    write_pool: Arc<PgPool>,
30    terminal_output: bool,
31    db_output: bool,
32}
33
34impl LoggingRepository {
35    pub fn new(db: &DbPool) -> Result<Self, LoggingError> {
36        let pool = db.pool_arc()?;
37        let write_pool = db.write_pool_arc()?;
38        Ok(Self {
39            pool,
40            write_pool,
41            terminal_output: true,
42            db_output: false,
43        })
44    }
45
46    #[must_use]
47    pub const fn with_terminal(mut self, enabled: bool) -> Self {
48        self.terminal_output = enabled;
49        self
50    }
51
52    #[must_use]
53    pub const fn with_database(mut self, enabled: bool) -> Self {
54        self.db_output = enabled;
55        self
56    }
57
58    pub async fn log(&self, entry: LogEntry) -> Result<(), LoggingError> {
59        entry.validate()?;
60
61        if self.terminal_output {
62            let mut stdout = std::io::stdout();
63            writeln!(stdout, "{entry}").ok();
64        }
65
66        if self.db_output {
67            operations::create_log(&self.write_pool, &entry).await?;
68        }
69
70        Ok(())
71    }
72
73    pub async fn get_recent_logs(&self, limit: i64) -> Result<Vec<LogEntry>, LoggingError> {
74        operations::list_logs(&self.pool, limit).await
75    }
76
77    pub async fn get_logs_by_module_patterns(
78        &self,
79        patterns: &[String],
80        limit: i64,
81    ) -> Result<Vec<LogEntry>, LoggingError> {
82        operations::list_logs_by_module_patterns(&self.pool, patterns, limit).await
83    }
84
85    pub async fn cleanup_old_logs(&self, older_than: DateTime<Utc>) -> Result<u64, LoggingError> {
86        operations::cleanup_logs_before(&self.write_pool, older_than).await
87    }
88
89    pub async fn count_logs_before(&self, cutoff: DateTime<Utc>) -> Result<u64, LoggingError> {
90        operations::count_logs_before(&self.pool, cutoff).await
91    }
92
93    pub async fn clear_all_logs(&self) -> Result<u64, LoggingError> {
94        operations::clear_all_logs(&self.write_pool).await
95    }
96
97    pub async fn get_logs_paginated(
98        &self,
99        filter: &LogFilter,
100    ) -> Result<(Vec<LogEntry>, i64), LoggingError> {
101        operations::list_logs_paginated(&self.pool, filter).await
102    }
103
104    pub async fn get_by_id(&self, id: &LogId) -> Result<Option<LogEntry>, LoggingError> {
105        operations::get_log(&self.pool, id).await
106    }
107
108    pub async fn update_log_entry(
109        &self,
110        id: &LogId,
111        entry: &LogEntry,
112    ) -> Result<bool, LoggingError> {
113        operations::update_log(&self.write_pool, id, entry).await
114    }
115
116    pub async fn delete_log_entry(&self, id: &LogId) -> Result<bool, LoggingError> {
117        operations::delete_log(&self.write_pool, id).await
118    }
119
120    pub async fn delete_log_entries(&self, ids: &[LogId]) -> Result<u64, LoggingError> {
121        operations::delete_logs_multiple(&self.write_pool, ids).await
122    }
123}
systemprompt_logging/repository/mod.rs

systemprompt_logging/repository/
mod.rs
