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}