teaql_runtime/context/
logging.rs

1use std::sync::{Arc, Mutex};
2use std::time::{Duration, SystemTime};
3
4use teaql_core::Value;
5use teaql_sql::{CompiledQuery, DatabaseKind};
6
7use super::UserContext;
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub enum SqlLogOperation {
11    Select,
12    Insert,
13    Update,
14    Delete,
15    Recover,
16}
17
18impl SqlLogOperation {
19    pub fn is_select(self) -> bool {
20        matches!(self, Self::Select)
21    }
22
23    pub fn is_mutation(self) -> bool {
24        !self.is_select()
25    }
26}
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub struct SqlLogOptions {
30    pub select: bool,
31    pub mutation: bool,
32}
33
34impl Default for SqlLogOptions {
35    fn default() -> Self {
36        Self::all()
37    }
38}
39
40impl SqlLogOptions {
41    pub fn disabled() -> Self {
42        Self {
43            select: false,
44            mutation: false,
45        }
46    }
47
48    pub fn select_only() -> Self {
49        Self {
50            select: true,
51            mutation: false,
52        }
53    }
54
55    pub fn mutation_only() -> Self {
56        Self {
57            select: false,
58            mutation: true,
59        }
60    }
61
62    pub fn all() -> Self {
63        Self {
64            select: true,
65            mutation: true,
66        }
67    }
68
69    pub fn enabled_for(self, operation: SqlLogOperation) -> bool {
70        if operation.is_select() {
71            self.select
72        } else {
73            self.mutation
74        }
75    }
76}
77
78#[derive(Debug, Clone, PartialEq)]
79pub struct SqlLogEntry {
80    pub operation: SqlLogOperation,
81    pub comment: Option<String>,
82    pub purpose: Option<String>,
83    pub audit_reason: Option<String>,
84    pub trace_path: Vec<teaql_core::TraceNode>,
85    pub sql: String,
86    pub params: Vec<Value>,
87    pub debug_sql: String,
88    pub pretty_sql: String,
89    pub started_at: SystemTime,
90    pub ended_at: SystemTime,
91    pub elapsed: Duration,
92    pub result_count: Option<usize>,
93    pub result_type: Option<String>,
94    pub affected_rows: Option<u64>,
95    pub result_summary: String,
96}
97
98#[derive(Debug, Clone, PartialEq)]
99pub struct UnifiedLogEntry {
100    pub timestamp: SystemTime,
101    pub user_identifier: Option<String>,
102    pub trace_chain: Vec<teaql_core::TraceNode>,
103    pub payload: LogPayload,
104}
105
106#[derive(Debug, Clone, PartialEq)]
107pub enum LogPayload {
108    Sql(SqlLogEntry),
109    Info(InfoLogEntry),
110}
111
112#[derive(Debug, Clone, PartialEq)]
113pub struct InfoLogEntry {
114    pub message: String,
115}
116
117#[derive(Clone, Default)]
118pub struct UnifiedLogBuffer {
119    pub entries: Arc<Mutex<Vec<UnifiedLogEntry>>>,
120}
121
122impl UserContext {
123    pub fn with_sql_log_options(mut self, options: SqlLogOptions) -> Self {
124        self.sql_log_options = options;
125        self
126    }
127
128    pub fn set_sql_log_options(&mut self, options: SqlLogOptions) {
129        self.sql_log_options = options;
130    }
131
132    pub fn enable_select_sql_log(&mut self) {
133        self.sql_log_options.select = true;
134    }
135
136    pub fn enable_mutation_sql_log(&mut self) {
137        self.sql_log_options.mutation = true;
138    }
139
140    pub fn disable_select_sql_log(&mut self) {
141        self.sql_log_options.select = false;
142    }
143
144    pub fn disable_mutation_sql_log(&mut self) {
145        self.sql_log_options.mutation = false;
146    }
147
148    pub fn enable_all_sql_log(&mut self) {
149        self.sql_log_options = SqlLogOptions::all();
150    }
151
152    pub fn disable_sql_log(&mut self) {
153        self.sql_log_options = SqlLogOptions::disabled();
154        self.clear_sql_logs();
155    }
156
157    pub fn sql_log_options(&self) -> SqlLogOptions {
158        self.sql_log_options
159    }
160
161    pub fn sql_logs(&self) -> Vec<SqlLogEntry> {
162        self.sql_log_entries
163            .lock()
164            .map(|entries| entries.clone())
165            .unwrap_or_default()
166    }
167
168    pub fn clear_sql_logs(&self) {
169        if let Ok(mut entries) = self.sql_log_entries.lock() {
170            entries.clear();
171        }
172    }
173
174    #[allow(clippy::too_many_arguments)]
175    pub(crate) fn record_sql_log(
176        &self,
177        operation: SqlLogOperation,
178        query: &CompiledQuery,
179        database_kind: DatabaseKind,
180        started_at: SystemTime,
181        ended_at: SystemTime,
182        elapsed: Duration,
183        result_count: Option<usize>,
184        result_type: Option<String>,
185        affected_rows: Option<u64>,
186        trace_chain: Vec<teaql_core::TraceNode>,
187    ) {
188        if !self.sql_log_options.enabled_for(operation) {
189            return;
190        }
191        let debug_sql = query.debug_sql(database_kind);
192        let result_summary = sql_result_summary(
193            operation,
194            result_count,
195            result_type.as_deref(),
196            affected_rows,
197            &debug_sql,
198        );
199        let trace_path = canonical_sql_trace_path(
200            operation,
201            &format!("{database_kind:?}").to_ascii_lowercase(),
202            &trace_chain,
203        );
204        let entry = SqlLogEntry {
205            operation,
206            comment: trace_value(&trace_chain, teaql_core::TraceKind::Comment),
207            purpose: trace_value(&trace_chain, teaql_core::TraceKind::Purpose),
208            audit_reason: trace_value(&trace_chain, teaql_core::TraceKind::AuditReason),
209            trace_path: trace_path.clone(),
210            sql: query.sql.clone(),
211            params: query.params.clone(),
212            pretty_sql: pretty_sql(&debug_sql),
213            debug_sql,
214            started_at,
215            ended_at,
216            elapsed,
217            result_summary,
218            result_count,
219            result_type,
220            affected_rows,
221        };
222        self.append_sql_log(started_at, trace_path, entry);
223    }
224
225    pub(crate) fn record_metadata_log(&self, metadata: &teaql_data_service::ExecutionMetadata) {
226        let operation = match metadata.operation {
227            teaql_data_service::DataServiceOperation::Query => SqlLogOperation::Select,
228            teaql_data_service::DataServiceOperation::Insert => SqlLogOperation::Insert,
229            teaql_data_service::DataServiceOperation::Update => SqlLogOperation::Update,
230            teaql_data_service::DataServiceOperation::Delete => SqlLogOperation::Delete,
231            teaql_data_service::DataServiceOperation::Recover => SqlLogOperation::Update,
232            teaql_data_service::DataServiceOperation::Batch => SqlLogOperation::Update,
233            teaql_data_service::DataServiceOperation::Schema => SqlLogOperation::Update,
234        };
235        if !self.sql_log_options.enabled_for(operation) {
236            return;
237        }
238        let Some(debug_sql) = &metadata.debug_query else {
239            return;
240        };
241        let trace_path =
242            canonical_sql_trace_path(operation, &metadata.backend, &metadata.trace_chain);
243        let result_summary = metadata
244            .result_count
245            .map(|count| format!("{count} rows returned"))
246            .or_else(|| {
247                metadata
248                    .affected_rows
249                    .map(|affected| format!("{affected} rows affected"))
250            })
251            .unwrap_or_default();
252        let entry = SqlLogEntry {
253            operation,
254            comment: trace_value(&metadata.trace_chain, teaql_core::TraceKind::Comment)
255                .or_else(|| metadata.comment.clone()),
256            purpose: trace_value(&metadata.trace_chain, teaql_core::TraceKind::Purpose),
257            audit_reason: trace_value(&metadata.trace_chain, teaql_core::TraceKind::AuditReason),
258            trace_path: trace_path.clone(),
259            sql: metadata.parameterized_query.clone().unwrap_or_default(),
260            params: metadata.params.clone(),
261            pretty_sql: pretty_sql(debug_sql),
262            debug_sql: debug_sql.clone(),
263            started_at: metadata.started_at,
264            ended_at: metadata.ended_at,
265            elapsed: metadata
266                .ended_at
267                .duration_since(metadata.started_at)
268                .unwrap_or_default(),
269            result_count: metadata.result_count,
270            result_type: None,
271            affected_rows: metadata.affected_rows,
272            result_summary,
273        };
274        self.append_sql_log(metadata.started_at, trace_path, entry);
275    }
276
277    fn append_sql_log(
278        &self,
279        timestamp: SystemTime,
280        trace_path: Vec<teaql_core::TraceNode>,
281        entry: SqlLogEntry,
282    ) {
283        if let Ok(mut entries) = self.sql_log_entries.lock() {
284            entries.push(entry.clone());
285        }
286        if let Some(buffer) = self.get_resource::<UnifiedLogBuffer>() {
287            if let Ok(mut entries) = buffer.entries.lock() {
288                entries.push(UnifiedLogEntry {
289                    timestamp,
290                    user_identifier: self.user_identifier.clone(),
291                    trace_chain: trace_path.clone(),
292                    payload: LogPayload::Sql(entry.clone()),
293                });
294            }
295        }
296        crate::log_formatter::LogManager::write_sql_log(&trace_path, &entry);
297    }
298}
299
300fn extract_id_from_sql(sql: &str) -> Option<String> {
301    let sql_lower = sql.to_lowercase();
302    let where_clause = &sql_lower[sql_lower.find("where")? + 5..];
303    let bytes = where_clause.as_bytes();
304    let mut index = 0;
305    while index + 1 < bytes.len() {
306        if &bytes[index..index + 2] == b"id" {
307            let before_is_boundary = index == 0 || {
308                let previous = bytes[index - 1] as char;
309                !previous.is_ascii_alphanumeric() && previous != '_' && previous != '.'
310            };
311            let after_is_boundary = index + 2 == bytes.len() || {
312                let next = bytes[index + 2] as char;
313                !next.is_ascii_alphanumeric() && next != '_'
314            };
315            if before_is_boundary && after_is_boundary {
316                let mut value_index = index + 2;
317                while value_index < bytes.len() && (bytes[value_index] as char).is_whitespace() {
318                    value_index += 1;
319                }
320                if value_index < bytes.len() && bytes[value_index] == b'=' {
321                    value_index += 1;
322                    while value_index < bytes.len() && (bytes[value_index] as char).is_whitespace()
323                    {
324                        value_index += 1;
325                    }
326                    let quoted = value_index < bytes.len() && bytes[value_index] == b'\'';
327                    if quoted {
328                        value_index += 1;
329                    }
330                    let mut value = String::new();
331                    while value_index < bytes.len() {
332                        let character = bytes[value_index] as char;
333                        if (quoted && character == '\'')
334                            || (!quoted
335                                && !character.is_ascii_alphanumeric()
336                                && character != '_'
337                                && character != '-')
338                        {
339                            break;
340                        }
341                        value.push(character);
342                        value_index += 1;
343                    }
344                    if !value.is_empty() {
345                        return Some(value);
346                    }
347                }
348            }
349        }
350        index += 1;
351    }
352    None
353}
354
355fn sql_result_summary(
356    operation: SqlLogOperation,
357    result_count: Option<usize>,
358    result_type: Option<&str>,
359    affected_rows: Option<u64>,
360    debug_sql: &str,
361) -> String {
362    match operation {
363        SqlLogOperation::Select => match result_count.unwrap_or(0) {
364            0 => "MISS".to_owned(),
365            1 => result_type
366                .map(|result_type| {
367                    extract_id_from_sql(debug_sql)
368                        .map(|id| format!("{result_type}({id})"))
369                        .unwrap_or_else(|| result_type.to_owned())
370                })
371                .unwrap_or_else(|| "row".to_owned()),
372            count => result_type
373                .map(|result_type| format!("{count}*{result_type}"))
374                .unwrap_or_else(|| format!("{count}*rows")),
375        },
376        _ => format!("{} UPDATED", affected_rows.unwrap_or(0)),
377    }
378}
379
380fn trace_value(
381    trace_path: &[teaql_core::TraceNode],
382    kind: teaql_core::TraceKind,
383) -> Option<String> {
384    trace_path
385        .iter()
386        .rev()
387        .find(|node| node.kind == kind)
388        .map(|node| node.comment.clone())
389}
390
391fn canonical_sql_trace_path(
392    operation: SqlLogOperation,
393    backend: &str,
394    source: &[teaql_core::TraceNode],
395) -> Vec<teaql_core::TraceNode> {
396    use teaql_core::{TraceKind, TraceNode};
397
398    if source.iter().any(|node| node.kind == TraceKind::Operation)
399        && source.iter().any(|node| node.kind == TraceKind::Provider)
400        && source.iter().any(|node| node.kind == TraceKind::Sql)
401    {
402        return source
403            .iter()
404            .filter(|node| {
405                !matches!(
406                    node.kind,
407                    TraceKind::Comment | TraceKind::Purpose | TraceKind::AuditReason
408                )
409            })
410            .cloned()
411            .collect();
412    }
413    let entity = source
414        .iter()
415        .find(|node| !node.entity_type.trim().is_empty())
416        .map(|node| node.entity_type.clone())
417        .unwrap_or_else(|| "unknown".to_owned());
418    let family = if operation.is_select() {
419        "query"
420    } else {
421        "mutation"
422    };
423    let statement = match operation {
424        SqlLogOperation::Select => "select",
425        SqlLogOperation::Insert => "insert",
426        SqlLogOperation::Update => "update",
427        SqlLogOperation::Delete => "delete",
428        SqlLogOperation::Recover => "recover",
429    };
430    let mut path = vec![TraceNode::typed(
431        TraceKind::Operation,
432        entity.clone(),
433        None,
434        family,
435    )];
436    path.push(TraceNode::typed(
437        if operation.is_select() {
438            TraceKind::Request
439        } else {
440            TraceKind::Entity
441        },
442        entity,
443        None,
444        "",
445    ));
446    path.extend(
447        source
448            .iter()
449            .filter(|node| node.kind == TraceKind::Relation)
450            .cloned(),
451    );
452    path.push(TraceNode::typed(
453        TraceKind::Provider,
454        if backend.trim().is_empty() {
455            "unknown"
456        } else {
457            backend
458        },
459        None,
460        "",
461    ));
462    path.push(TraceNode::typed(TraceKind::Sql, statement, None, ""));
463    path
464}
465
466fn pretty_sql(sql: &str) -> String {
467    let mut pretty = sql.to_owned();
468    for keyword in [
469        " FROM ",
470        " WHERE ",
471        " GROUP BY ",
472        " HAVING ",
473        " ORDER BY ",
474        " LIMIT ",
475        " OFFSET ",
476        " RETURNING ",
477    ] {
478        pretty = pretty.replace(keyword, &format!("\n{}", keyword.trim_start()));
479    }
480    pretty.replace(" AND ", "\n  AND ")
481}
teaql_runtime/context/logging.rs

teaql_runtime/context/
logging.rs
