teaql_runtime/
context.rs

1use std::any::{Any, TypeId};
2use std::collections::{BTreeMap, HashMap};
3use std::future::Future;
4
5use std::pin::Pin;
6use std::sync::Mutex;
7use std::time::{Duration, SystemTime};
8
9use teaql_core::{EntityDescriptor, Record, UpdateCommand, Value};
10use teaql_sql::{CompiledQuery, DatabaseKind};
11
12use crate::{
13    CheckResults, CheckerRegistry, ContextError, EntityDataServiceBehavior,
14    EntityDataServiceBehaviorRegistry, EntityRegistry, GraphNode, InternalIdGenerator, Language,
15    MetadataStore, ObjectLocation, RawAuditEvent, RawAuditEventSink, RequestPolicy, RuntimeError,
16    local_id_generator, translate_check_result,
17};
18use crate::{DataServiceError, EntityRoot};
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum SqlLogOperation {
22    Select,
23    Insert,
24    Update,
25    Delete,
26    Recover,
27}
28
29impl SqlLogOperation {
30    pub fn is_select(self) -> bool {
31        matches!(self, Self::Select)
32    }
33
34    pub fn is_mutation(self) -> bool {
35        !self.is_select()
36    }
37}
38
39#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
40pub struct SqlLogOptions {
41    pub select: bool,
42    pub mutation: bool,
43}
44
45impl SqlLogOptions {
46    pub fn disabled() -> Self {
47        Self {
48            select: false,
49            mutation: false,
50        }
51    }
52
53    pub fn select_only() -> Self {
54        Self {
55            select: true,
56            mutation: false,
57        }
58    }
59
60    pub fn mutation_only() -> Self {
61        Self {
62            select: false,
63            mutation: true,
64        }
65    }
66
67    pub fn all() -> Self {
68        Self {
69            select: true,
70            mutation: true,
71        }
72    }
73
74    pub fn enabled_for(self, operation: SqlLogOperation) -> bool {
75        if operation.is_select() {
76            self.select
77        } else {
78            self.mutation
79        }
80    }
81}
82
83#[derive(Debug, Clone, PartialEq)]
84pub struct SqlLogEntry {
85    pub operation: SqlLogOperation,
86    pub sql: String,
87    pub params: Vec<Value>,
88    pub debug_sql: String,
89    pub pretty_sql: String,
90    pub started_at: SystemTime,
91    pub ended_at: SystemTime,
92    pub elapsed: Duration,
93    pub result_count: Option<usize>,
94    pub result_type: Option<String>,
95    pub affected_rows: Option<u64>,
96    pub result_summary: String,
97}
98
99#[derive(Debug, Clone, PartialEq)]
100pub struct UnifiedLogEntry {
101    pub timestamp: SystemTime,
102    pub user_identifier: Option<String>,
103    pub trace_chain: Vec<teaql_core::TraceNode>,
104    pub payload: LogPayload,
105}
106
107#[derive(Debug, Clone, PartialEq)]
108pub enum LogPayload {
109    Sql(SqlLogEntry),
110    Info(InfoLogEntry),
111}
112
113#[derive(Debug, Clone, PartialEq)]
114pub struct InfoLogEntry {
115    pub message: String,
116}
117
118#[derive(Clone, Default)]
119pub struct UnifiedLogBuffer {
120    pub entries: std::sync::Arc<Mutex<Vec<UnifiedLogEntry>>>,
121}
122
123pub trait SchemaProvider: Send + Sync {
124    fn ensure_schema<'a>(
125        &'a self,
126        ctx: &'a UserContext,
127    ) -> Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>>;
128}
129
130pub struct UserContext {
131    pub(crate) metadata: Option<Box<dyn MetadataStore>>,
132    pub(crate) entity_registry: Option<Box<dyn EntityRegistry>>,
133    pub(crate) entity_data_service_behavior_registry:
134        Option<Box<dyn EntityDataServiceBehaviorRegistry>>,
135    pub(crate) request_policy: Option<Box<dyn RequestPolicy>>,
136    pub(crate) checker_registry: Option<Box<dyn CheckerRegistry>>,
137    pub(crate) event_sink: Option<Box<dyn RawAuditEventSink>>,
138    pub(crate) custom_event_sink: Option<Box<dyn crate::SafeAuditEventSink>>,
139    pub(crate) internal_id_generator: Option<Box<dyn InternalIdGenerator>>,
140    schema_provider: Option<Box<dyn SchemaProvider>>,
141    language: Language,
142    typed_resources: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
143    named_resources: BTreeMap<String, Box<dyn Any + Send + Sync>>,
144    locals: BTreeMap<String, Value>,
145    pub(crate) initial_graphs: Vec<GraphNode>,
146    entity_root: EntityRoot,
147    sql_log_options: SqlLogOptions,
148    sql_log_entries: Mutex<Vec<SqlLogEntry>>,
149    user_identifier: Option<String>,
150    timezone: Option<String>,
151    trace_id: String,
152}
153
154impl Default for UserContext {
155    fn default() -> Self {
156        let pid = std::process::id();
157        let thread_id_str = format!("{:?}", std::thread::current().id());
158        let numeric_thread_id = thread_id_str
159            .strip_prefix("ThreadId(")
160            .and_then(|s| s.strip_suffix(")"))
161            .unwrap_or(&thread_id_str);
162        let os_user = std::env::var("USER")
163            .or_else(|_| std::env::var("USERNAME"))
164            .unwrap_or_else(|_| "main".to_owned());
165        let user_id = format!("{os_user}@pid-{pid}.tid-{numeric_thread_id}");
166        Self {
167            metadata: None,
168            entity_registry: None,
169            entity_data_service_behavior_registry: None,
170            request_policy: None,
171            checker_registry: None,
172            event_sink: None,
173            custom_event_sink: None,
174            internal_id_generator: None,
175            schema_provider: None,
176            language: Language::default(),
177            typed_resources: HashMap::new(),
178            named_resources: BTreeMap::new(),
179            locals: BTreeMap::new(),
180            initial_graphs: Vec::new(),
181            entity_root: EntityRoot::default(),
182            sql_log_options: SqlLogOptions::all(),
183            sql_log_entries: Mutex::new(Vec::new()),
184            user_identifier: Some(user_id),
185            timezone: Some("UTC".to_owned()),
186            trace_id: format!(
187                "req-{pid}-{numeric_thread_id}-{:x}",
188                std::time::SystemTime::now()
189                    .duration_since(std::time::UNIX_EPOCH)
190                    .unwrap_or_default()
191                    .as_micros()
192            ),
193        }
194    }
195}
196
197#[async_trait::async_trait]
198pub trait DataStore: Send + Sync + 'static {
199    async fn get(&self, key: &str) -> Option<Value>;
200    async fn put(&self, key: &str, value: Value, timeout_seconds: Option<u64>);
201    async fn remove(&self, key: &str);
202}
203
204#[derive(Default)]
205pub struct InMemoryDataStore {
206    cache: std::sync::RwLock<HashMap<String, (Value, Option<std::time::Instant>)>>,
207}
208
209#[async_trait::async_trait]
210impl DataStore for InMemoryDataStore {
211    async fn get(&self, key: &str) -> Option<Value> {
212        let lock = self.cache.read().unwrap();
213        if let Some((val, expires_at)) = lock.get(key) {
214            if let Some(exp) = expires_at {
215                if std::time::Instant::now() > *exp {
216                    return None;
217                }
218            }
219            return Some(val.clone());
220        }
221        None
222    }
223
224    async fn put(&self, key: &str, value: Value, timeout_seconds: Option<u64>) {
225        let mut lock = self.cache.write().unwrap();
226        let expires_at = timeout_seconds
227            .map(|secs| std::time::Instant::now() + std::time::Duration::from_secs(secs));
228        lock.insert(key.to_string(), (value, expires_at));
229    }
230
231    async fn remove(&self, key: &str) {
232        let mut lock = self.cache.write().unwrap();
233        lock.remove(key);
234    }
235}
236
237impl UserContext {
238    pub fn new() -> Self {
239        Self::default()
240    }
241
242    pub fn user_identifier(&self) -> Option<&str> {
243        self.user_identifier.as_deref()
244    }
245
246    pub fn set_user_identifier(&mut self, user_identifier: impl Into<String>) {
247        self.user_identifier = Some(user_identifier.into());
248    }
249
250    pub fn with_user_identifier(mut self, user_identifier: impl Into<String>) -> Self {
251        self.user_identifier = Some(user_identifier.into());
252        self
253    }
254
255    pub fn set_user_identifier_option(&mut self, user_identifier: Option<String>) {
256        self.user_identifier = user_identifier;
257    }
258
259    pub fn with_user_identifier_option(mut self, user_identifier: Option<String>) -> Self {
260        self.user_identifier = user_identifier;
261        self
262    }
263
264    pub fn timezone(&self) -> Option<&str> {
265        self.timezone.as_deref()
266    }
267
268    pub fn set_timezone(&mut self, timezone: impl Into<String>) {
269        self.timezone = Some(timezone.into());
270    }
271
272    pub fn with_timezone(mut self, timezone: impl Into<String>) -> Self {
273        self.timezone = Some(timezone.into());
274        self
275    }
276
277    pub fn trace_id(&self) -> &str {
278        &self.trace_id
279    }
280
281    pub fn set_trace_id(&mut self, trace_id: impl Into<String>) {
282        self.trace_id = trace_id.into();
283    }
284
285    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
286        self.trace_id = trace_id.into();
287        self
288    }
289
290    pub fn with_module(mut self, module: crate::RuntimeModule) -> Self {
291        module.apply_to(&mut self);
292        self
293    }
294
295    pub fn entity_root(&self) -> EntityRoot {
296        self.entity_root.clone()
297    }
298
299    pub fn initial_graphs(&self) -> &[GraphNode] {
300        &self.initial_graphs
301    }
302
303    pub fn set_initial_graphs(&mut self, graphs: Vec<GraphNode>) {
304        self.initial_graphs = graphs;
305    }
306
307    pub fn with_metadata(mut self, metadata: impl MetadataStore + 'static) -> Self {
308        self.metadata = Some(Box::new(metadata));
309        self
310    }
311
312    pub fn set_metadata(&mut self, metadata: impl MetadataStore + 'static) {
313        self.metadata = Some(Box::new(metadata));
314    }
315
316    pub fn with_entity_registry(mut self, registry: impl EntityRegistry + 'static) -> Self {
317        self.entity_registry = Some(Box::new(registry));
318        self
319    }
320
321    pub fn set_entity_registry(&mut self, registry: impl EntityRegistry + 'static) {
322        self.entity_registry = Some(Box::new(registry));
323    }
324
325    pub fn with_entity_data_service_behavior_registry(
326        mut self,
327        registry: impl EntityDataServiceBehaviorRegistry + 'static,
328    ) -> Self {
329        self.entity_data_service_behavior_registry = Some(Box::new(registry));
330        self
331    }
332
333    pub fn set_entity_data_service_behavior_registry(
334        &mut self,
335        registry: impl EntityDataServiceBehaviorRegistry + 'static,
336    ) {
337        self.entity_data_service_behavior_registry = Some(Box::new(registry));
338    }
339
340    pub fn with_request_policy(mut self, policy: impl RequestPolicy + 'static) -> Self {
341        self.request_policy = Some(Box::new(policy));
342        self
343    }
344
345    pub fn set_request_policy(&mut self, policy: impl RequestPolicy + 'static) {
346        self.request_policy = Some(Box::new(policy));
347    }
348
349    pub fn clear_request_policy(&mut self) {
350        self.request_policy = None;
351    }
352
353    pub fn with_checker_registry(mut self, registry: impl CheckerRegistry + 'static) -> Self {
354        self.checker_registry = Some(Box::new(registry));
355        self
356    }
357
358    pub fn set_checker_registry(&mut self, registry: impl CheckerRegistry + 'static) {
359        self.checker_registry = Some(Box::new(registry));
360    }
361
362    pub(crate) fn with_event_sink(mut self, sink: impl RawAuditEventSink + 'static) -> Self {
363        self.event_sink = Some(Box::new(sink));
364        self
365    }
366
367    pub(crate) fn set_event_sink(&mut self, sink: impl RawAuditEventSink + 'static) {
368        self.event_sink = Some(Box::new(sink));
369    }
370
371    pub fn with_custom_event_sink(
372        mut self,
373        sink: impl crate::SafeAuditEventSink + 'static,
374    ) -> Self {
375        self.custom_event_sink = Some(Box::new(sink));
376        self
377    }
378
379    pub fn set_custom_event_sink(&mut self, sink: impl crate::SafeAuditEventSink + 'static) {
380        self.custom_event_sink = Some(Box::new(sink));
381    }
382
383    pub fn with_internal_id_generator(
384        mut self,
385        generator: impl InternalIdGenerator + 'static,
386    ) -> Self {
387        self.internal_id_generator = Some(Box::new(generator));
388        self
389    }
390
391    pub fn set_internal_id_generator(&mut self, generator: impl InternalIdGenerator + 'static) {
392        self.internal_id_generator = Some(Box::new(generator));
393    }
394
395    pub fn with_schema_provider(mut self, provider: impl SchemaProvider + 'static) -> Self {
396        self.schema_provider = Some(Box::new(provider));
397        self
398    }
399
400    pub fn set_schema_provider(&mut self, provider: impl SchemaProvider + 'static) {
401        self.schema_provider = Some(Box::new(provider));
402    }
403
404    pub async fn ensure_schema(&self) -> Result<(), RuntimeError> {
405        let provider = self
406            .schema_provider
407            .as_ref()
408            .ok_or_else(|| RuntimeError::Schema("missing schema provider".to_owned()))?;
409        provider.ensure_schema(self).await
410    }
411
412    pub fn with_language(mut self, language: Language) -> Self {
413        self.language = language;
414        self
415    }
416
417    pub fn set_language(&mut self, language: Language) {
418        self.language = language;
419    }
420
421    pub fn with_sql_log_options(mut self, options: SqlLogOptions) -> Self {
422        self.sql_log_options = options;
423        self
424    }
425
426    pub fn set_sql_log_options(&mut self, options: SqlLogOptions) {
427        self.sql_log_options = options;
428    }
429
430    pub fn enable_select_sql_log(&mut self) {
431        self.sql_log_options.select = true;
432    }
433
434    pub fn enable_mutation_sql_log(&mut self) {
435        self.sql_log_options.mutation = true;
436    }
437
438    pub fn enable_all_sql_log(&mut self) {
439        self.sql_log_options = SqlLogOptions::all();
440    }
441
442    pub fn disable_sql_log(&mut self) {
443        self.sql_log_options = SqlLogOptions::disabled();
444        self.clear_sql_logs();
445    }
446
447    pub fn sql_log_options(&self) -> SqlLogOptions {
448        self.sql_log_options
449    }
450
451    pub fn sql_logs(&self) -> Vec<SqlLogEntry> {
452        self.sql_log_entries
453            .lock()
454            .map(|entries| entries.clone())
455            .unwrap_or_default()
456    }
457
458    pub fn clear_sql_logs(&self) {
459        if let Ok(mut entries) = self.sql_log_entries.lock() {
460            entries.clear();
461        }
462    }
463
464    pub(crate) fn record_sql_log(
465        &self,
466        operation: SqlLogOperation,
467        query: &CompiledQuery,
468        database_kind: DatabaseKind,
469        started_at: SystemTime,
470        ended_at: SystemTime,
471        elapsed: Duration,
472        result_count: Option<usize>,
473        result_type: Option<String>,
474        affected_rows: Option<u64>,
475        trace_chain: Vec<teaql_core::TraceNode>,
476    ) {
477        if !self.sql_log_options.enabled_for(operation) {
478            return;
479        }
480        let debug_sql = query.debug_sql(database_kind);
481        let result_summary = sql_result_summary(
482            operation,
483            result_count,
484            result_type.as_deref(),
485            affected_rows,
486            &debug_sql,
487        );
488
489        let sql_log_entry = SqlLogEntry {
490            operation,
491            sql: query.sql.clone(),
492            params: query.params.clone(),
493            pretty_sql: pretty_sql(&debug_sql),
494            debug_sql: debug_sql.clone(),
495            started_at,
496            ended_at,
497            elapsed,
498            result_summary: result_summary.clone(),
499            result_count,
500            result_type,
501            affected_rows,
502        };
503
504        if let Ok(mut entries) = self.sql_log_entries.lock() {
505            // Keep sql_log_entries backwards-compatible for now if needed,
506            // wait, we modified SqlLogEntry. We can just push it directly since we removed comment.
507            // Wait, we need to push a cloned SqlLogEntry since it doesn't have comment.
508            entries.push(sql_log_entry.clone());
509        }
510
511        if let Some(buf) = self.get_resource::<UnifiedLogBuffer>() {
512            if let Ok(mut entries) = buf.entries.lock() {
513                entries.push(UnifiedLogEntry {
514                    timestamp: started_at,
515                    user_identifier: self.user_identifier.clone(),
516                    trace_chain: trace_chain.clone(),
517                    payload: LogPayload::Sql(sql_log_entry.clone()),
518                });
519            }
520        }
521
522        crate::log_formatter::LogManager::write_sql_log(&trace_chain, &sql_log_entry);
523    }
524
525    pub(crate) fn record_metadata_log(&self, metadata: &teaql_data_service::ExecutionMetadata) {
526        if let Some(debug_sql) = &metadata.debug_query {
527            let sql_log_entry = SqlLogEntry {
528                operation: match metadata.operation {
529                    teaql_data_service::DataServiceOperation::Query => SqlLogOperation::Select,
530                    teaql_data_service::DataServiceOperation::Insert => SqlLogOperation::Insert,
531                    teaql_data_service::DataServiceOperation::Update => SqlLogOperation::Update,
532                    teaql_data_service::DataServiceOperation::Delete => SqlLogOperation::Delete,
533                    teaql_data_service::DataServiceOperation::Recover => SqlLogOperation::Update, // Approximate
534                    teaql_data_service::DataServiceOperation::Batch => SqlLogOperation::Update,
535                    teaql_data_service::DataServiceOperation::Schema => SqlLogOperation::Update,
536                },
537                sql: String::new(), // Not available in metadata
538                params: Vec::new(), // Not available in metadata
539                pretty_sql: pretty_sql(debug_sql),
540                debug_sql: debug_sql.clone(),
541                started_at: metadata.started_at,
542                ended_at: metadata.ended_at,
543                elapsed: metadata
544                    .ended_at
545                    .duration_since(metadata.started_at)
546                    .unwrap_or_default(),
547                result_count: metadata.result_count,
548                result_type: None, // Not directly available
549                affected_rows: metadata.affected_rows,
550                result_summary: String::new(), // We can synthesize this if needed, or leave it empty/basic
551            };
552
553            // synthesize a summary for the log
554            let mut summary = String::new();
555            if let Some(c) = metadata.result_count {
556                summary = format!("{} rows returned", c);
557            } else if let Some(a) = metadata.affected_rows {
558                summary = format!("{} rows affected", a);
559            }
560
561            let mut final_entry = sql_log_entry;
562            final_entry.result_summary = summary;
563
564            if let Ok(mut entries) = self.sql_log_entries.lock() {
565                entries.push(final_entry.clone());
566            }
567
568            if let Some(buf) = self.get_resource::<UnifiedLogBuffer>() {
569                if let Ok(mut entries) = buf.entries.lock() {
570                    entries.push(UnifiedLogEntry {
571                        timestamp: metadata.started_at,
572                        user_identifier: self.user_identifier.clone(),
573                        trace_chain: metadata.trace_chain.clone(),
574                        payload: LogPayload::Sql(final_entry.clone()),
575                    });
576                }
577            }
578
579            crate::log_formatter::LogManager::write_sql_log(&metadata.trace_chain, &final_entry);
580        }
581    }
582
583    pub fn language(&self) -> Language {
584        self.language
585    }
586
587    pub fn set_language_code(&mut self, code: &str) -> Result<(), RuntimeError> {
588        let Some(language) = Language::from_code(code) else {
589            return Err(RuntimeError::Language(format!(
590                "unsupported language code: {code}"
591            )));
592        };
593        self.language = language;
594        Ok(())
595    }
596
597    pub fn generate_id(&self, entity: &str) -> Result<Option<u64>, RuntimeError> {
598        self.internal_id_generator
599            .as_ref()
600            .map(|generator| generator.generate_id(entity))
601            .transpose()
602    }
603
604    pub fn next_id(&self, entity: &str) -> Result<u64, RuntimeError> {
605        match self.generate_id(entity)? {
606            Some(id) => Ok(id),
607            None => local_id_generator().generate_id(entity),
608        }
609    }
610
611    pub fn entity(&self, name: &str) -> Option<&EntityDescriptor> {
612        self.metadata
613            .as_ref()
614            .and_then(|metadata| metadata.entity(name))
615    }
616
617    pub fn all_entities(&self) -> Vec<&EntityDescriptor> {
618        self.metadata
619            .as_ref()
620            .map(|metadata| metadata.all_entities())
621            .unwrap_or_default()
622    }
623
624    pub fn require_entity(&self, name: &str) -> Result<&EntityDescriptor, RuntimeError> {
625        self.entity(name)
626            .ok_or_else(|| RuntimeError::MissingEntity(name.to_owned()))
627    }
628
629    pub fn insert_resource<T>(&mut self, resource: T)
630    where
631        T: Send + Sync + 'static,
632    {
633        self.typed_resources
634            .insert(TypeId::of::<T>(), Box::new(resource));
635    }
636
637    pub fn get_resource<T>(&self) -> Option<&T>
638    where
639        T: Send + Sync + 'static,
640    {
641        self.typed_resources
642            .get(&TypeId::of::<T>())
643            .and_then(|value| value.downcast_ref::<T>())
644    }
645
646    pub fn require_resource<T>(&self) -> Result<&T, ContextError>
647    where
648        T: Send + Sync + 'static,
649    {
650        self.get_resource::<T>()
651            .ok_or(ContextError::MissingTypedResource(
652                std::any::type_name::<T>(),
653            ))
654    }
655
656    pub fn insert_named_resource<T>(&mut self, name: impl Into<String>, resource: T)
657    where
658        T: Send + Sync + 'static,
659    {
660        self.named_resources.insert(name.into(), Box::new(resource));
661    }
662
663    pub fn get_named_resource<T>(&self, name: &str) -> Option<&T>
664    where
665        T: Send + Sync + 'static,
666    {
667        self.named_resources
668            .get(name)
669            .and_then(|value| value.downcast_ref::<T>())
670    }
671
672    pub fn require_named_resource<T>(&self, name: &str) -> Result<&T, ContextError>
673    where
674        T: Send + Sync + 'static,
675    {
676        self.get_named_resource::<T>(name)
677            .ok_or_else(|| ContextError::MissingResource(name.to_owned()))
678    }
679
680    pub fn put_local(&mut self, key: impl Into<String>, value: impl Into<Value>) {
681        self.locals.insert(key.into(), value.into());
682    }
683
684    pub fn local(&self, key: &str) -> Option<&Value> {
685        self.locals.get(key)
686    }
687
688    pub fn remove_local(&mut self, key: &str) -> Option<Value> {
689        self.locals.remove(key)
690    }
691
692    pub fn has_entity_data_service(&self, entity: &str) -> bool {
693        let in_registry = self
694            .entity_registry
695            .as_ref()
696            .map(|registry| registry.contains(entity))
697            .unwrap_or(false);
698        in_registry || self.entity(entity).is_some()
699    }
700
701    pub fn entity_data_service_behavior(
702        &self,
703        entity: &str,
704    ) -> Option<std::sync::Arc<dyn EntityDataServiceBehavior>> {
705        self.entity_data_service_behavior_registry
706            .as_ref()
707            .and_then(|registry| registry.behavior(entity))
708    }
709
710    pub fn has_checker(&self, entity: &str) -> bool {
711        self.checker_registry
712            .as_ref()
713            .and_then(|registry| registry.checker(entity))
714            .is_some()
715    }
716
717    pub fn check_and_fix_record(
718        &self,
719        entity: &str,
720        record: &mut Record,
721    ) -> Result<(), RuntimeError> {
722        self.check_and_fix_record_at(entity, record, &ObjectLocation::root())
723    }
724
725    pub fn check_and_fix_record_at(
726        &self,
727        entity: &str,
728        record: &mut Record,
729        location: &ObjectLocation,
730    ) -> Result<(), RuntimeError> {
731        let Some(checker) = self
732            .checker_registry
733            .as_ref()
734            .and_then(|registry| registry.checker(entity))
735        else {
736            return Ok(());
737        };
738        let mut results = CheckResults::new();
739        checker.check_and_fix(self, record, location, &mut results);
740        if results.is_empty() {
741            Ok(())
742        } else {
743            self.translate_check_results(&mut results);
744            Err(RuntimeError::Check(results))
745        }
746    }
747
748    pub fn translate_check_results(&self, results: &mut CheckResults) {
749        for result in results {
750            result.message = Some(translate_check_result(self.language, result));
751        }
752    }
753
754    pub fn send_event(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
755        if let Some(sink) = self.event_sink.as_ref() {
756            sink.on_event(self, &event)?;
757        }
758        if let Some(sink) = self.custom_event_sink.as_ref() {
759            let (mask_fields, max_len) = if let Some(metadata) = &self.metadata {
760                if let Some(desc) = metadata.entity(&event.entity) {
761                    (desc.audit_mask_fields.clone(), desc.audit_value_max_len)
762                } else {
763                    (vec![], None)
764                }
765            } else {
766                (vec![], None)
767            };
768
769            let safe_event = event.build_safe_event(&mask_fields, max_len);
770            sink.on_safe_event(self, &safe_event)?;
771        }
772
773        crate::log_formatter::LogManager::write_audit_log(&event);
774
775        Ok(())
776    }
777
778    pub async fn commit_changes<E>(&self) -> Result<(), DataServiceError<E::Error>>
779    where
780        E: teaql_data_service::MutationExecutor + Send + Sync + 'static,
781    {
782        let executor = self.require_resource::<E>().map_err(|err| {
783            DataServiceError::Runtime(RuntimeError::Graph(format!(
784                "cannot commit changes without executor: {err}"
785            )))
786        })?;
787        let change_set = self.entity_root.current_change_set();
788
789        for (key, changes) in change_set.changes() {
790            if changes.is_empty() {
791                continue;
792            }
793            let _entity = self
794                .require_entity(&key.entity)
795                .map_err(DataServiceError::Runtime)?;
796            let mut command = UpdateCommand::new(&key.entity, key.id.clone());
797            for (field, value) in changes {
798                command = command.value(field.clone(), value.clone());
799            }
800            let request = teaql_data_service::MutationRequest::Update(command);
801            executor
802                .mutate(request)
803                .await
804                .map_err(DataServiceError::Executor)?;
805        }
806
807        self.entity_root.clear_current_change_set();
808        Ok(())
809    }
810
811    pub async fn get_in_store(&self, key: &str) -> Option<Value> {
812        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
813            store.get(key).await
814        } else {
815            None
816        }
817    }
818
819    pub async fn put_in_store(
820        &self,
821        key: &str,
822        value: impl Into<Value>,
823        timeout_seconds: Option<u64>,
824    ) {
825        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
826            store.put(key, value.into(), timeout_seconds).await;
827        }
828    }
829
830    pub async fn clear_in_store(&self, key: &str) {
831        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
832            store.remove(key).await;
833        }
834    }
835}
836
837fn extract_id_from_sql(sql: &str) -> Option<String> {
838    let sql_lower = sql.to_lowercase();
839    let where_idx = sql_lower.find("where")?;
840    let where_clause = &sql_lower[where_idx + 5..];
841
842    let bytes = where_clause.as_bytes();
843    let mut i = 0;
844    while i < bytes.len() {
845        if i + 1 < bytes.len() && &bytes[i..i + 2] == b"id" {
846            // Check boundary before
847            let prev_ok = if i == 0 {
848                true
849            } else {
850                let prev_char = bytes[i - 1] as char;
851                !prev_char.is_ascii_alphanumeric() && prev_char != '_' && prev_char != '.'
852            };
853            // Check boundary after
854            let next_ok = if i + 2 == bytes.len() {
855                true
856            } else {
857                let next_char = bytes[i + 2] as char;
858                !next_char.is_ascii_alphanumeric() && next_char != '_'
859            };
860
861            if prev_ok && next_ok {
862                // Found the standalone "id" word!
863                // Now look for "=" after it
864                let mut j = i + 2;
865                while j < bytes.len() && (bytes[j] as char).is_whitespace() {
866                    j += 1;
867                }
868                if j < bytes.len() && bytes[j] == b'=' {
869                    j += 1;
870                    while j < bytes.len() && (bytes[j] as char).is_whitespace() {
871                        j += 1;
872                    }
873                    // Now extract the value
874                    let mut val_str = String::new();
875                    if j < bytes.len() && bytes[j] == b'\'' {
876                        j += 1; // consume single quote
877                        while j < bytes.len() && bytes[j] != b'\'' {
878                            val_str.push(bytes[j] as char);
879                            j += 1;
880                        }
881                        return Some(val_str);
882                    } else {
883                        while j < bytes.len() {
884                            let c = bytes[j] as char;
885                            if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
886                                val_str.push(c);
887                                j += 1;
888                            } else {
889                                break;
890                            }
891                        }
892                        if !val_str.is_empty() {
893                            return Some(val_str);
894                        }
895                    }
896                }
897            }
898        }
899        i += 1;
900    }
901    None
902}
903
904fn sql_result_summary(
905    operation: SqlLogOperation,
906    result_count: Option<usize>,
907    result_type: Option<&str>,
908    affected_rows: Option<u64>,
909    debug_sql: &str,
910) -> String {
911    match operation {
912        SqlLogOperation::Select => {
913            let count = result_count.unwrap_or(0);
914            if count == 0 {
915                "MISS".to_owned()
916            } else if count > 1 {
917                match result_type {
918                    Some(result_type) => format!("{count}*{result_type}"),
919                    None => format!("{count}*rows"),
920                }
921            } else {
922                match result_type {
923                    Some(result_type) => {
924                        if let Some(id) = extract_id_from_sql(debug_sql) {
925                            format!("{result_type}({id})")
926                        } else {
927                            result_type.to_owned()
928                        }
929                    }
930                    None => "row".to_owned(),
931                }
932            }
933        }
934        _ => {
935            let affected = affected_rows.unwrap_or(0);
936            format!("{affected} UPDATED")
937        }
938    }
939}
940
941fn pretty_sql(sql: &str) -> String {
942    let mut pretty = sql.to_owned();
943    for keyword in [
944        " FROM ",
945        " WHERE ",
946        " GROUP BY ",
947        " HAVING ",
948        " ORDER BY ",
949        " LIMIT ",
950        " OFFSET ",
951        " RETURNING ",
952    ] {
953        pretty = pretty.replace(keyword, &format!("\n{}", keyword.trim_start()));
954    }
955    pretty.replace(" AND ", "\n  AND ")
956}
teaql_runtime/context.rs

teaql_runtime/
context.rs
