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

teaql_runtime/
context.rs
