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)] pub 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 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}