teaql_runtime/context/
logging.rs

1use std::sync::{Arc, Mutex};
2use std::time::{Duration, SystemTime};
3
4use super::UserContext;
5use teaql_core::Value;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum SqlLogOperation {
9    Select,
10    Insert,
11    Update,
12    Delete,
13    Recover,
14}
15
16impl SqlLogOperation {
17    pub fn is_select(self) -> bool {
18        matches!(self, Self::Select)
19    }
20
21    pub fn is_mutation(self) -> bool {
22        !self.is_select()
23    }
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub struct SqlLogOptions {
28    pub select: bool,
29    pub mutation: bool,
30}
31
32impl Default for SqlLogOptions {
33    fn default() -> Self {
34        Self::all()
35    }
36}
37
38impl SqlLogOptions {
39    pub fn disabled() -> Self {
40        Self {
41            select: false,
42            mutation: false,
43        }
44    }
45
46    pub fn select_only() -> Self {
47        Self {
48            select: true,
49            mutation: false,
50        }
51    }
52
53    pub fn mutation_only() -> Self {
54        Self {
55            select: false,
56            mutation: true,
57        }
58    }
59
60    pub fn all() -> Self {
61        Self {
62            select: true,
63            mutation: true,
64        }
65    }
66
67    pub fn enabled_for(self, operation: SqlLogOperation) -> bool {
68        if operation.is_select() {
69            self.select
70        } else {
71            self.mutation
72        }
73    }
74}
75
76#[derive(Debug, Clone, PartialEq)]
77pub struct SqlLogEntry {
78    pub operation: SqlLogOperation,
79    pub comment: Option<String>,
80    pub purpose: Option<String>,
81    pub audit_reason: Option<String>,
82    pub trace_path: Vec<teaql_core::TraceNode>,
83    pub sql: String,
84    pub params: Vec<Value>,
85    pub debug_sql: String,
86    pub pretty_sql: String,
87    pub started_at: SystemTime,
88    pub ended_at: SystemTime,
89    pub elapsed: Duration,
90    pub result_count: Option<usize>,
91    pub result_type: Option<String>,
92    pub affected_rows: Option<u64>,
93    pub result_summary: String,
94}
95
96#[derive(Debug, Clone, PartialEq)]
97pub struct UnifiedLogEntry {
98    pub timestamp: SystemTime,
99    pub user_identifier: Option<String>,
100    pub trace_chain: Vec<teaql_core::TraceNode>,
101    pub payload: LogPayload,
102}
103
104#[derive(Debug, Clone, PartialEq)]
105#[allow(clippy::large_enum_variant)] // Boxing Sql would break the public constructor shape.
106pub enum LogPayload {
107    Sql(SqlLogEntry),
108    Info(InfoLogEntry),
109}
110
111#[derive(Debug, Clone, PartialEq)]
112pub struct InfoLogEntry {
113    pub message: String,
114}
115
116#[derive(Clone, Default)]
117pub struct UnifiedLogBuffer {
118    pub entries: Arc<Mutex<Vec<UnifiedLogEntry>>>,
119}
120
121impl UserContext {
122    pub fn with_sql_log_options(mut self, options: SqlLogOptions) -> Self {
123        self.sql_log_options = options;
124        self
125    }
126
127    pub fn set_sql_log_options(&mut self, options: SqlLogOptions) {
128        self.sql_log_options = options;
129    }
130
131    pub fn enable_select_sql_log(&mut self) {
132        self.sql_log_options.select = true;
133    }
134
135    pub fn enable_mutation_sql_log(&mut self) {
136        self.sql_log_options.mutation = true;
137    }
138
139    pub fn disable_select_sql_log(&mut self) {
140        self.sql_log_options.select = false;
141    }
142
143    pub fn disable_mutation_sql_log(&mut self) {
144        self.sql_log_options.mutation = false;
145    }
146
147    pub fn enable_all_sql_log(&mut self) {
148        self.sql_log_options = SqlLogOptions::all();
149    }
150
151    pub fn disable_sql_log(&mut self) {
152        self.sql_log_options = SqlLogOptions::disabled();
153        self.clear_sql_logs();
154    }
155
156    pub fn sql_log_options(&self) -> SqlLogOptions {
157        self.sql_log_options
158    }
159
160    pub fn sql_logs(&self) -> Vec<SqlLogEntry> {
161        self.sql_log_entries
162            .lock()
163            .map(|entries| entries.clone())
164            .unwrap_or_default()
165    }
166
167    pub fn clear_sql_logs(&self) {
168        if let Ok(mut entries) = self.sql_log_entries.lock() {
169            entries.clear();
170        }
171    }
172
173    pub(crate) fn record_metadata_log(&self, metadata: &teaql_data_service::ExecutionMetadata) {
174        let operation = match metadata.operation {
175            teaql_data_service::DataServiceOperation::Query => SqlLogOperation::Select,
176            teaql_data_service::DataServiceOperation::Insert => SqlLogOperation::Insert,
177            teaql_data_service::DataServiceOperation::Update => SqlLogOperation::Update,
178            teaql_data_service::DataServiceOperation::Delete => SqlLogOperation::Delete,
179            teaql_data_service::DataServiceOperation::Recover => SqlLogOperation::Update,
180            teaql_data_service::DataServiceOperation::Batch => SqlLogOperation::Update,
181            teaql_data_service::DataServiceOperation::Schema => SqlLogOperation::Update,
182        };
183        if !self.sql_log_options.enabled_for(operation) {
184            return;
185        }
186        let trace_path =
187            canonical_sql_trace_path(operation, &metadata.backend, &metadata.trace_chain);
188        let result_summary = metadata
189            .result_count
190            .map(|count| format!("{count} rows returned"))
191            .or_else(|| {
192                metadata
193                    .affected_rows
194                    .map(|affected| format!("{affected} rows affected"))
195            })
196            .unwrap_or_default();
197        let debug_sql = metadata.debug_query.as_deref().unwrap_or_default();
198        let sensitive_entry = SqlLogEntry {
199            operation,
200            comment: trace_value(&metadata.trace_chain, teaql_core::TraceKind::Comment)
201                .or_else(|| metadata.comment.clone()),
202            purpose: trace_value(&metadata.trace_chain, teaql_core::TraceKind::Purpose),
203            audit_reason: trace_value(&metadata.trace_chain, teaql_core::TraceKind::AuditReason),
204            trace_path: trace_path.clone(),
205            sql: metadata.parameterized_query.clone().unwrap_or_default(),
206            params: metadata.params.clone(),
207            pretty_sql: pretty_sql(debug_sql),
208            debug_sql: debug_sql.to_owned(),
209            started_at: metadata.started_at,
210            ended_at: metadata.ended_at,
211            elapsed: metadata
212                .ended_at
213                .duration_since(metadata.started_at)
214                .unwrap_or_default(),
215            result_count: metadata.result_count,
216            result_type: None,
217            affected_rows: metadata.affected_rows,
218            result_summary,
219        };
220        // The ordinary context buffer and default operator log are safe
221        // telemetry. Values and copy-paste SQL are only sent to an explicitly
222        // configured diagnostic sink, never retained in this buffer.
223        let mut safe_entry = sensitive_entry.clone();
224        safe_entry.params.clear();
225        safe_entry.debug_sql.clear();
226        safe_entry.pretty_sql.clear();
227        self.append_sql_log(metadata.started_at, trace_path, safe_entry, sensitive_entry);
228    }
229
230    fn append_sql_log(
231        &self,
232        timestamp: SystemTime,
233        trace_path: Vec<teaql_core::TraceNode>,
234        safe_entry: SqlLogEntry,
235        sensitive_entry: SqlLogEntry,
236    ) {
237        if let Ok(mut entries) = self.sql_log_entries.lock() {
238            entries.push(safe_entry.clone());
239        }
240        if let Some(buffer) = self.get_resource::<UnifiedLogBuffer>()
241            && let Ok(mut entries) = buffer.entries.lock()
242        {
243            entries.push(UnifiedLogEntry {
244                timestamp,
245                user_identifier: self.user_identifier.clone(),
246                trace_chain: trace_path.clone(),
247                payload: LogPayload::Sql(safe_entry.clone()),
248            });
249        }
250        crate::log_formatter::LogManager::write_sql_log(&trace_path, &safe_entry);
251        crate::log_formatter::LogManager::write_sensitive_sql_log(&trace_path, &sensitive_entry);
252    }
253}
254
255fn trace_value(
256    trace_path: &[teaql_core::TraceNode],
257    kind: teaql_core::TraceKind,
258) -> Option<String> {
259    trace_path
260        .iter()
261        .rev()
262        .find(|node| node.kind == kind)
263        .map(|node| node.comment.clone())
264}
265
266fn canonical_sql_trace_path(
267    operation: SqlLogOperation,
268    backend: &str,
269    source: &[teaql_core::TraceNode],
270) -> Vec<teaql_core::TraceNode> {
271    use teaql_core::{TraceKind, TraceNode};
272
273    if source.iter().any(|node| node.kind == TraceKind::Operation)
274        && source.iter().any(|node| node.kind == TraceKind::Provider)
275        && source.iter().any(|node| node.kind == TraceKind::Sql)
276    {
277        return source
278            .iter()
279            .filter(|node| {
280                !matches!(
281                    node.kind,
282                    TraceKind::Comment | TraceKind::Purpose | TraceKind::AuditReason
283                )
284            })
285            .cloned()
286            .collect();
287    }
288    let operation_entity = source
289        .iter()
290        .find(|node| !node.entity_type.trim().is_empty())
291        .map(|node| node.entity_type.clone())
292        .unwrap_or_else(|| "unknown".to_owned());
293    let statement_entity = if operation.is_select() {
294        operation_entity.clone()
295    } else {
296        source
297            .iter()
298            .rev()
299            .find(|node| node.kind == TraceKind::Entity && !node.entity_type.trim().is_empty())
300            .map(|node| node.entity_type.clone())
301            .unwrap_or_else(|| operation_entity.clone())
302    };
303    let family = if operation.is_select() {
304        "query"
305    } else {
306        "mutation"
307    };
308    let statement = match operation {
309        SqlLogOperation::Select => "select",
310        SqlLogOperation::Insert => "insert",
311        SqlLogOperation::Update => "update",
312        SqlLogOperation::Delete => "delete",
313        SqlLogOperation::Recover => "recover",
314    };
315    let mut path = vec![TraceNode::typed(
316        TraceKind::Operation,
317        operation_entity,
318        None,
319        family,
320    )];
321    path.push(TraceNode::typed(
322        if operation.is_select() {
323            TraceKind::Request
324        } else {
325            TraceKind::Entity
326        },
327        statement_entity,
328        None,
329        "",
330    ));
331    path.extend(
332        source
333            .iter()
334            .filter(|node| node.kind == TraceKind::Relation)
335            .cloned(),
336    );
337    path.push(TraceNode::typed(
338        TraceKind::Provider,
339        if backend.trim().is_empty() {
340            "unknown"
341        } else {
342            backend
343        },
344        None,
345        "",
346    ));
347    path.push(TraceNode::typed(TraceKind::Sql, statement, None, ""));
348    path
349}
350
351fn pretty_sql(sql: &str) -> String {
352    let mut pretty = sql.to_owned();
353    for keyword in [
354        " FROM ",
355        " WHERE ",
356        " GROUP BY ",
357        " HAVING ",
358        " ORDER BY ",
359        " LIMIT ",
360        " OFFSET ",
361        " RETURNING ",
362    ] {
363        pretty = pretty.replace(keyword, &format!("\n{}", keyword.trim_start()));
364    }
365    pretty.replace(" AND ", "\n  AND ")
366}
teaql_runtime/context/logging.rs

teaql_runtime/context/
logging.rs
