systemprompt_runtime/trace/repository/
log_search.rs

1//! Full-text log search queries for trace tooling.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use chrono::{DateTime, Utc};
7
8use systemprompt_identifiers::{LogId, McpServerId, McpToolName, TraceId};
9use systemprompt_logging::LogLevel;
10
11use super::{Result, TraceRepository};
12use crate::trace::models::{LogSearchItem, ToolExecutionItem};
13
14struct LogRow {
15    id: LogId,
16    trace_id: TraceId,
17    timestamp: DateTime<Utc>,
18    level: String,
19    module: String,
20    message: String,
21    metadata: Option<String>,
22}
23
24impl TraceRepository {
25    pub async fn search_logs(
26        &self,
27        pattern: &str,
28        since: Option<DateTime<Utc>>,
29        level: Option<LogLevel>,
30        limit: i64,
31    ) -> Result<Vec<LogSearchItem>> {
32        let rows = sqlx::query_as!(
33            LogRow,
34            r#"
35        SELECT
36            id as "id!: LogId",
37            trace_id as "trace_id!: TraceId",
38            timestamp as "timestamp!",
39            level as "level!",
40            module as "module!",
41            message as "message!",
42            metadata
43        FROM logs
44        WHERE message ILIKE $1
45          AND ($2::timestamptz IS NULL OR timestamp >= $2)
46          AND ($3::text IS NULL OR UPPER(level) = $3)
47        ORDER BY timestamp DESC
48        LIMIT $4
49        "#,
50            pattern,
51            since,
52            level.map(LogLevel::as_str),
53            limit
54        )
55        .fetch_all(&*self.pool)
56        .await?;
57
58        rows.into_iter()
59            .map(|r| -> Result<LogSearchItem> {
60                Ok(LogSearchItem {
61                    id: r.id,
62                    trace_id: r.trace_id,
63                    timestamp: r.timestamp,
64                    level: r.level.parse()?,
65                    module: r.module,
66                    message: r.message,
67                    metadata: r.metadata,
68                })
69            })
70            .collect()
71    }
72
73    pub async fn search_tool_executions(
74        &self,
75        pattern: &str,
76        since: Option<DateTime<Utc>>,
77        limit: i64,
78    ) -> Result<Vec<ToolExecutionItem>> {
79        let rows = sqlx::query!(
80            r#"
81        SELECT
82            started_at as "timestamp!",
83            trace_id as "trace_id!: TraceId",
84            tool_name as "tool_name!",
85            server_name,
86            status as "status!",
87            execution_time_ms
88        FROM mcp_tool_executions
89        WHERE (tool_name ILIKE $1 OR server_name ILIKE $1)
90          AND ($2::timestamptz IS NULL OR started_at >= $2)
91        ORDER BY started_at DESC
92        LIMIT $3
93        "#,
94            pattern,
95            since,
96            limit
97        )
98        .fetch_all(&*self.pool)
99        .await?;
100
101        Ok(rows
102            .into_iter()
103            .map(|r| ToolExecutionItem {
104                timestamp: r.timestamp,
105                trace_id: r.trace_id,
106                tool_name: McpToolName::new(r.tool_name),
107                server_name: Some(McpServerId::new(r.server_name)),
108                status: r.status,
109                execution_time_ms: r.execution_time_ms,
110            })
111            .collect())
112    }
113}
systemprompt_runtime/trace/repository/log_search.rs

systemprompt_runtime/trace/repository/
log_search.rs
