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
96impl SqlLogEntry {
97    /// Number of bound parameters without exposing their values.
98    ///
99    /// Safe/default log entries retain one `Value::Null` slot per parameter;
100    /// sensitive diagnostic entries retain the original values. The count is
101    /// therefore stable across both views of the same execution.
102    pub fn parameter_count(&self) -> usize {
103        self.params.len()
104    }
105}
106
107#[derive(Debug, Clone, PartialEq)]
108pub struct UnifiedLogEntry {
109    pub timestamp: SystemTime,
110    pub user_identifier: Option<String>,
111    pub trace_chain: Vec<teaql_core::TraceNode>,
112    pub payload: LogPayload,
113}
114
115#[derive(Debug, Clone, PartialEq)]
116#[allow(clippy::large_enum_variant)] // Boxing Sql would break the public constructor shape.
117pub enum LogPayload {
118    Sql(SqlLogEntry),
119    Info(InfoLogEntry),
120}
121
122#[derive(Debug, Clone, PartialEq)]
123pub struct InfoLogEntry {
124    pub message: String,
125}
126
127#[derive(Clone, Default)]
128pub struct UnifiedLogBuffer {
129    pub entries: Arc<Mutex<Vec<UnifiedLogEntry>>>,
130}
131
132impl UserContext {
133    pub fn with_sql_log_options(mut self, options: SqlLogOptions) -> Self {
134        self.sql_log_options = options;
135        self
136    }
137
138    pub fn set_sql_log_options(&mut self, options: SqlLogOptions) {
139        self.sql_log_options = options;
140    }
141
142    pub fn enable_select_sql_log(&mut self) {
143        self.sql_log_options.select = true;
144    }
145
146    pub fn enable_mutation_sql_log(&mut self) {
147        self.sql_log_options.mutation = true;
148    }
149
150    pub fn disable_select_sql_log(&mut self) {
151        self.sql_log_options.select = false;
152    }
153
154    pub fn disable_mutation_sql_log(&mut self) {
155        self.sql_log_options.mutation = false;
156    }
157
158    pub fn enable_all_sql_log(&mut self) {
159        self.sql_log_options = SqlLogOptions::all();
160    }
161
162    pub fn disable_sql_log(&mut self) {
163        self.sql_log_options = SqlLogOptions::disabled();
164        self.clear_sql_logs();
165    }
166
167    pub fn sql_log_options(&self) -> SqlLogOptions {
168        self.sql_log_options
169    }
170
171    pub fn sql_logs(&self) -> Vec<SqlLogEntry> {
172        self.sql_log_entries
173            .lock()
174            .map(|entries| entries.clone())
175            .unwrap_or_default()
176    }
177
178    pub fn clear_sql_logs(&self) {
179        if let Ok(mut entries) = self.sql_log_entries.lock() {
180            entries.clear();
181        }
182    }
183
184    pub(crate) fn record_metadata_log(&self, metadata: &teaql_data_service::ExecutionMetadata) {
185        let operation = match metadata.operation {
186            teaql_data_service::DataServiceOperation::Query => SqlLogOperation::Select,
187            teaql_data_service::DataServiceOperation::Insert => SqlLogOperation::Insert,
188            teaql_data_service::DataServiceOperation::Update => SqlLogOperation::Update,
189            teaql_data_service::DataServiceOperation::Delete => SqlLogOperation::Delete,
190            teaql_data_service::DataServiceOperation::Recover => SqlLogOperation::Update,
191            teaql_data_service::DataServiceOperation::Batch => SqlLogOperation::Update,
192            teaql_data_service::DataServiceOperation::Schema => SqlLogOperation::Update,
193        };
194        if !self.sql_log_options.enabled_for(operation) {
195            return;
196        }
197        let trace_path =
198            canonical_sql_trace_path(operation, &metadata.backend, &metadata.trace_chain);
199        let result_summary = metadata
200            .result_count
201            .map(|count| format!("{count} rows returned"))
202            .or_else(|| {
203                metadata
204                    .affected_rows
205                    .map(|affected| format!("{affected} rows affected"))
206            })
207            .unwrap_or_default();
208        let debug_sql = metadata.debug_query.as_deref().unwrap_or_default();
209        let sensitive_entry = SqlLogEntry {
210            operation,
211            comment: trace_value(&metadata.trace_chain, teaql_core::TraceKind::Comment)
212                .or_else(|| metadata.comment.clone()),
213            purpose: trace_value(&metadata.trace_chain, teaql_core::TraceKind::Purpose),
214            audit_reason: trace_value(&metadata.trace_chain, teaql_core::TraceKind::AuditReason),
215            trace_path: trace_path.clone(),
216            sql: metadata.parameterized_query.clone().unwrap_or_default(),
217            params: metadata.params.clone(),
218            pretty_sql: pretty_sql(debug_sql),
219            debug_sql: debug_sql.to_owned(),
220            started_at: metadata.started_at,
221            ended_at: metadata.ended_at,
222            elapsed: metadata
223                .ended_at
224                .duration_since(metadata.started_at)
225                .unwrap_or_default(),
226            result_count: metadata.result_count,
227            result_type: None,
228            affected_rows: metadata.affected_rows,
229            result_summary,
230        };
231        // The ordinary context buffer and default operator log are safe
232        // telemetry. Values and copy-paste SQL are only sent to an explicitly
233        // configured diagnostic sink, never retained in this buffer.
234        let mut safe_entry = sensitive_entry.clone();
235        // Preserve only the non-sensitive shape. Clearing the vector used to
236        // make a parameterized query indistinguishable from a literal-only
237        // query, while retaining the values would leak customer data.
238        safe_entry
239            .params
240            .iter_mut()
241            .for_each(|value| *value = Value::Null);
242        safe_entry.debug_sql.clear();
243        safe_entry.pretty_sql.clear();
244        self.append_sql_log(metadata.started_at, trace_path, safe_entry, sensitive_entry);
245    }
246
247    fn append_sql_log(
248        &self,
249        timestamp: SystemTime,
250        trace_path: Vec<teaql_core::TraceNode>,
251        safe_entry: SqlLogEntry,
252        sensitive_entry: SqlLogEntry,
253    ) {
254        if let Ok(mut entries) = self.sql_log_entries.lock() {
255            entries.push(safe_entry.clone());
256        }
257        if let Some(buffer) = self.get_resource::<UnifiedLogBuffer>()
258            && let Ok(mut entries) = buffer.entries.lock()
259        {
260            entries.push(UnifiedLogEntry {
261                timestamp,
262                user_identifier: self.user_identifier.clone(),
263                trace_chain: trace_path.clone(),
264                payload: LogPayload::Sql(safe_entry.clone()),
265            });
266        }
267        crate::log_formatter::LogManager::write_sql_log(&trace_path, &safe_entry);
268        crate::log_formatter::LogManager::write_sensitive_sql_log(&trace_path, &sensitive_entry);
269    }
270}
271
272fn trace_value(
273    trace_path: &[teaql_core::TraceNode],
274    kind: teaql_core::TraceKind,
275) -> Option<String> {
276    trace_path
277        .iter()
278        .rev()
279        .find(|node| node.kind == kind)
280        .map(|node| node.comment.clone())
281}
282
283fn canonical_sql_trace_path(
284    operation: SqlLogOperation,
285    backend: &str,
286    source: &[teaql_core::TraceNode],
287) -> Vec<teaql_core::TraceNode> {
288    use teaql_core::{TraceKind, TraceNode};
289
290    if source.iter().any(|node| node.kind == TraceKind::Operation)
291        && source.iter().any(|node| node.kind == TraceKind::Provider)
292        && source.iter().any(|node| node.kind == TraceKind::Sql)
293    {
294        return source
295            .iter()
296            .filter(|node| {
297                !matches!(
298                    node.kind,
299                    TraceKind::Comment | TraceKind::Purpose | TraceKind::AuditReason
300                )
301            })
302            .cloned()
303            .collect();
304    }
305    let operation_entity = source
306        .iter()
307        .find(|node| !node.entity_type.trim().is_empty())
308        .map(|node| node.entity_type.clone())
309        .unwrap_or_else(|| "unknown".to_owned());
310    let statement_entity = if operation.is_select() {
311        operation_entity.clone()
312    } else {
313        source
314            .iter()
315            .rev()
316            .find(|node| node.kind == TraceKind::Entity && !node.entity_type.trim().is_empty())
317            .map(|node| node.entity_type.clone())
318            .unwrap_or_else(|| operation_entity.clone())
319    };
320    let family = if operation.is_select() {
321        "query"
322    } else {
323        "mutation"
324    };
325    let statement = match operation {
326        SqlLogOperation::Select => "select",
327        SqlLogOperation::Insert => "insert",
328        SqlLogOperation::Update => "update",
329        SqlLogOperation::Delete => "delete",
330        SqlLogOperation::Recover => "recover",
331    };
332    let mut path = vec![TraceNode::typed(
333        TraceKind::Operation,
334        operation_entity,
335        None,
336        family,
337    )];
338    path.push(TraceNode::typed(
339        if operation.is_select() {
340            TraceKind::Request
341        } else {
342            TraceKind::Entity
343        },
344        statement_entity,
345        None,
346        "",
347    ));
348    path.extend(
349        source
350            .iter()
351            .filter(|node| node.kind == TraceKind::Relation)
352            .cloned(),
353    );
354    path.push(TraceNode::typed(
355        TraceKind::Provider,
356        if backend.trim().is_empty() {
357            "unknown"
358        } else {
359            backend
360        },
361        None,
362        "",
363    ));
364    path.push(TraceNode::typed(TraceKind::Sql, statement, None, ""));
365    path
366}
367
368fn pretty_sql(sql: &str) -> String {
369    let mut pretty = sql.to_owned();
370    for keyword in [
371        " FROM ",
372        " WHERE ",
373        " GROUP BY ",
374        " HAVING ",
375        " ORDER BY ",
376        " LIMIT ",
377        " OFFSET ",
378        " RETURNING ",
379    ] {
380        pretty = pretty.replace(keyword, &format!("\n{}", keyword.trim_start()));
381    }
382    pretty.replace(" AND ", "\n  AND ")
383}
teaql_runtime/context/logging.rs

teaql_runtime/context/
logging.rs
