Skip to main content

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::{Arc, Mutex, OnceLock};
7use std::time::SystemTime;
8
9use crate::EntityRuntimeState;
10use crate::{
11    CheckObjectStatus, CheckResult, CheckResults, CheckerRegistry, ContextError,
12    EntityDataServiceBehavior, EntityDataServiceBehaviorRegistry, EntityGraphBuilder,
13    EntityRegistry, GraphNode, InMemoryEntityGraphDecoderRegistry, InternalIdGenerator, Language,
14    MetadataStore, ObjectLocation, RawAuditEvent, RawAuditEventSink, RequestPolicy, RuntimeError,
15    local_id_generator,
16};
17use teaql_core::{EntityDescriptor, Value};
18
19mod locking;
20mod logging;
21mod pagination;
22mod transaction;
23pub use locking::RemoteLockProvider;
24use locking::next_local_lock_owner;
25pub use logging::{
26    InfoLogEntry, LogPayload, SqlLogEntry, SqlLogOperation, SqlLogOptions, UnifiedLogBuffer,
27    UnifiedLogEntry,
28};
29pub use pagination::{
30    ContinuousPageCursor, ContinuousPageCursorStore, IdSetStore, InMemoryContinuousPageCursorStore,
31    InMemoryIdSetStore, RetainedIdSet,
32};
33pub use transaction::TransactionScope;
34
35tokio::task_local! {
36    static GENERATED_SCHEMA_BOOTSTRAP_MODE: ();
37}
38
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct ContextEntityRef {
41    pub entity_type: String,
42    pub id: u64,
43}
44
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct ContextRootError {
47    pub expected_entity_type: String,
48    pub actual_root: Option<ContextEntityRef>,
49}
50
51impl std::fmt::Display for ContextRootError {
52    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        match &self.actual_root {
54            None => write!(
55                formatter,
56                "active root {} is missing from UserContext",
57                self.expected_entity_type
58            ),
59            Some(actual) => write!(
60                formatter,
61                "active root type is {}, expected {}",
62                actual.entity_type, self.expected_entity_type
63            ),
64        }
65    }
66}
67
68impl std::error::Error for ContextRootError {}
69
70#[cfg(test)]
71mod active_root_tests {
72    use super::UserContext;
73
74    #[test]
75    fn active_root_is_typed_and_fails_closed() {
76        let context = UserContext::new().with_active_root("Tenant", 42);
77        assert_eq!(context.require_active_root("Tenant").unwrap().id, 42);
78        assert!(context.require_active_root("Organization").is_err());
79        assert!(UserContext::new().require_active_root("Tenant").is_err());
80    }
81}
82
83/// Context-owned proof required by the provider SPI. Its private field prevents
84/// application crates from invoking a schema provider directly.
85///
86/// ```compile_fail
87/// let _ = teaql_runtime::SchemaInvocation { _context_owned: () };
88/// ```
89pub struct SchemaInvocation {
90    _context_owned: (),
91}
92
93pub trait SchemaProvider: Send + Sync {
94    fn ensure_schema<'a>(
95        &'a self,
96        context: &'a UserContext,
97        invocation: &'a SchemaInvocation,
98    ) -> Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>>;
99}
100
101pub type GeneratedSchemaBootstrapFuture<'a> =
102    Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>>;
103pub type GeneratedSchemaBootstrap =
104    for<'a> fn(&'a UserContext) -> GeneratedSchemaBootstrapFuture<'a>;
105
106#[derive(Clone, Debug, PartialEq, Eq)]
107pub enum FixEvidenceSource {
108    Clock,
109    Context,
110}
111
112#[derive(Clone, Debug, PartialEq, Eq)]
113pub struct FixEvidence {
114    pub entity_type: String,
115    pub model_path: String,
116    pub source: FixEvidenceSource,
117    pub source_label: String,
118}
119
120impl FixEvidence {
121    pub fn new(
122        entity_type: &str,
123        model_path: &str,
124        source: FixEvidenceSource,
125        source_label: &str,
126    ) -> Self {
127        assert!(
128            !entity_type.trim().is_empty(),
129            "entity_type must not be blank"
130        );
131        assert!(
132            !model_path.trim().is_empty(),
133            "model_path must not be blank"
134        );
135        assert!(
136            !source_label.trim().is_empty(),
137            "source_label must not be blank"
138        );
139        let normalized = source_label.to_ascii_lowercase();
140        assert!(
141            !normalized.contains("authorization")
142                && !normalized.contains("cookie")
143                && !normalized.contains("token="),
144            "source_label must be a safe framework label"
145        );
146        Self {
147            entity_type: entity_type.to_owned(),
148            model_path: model_path.to_owned(),
149            source,
150            source_label: source_label.to_owned(),
151        }
152    }
153}
154
155pub struct UserContext {
156    active_root: OnceLock<ContextEntityRef>,
157    pub(crate) metadata: Option<Box<dyn MetadataStore>>,
158    pub(crate) entity_registry: Option<Box<dyn EntityRegistry>>,
159    pub(crate) entity_graph_decoders: InMemoryEntityGraphDecoderRegistry,
160    pub(crate) entity_data_service_behavior_registry:
161        Option<Box<dyn EntityDataServiceBehaviorRegistry>>,
162    pub(crate) request_policy: Option<Box<dyn RequestPolicy>>,
163    pub(crate) checker_registry: Option<Box<dyn CheckerRegistry>>,
164    pub(crate) event_sink: Option<Box<dyn RawAuditEventSink>>,
165    pub(crate) custom_event_sink: Option<Box<dyn crate::SafeAuditEventSink>>,
166    pub(crate) internal_id_generator: Option<Box<dyn InternalIdGenerator>>,
167    schema_provider: Option<Box<dyn SchemaProvider>>,
168    generated_schema_bootstraps: Vec<GeneratedSchemaBootstrap>,
169    language: Language,
170    i18n_catalog: Arc<crate::I18nCatalog>,
171    typed_resources: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
172    named_resources: BTreeMap<String, Box<dyn Any + Send + Sync>>,
173    locals: BTreeMap<String, Value>,
174    pub(crate) initial_graphs: Vec<GraphNode>,
175    pub(crate) root_graphs: Vec<GraphNode>,
176    entity_runtime_state: EntityRuntimeState,
177    sql_log_options: SqlLogOptions,
178    sql_log_entries: Mutex<Vec<SqlLogEntry>>,
179    user_identifier: Option<String>,
180    timezone: Option<String>,
181    trace_id: String,
182    continuous_page_cursor_store: std::sync::Arc<dyn ContinuousPageCursorStore>,
183    continuous_page_observation: Mutex<(String, Option<String>)>,
184    id_set_store: Arc<dyn IdSetStore>,
185    id_set_observation: Mutex<(String, Option<u64>)>,
186    local_lock_owner: u64,
187    remote_lock_owner: String,
188    runtime_telemetry: Arc<dyn crate::RuntimeTelemetry>,
189    last_fix_evidence: Mutex<Vec<FixEvidence>>,
190}
191
192impl Default for UserContext {
193    fn default() -> Self {
194        let pid = std::process::id();
195        let thread_id_str = format!("{:?}", std::thread::current().id());
196        let numeric_thread_id = thread_id_str
197            .strip_prefix("ThreadId(")
198            .and_then(|s| s.strip_suffix(")"))
199            .unwrap_or(&thread_id_str);
200        let os_user = std::env::var("USER")
201            .or_else(|_| std::env::var("USERNAME"))
202            .unwrap_or_else(|_| "main".to_owned());
203        let user_id = format!("{os_user}@pid-{pid}.tid-{numeric_thread_id}");
204        let owner_sequence = next_local_lock_owner();
205        Self {
206            active_root: OnceLock::new(),
207            metadata: None,
208            entity_registry: None,
209            entity_graph_decoders: InMemoryEntityGraphDecoderRegistry::default(),
210            entity_data_service_behavior_registry: None,
211            request_policy: None,
212            checker_registry: None,
213            event_sink: None,
214            custom_event_sink: None,
215            internal_id_generator: None,
216            schema_provider: None,
217            language: Language::default(),
218            i18n_catalog: crate::I18nCatalog::builtin().clone(),
219            typed_resources: HashMap::new(),
220            named_resources: BTreeMap::new(),
221            locals: BTreeMap::new(),
222            initial_graphs: Vec::new(),
223            root_graphs: Vec::new(),
224            generated_schema_bootstraps: Vec::new(),
225            entity_runtime_state: EntityRuntimeState::default(),
226            // Query and Mutation diagnostic SQL logging are on by default.
227            // They remain independently controllable because performance and
228            // retention policy may differ between the two operation families.
229            sql_log_options: SqlLogOptions::default(),
230            sql_log_entries: Mutex::new(Vec::new()),
231            user_identifier: Some(user_id),
232            timezone: Some("UTC".to_owned()),
233            trace_id: format!(
234                "req-{pid}-{numeric_thread_id}-{:x}",
235                std::time::SystemTime::now()
236                    .duration_since(std::time::UNIX_EPOCH)
237                    .unwrap_or_default()
238                    .as_micros()
239            ),
240            continuous_page_cursor_store: std::sync::Arc::new(
241                InMemoryContinuousPageCursorStore::default(),
242            ),
243            continuous_page_observation: Mutex::new(("DISABLED".to_owned(), None)),
244            id_set_store: Arc::new(InMemoryIdSetStore::default()),
245            id_set_observation: Mutex::new(("ID_SET_DISABLED".to_owned(), None)),
246            local_lock_owner: owner_sequence,
247            remote_lock_owner: format!(
248                "teaql:{pid}:{owner_sequence}:{}",
249                SystemTime::now()
250                    .duration_since(SystemTime::UNIX_EPOCH)
251                    .unwrap_or_default()
252                    .as_nanos()
253            ),
254            runtime_telemetry: Arc::new(crate::NoopRuntimeTelemetry),
255            last_fix_evidence: Mutex::new(Vec::new()),
256        }
257    }
258}
259
260#[async_trait::async_trait]
261pub trait DataStore: Send + Sync + 'static {
262    async fn get(&self, key: &str) -> Option<Value>;
263    async fn put(&self, key: &str, value: Value, timeout_seconds: Option<u64>);
264    async fn remove(&self, key: &str);
265}
266
267#[derive(Default)]
268pub struct InMemoryDataStore {
269    cache: std::sync::RwLock<HashMap<String, (Value, Option<std::time::Instant>)>>,
270}
271
272#[async_trait::async_trait]
273impl DataStore for InMemoryDataStore {
274    async fn get(&self, key: &str) -> Option<Value> {
275        let lock = self.cache.read().unwrap();
276        if let Some((val, expires_at)) = lock.get(key) {
277            if let Some(exp) = expires_at {
278                if std::time::Instant::now() > *exp {
279                    return None;
280                }
281            }
282            return Some(val.clone());
283        }
284        None
285    }
286
287    async fn put(&self, key: &str, value: Value, timeout_seconds: Option<u64>) {
288        let mut lock = self.cache.write().unwrap();
289        let expires_at = timeout_seconds
290            .map(|secs| std::time::Instant::now() + std::time::Duration::from_secs(secs));
291        lock.insert(key.to_string(), (value, expires_at));
292    }
293
294    async fn remove(&self, key: &str) {
295        let mut lock = self.cache.write().unwrap();
296        lock.remove(key);
297    }
298}
299
300impl UserContext {
301    pub fn new() -> Self {
302        Self::default()
303    }
304
305    pub fn with_active_root(mut self, entity_type: impl Into<String>, id: u64) -> Self {
306        let entity_type = crate::canonical_id_space_entity(&entity_type.into());
307        assert!(
308            !entity_type.trim().is_empty(),
309            "active root entity type is required"
310        );
311        assert!(id > 0, "active root id must be positive");
312        self.active_root
313            .set(ContextEntityRef { entity_type, id })
314            .expect("active root may only be assigned once");
315        self
316    }
317
318    #[doc(hidden)]
319    pub fn set_generated_bootstrap_active_root(
320        &self,
321        entity_type: impl Into<String>,
322        id: u64,
323    ) -> Result<(), RuntimeError> {
324        let entity_type = crate::canonical_id_space_entity(&entity_type.into());
325        if entity_type.trim().is_empty() || id == 0 {
326            return Err(RuntimeError::Schema(
327                "invalid generated active root".to_owned(),
328            ));
329        }
330        match self.active_root.get() {
331            Some(existing) if existing.entity_type == entity_type && existing.id == id => Ok(()),
332            Some(existing) => Err(RuntimeError::Schema(format!(
333                "active root already set to {}:{}",
334                existing.entity_type, existing.id
335            ))),
336            None => self
337                .active_root
338                .set(ContextEntityRef { entity_type, id })
339                .map_err(|_| RuntimeError::Schema("active root initialization raced".to_owned())),
340        }
341    }
342
343    pub fn require_active_root(
344        &self,
345        expected_entity_type: &str,
346    ) -> Result<&ContextEntityRef, ContextRootError> {
347        let canonical_expected = crate::canonical_id_space_entity(expected_entity_type);
348        match self.active_root.get() {
349            Some(root) if root.entity_type == canonical_expected => Ok(root),
350            actual_root => Err(ContextRootError {
351                expected_entity_type: canonical_expected,
352                actual_root: actual_root.cloned(),
353            }),
354        }
355    }
356
357    pub(crate) fn active_root_ref(&self) -> Option<&ContextEntityRef> {
358        self.active_root.get()
359    }
360
361    pub fn with_runtime_telemetry(mut self, telemetry: Arc<dyn crate::RuntimeTelemetry>) -> Self {
362        self.runtime_telemetry = telemetry;
363        self
364    }
365
366    pub fn set_runtime_telemetry(&mut self, telemetry: Arc<dyn crate::RuntimeTelemetry>) {
367        self.runtime_telemetry = telemetry;
368    }
369
370    pub fn runtime_telemetry(&self) -> &Arc<dyn crate::RuntimeTelemetry> {
371        &self.runtime_telemetry
372    }
373
374    pub(crate) fn runtime_telemetry_is_noop(&self) -> bool {
375        self.runtime_telemetry.is_noop()
376    }
377
378    pub fn start_runtime_operation(
379        &self,
380        operation: crate::RuntimeOperation,
381    ) -> crate::FailOpenRuntimeTelemetryScope {
382        crate::start_runtime_operation(&self.runtime_telemetry, operation)
383    }
384
385    pub fn user_identifier(&self) -> Option<&str> {
386        self.user_identifier.as_deref()
387    }
388
389    pub fn set_user_identifier(&mut self, user_identifier: impl Into<String>) {
390        self.user_identifier = Some(user_identifier.into());
391    }
392
393    pub fn with_user_identifier(mut self, user_identifier: impl Into<String>) -> Self {
394        self.user_identifier = Some(user_identifier.into());
395        self
396    }
397
398    pub fn set_user_identifier_option(&mut self, user_identifier: Option<String>) {
399        self.user_identifier = user_identifier;
400    }
401
402    pub fn with_user_identifier_option(mut self, user_identifier: Option<String>) -> Self {
403        self.user_identifier = user_identifier;
404        self
405    }
406
407    pub fn timezone(&self) -> Option<&str> {
408        self.timezone.as_deref()
409    }
410
411    pub fn set_timezone(&mut self, timezone: impl Into<String>) {
412        self.timezone = Some(timezone.into());
413    }
414
415    pub fn with_timezone(mut self, timezone: impl Into<String>) -> Self {
416        self.timezone = Some(timezone.into());
417        self
418    }
419
420    pub fn trace_id(&self) -> &str {
421        &self.trace_id
422    }
423
424    pub fn set_trace_id(&mut self, trace_id: impl Into<String>) {
425        self.trace_id = trace_id.into();
426    }
427
428    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
429        self.trace_id = trace_id.into();
430        self
431    }
432
433    pub fn with_module(mut self, module: crate::RuntimeModule) -> Self {
434        module.apply_to(&mut self);
435        self
436    }
437
438    pub fn entity_runtime_state(&self) -> EntityRuntimeState {
439        // UserContext owns only the immutable identity-graph anchor. Every query/new-entity
440        // operation receives fresh mutation state, even when the same context is reused.
441        EntityRuntimeState::fresh_with_shared_graph(&self.entity_runtime_state)
442    }
443
444    pub fn initial_graphs(&self) -> &[GraphNode] {
445        &self.initial_graphs
446    }
447
448    pub fn set_initial_graphs(&mut self, graphs: Vec<GraphNode>) {
449        self.initial_graphs = graphs;
450    }
451
452    pub fn root_graphs(&self) -> &[GraphNode] {
453        &self.root_graphs
454    }
455
456    pub fn set_root_graphs(&mut self, graphs: Vec<GraphNode>) {
457        self.root_graphs = graphs;
458    }
459
460    pub fn with_metadata(mut self, metadata: impl MetadataStore + 'static) -> Self {
461        self.metadata = Some(Box::new(metadata));
462        self
463    }
464
465    pub fn set_metadata(&mut self, metadata: impl MetadataStore + 'static) {
466        self.metadata = Some(Box::new(metadata));
467    }
468
469    pub fn with_entity_registry(mut self, registry: impl EntityRegistry + 'static) -> Self {
470        self.entity_registry = Some(Box::new(registry));
471        self
472    }
473
474    pub fn set_entity_registry(&mut self, registry: impl EntityRegistry + 'static) {
475        self.entity_registry = Some(Box::new(registry));
476    }
477
478    pub fn set_entity_graph_decoder_registry(
479        &mut self,
480        registry: InMemoryEntityGraphDecoderRegistry,
481    ) {
482        self.entity_graph_decoders = registry;
483    }
484
485    pub(crate) fn has_entity_graph_decoder(&self, entity: &str) -> bool {
486        self.entity_graph_decoders.contains(entity)
487    }
488
489    pub(crate) fn decode_compact_entity_into_graph(
490        &self,
491        entity: &str,
492        row: teaql_core::CompactRow,
493        root: &EntityRuntimeState,
494        graph: &mut EntityGraphBuilder,
495    ) -> Result<(), teaql_core::EntityError> {
496        self.entity_graph_decoders
497            .decode_compact(entity, row, root, graph)
498    }
499
500    pub(crate) fn decode_compact_entity_list_into_graph(
501        &self,
502        entity: &str,
503        rows: Vec<teaql_core::CompactRow>,
504        root: &EntityRuntimeState,
505        graph: &mut EntityGraphBuilder,
506        owner_entity: &str,
507        owner_id: u64,
508        relation: &str,
509    ) -> Result<(), teaql_core::EntityError> {
510        self.entity_graph_decoders.decode_compact_list(
511            entity,
512            rows,
513            root,
514            graph,
515            owner_entity,
516            owner_id,
517            relation,
518        )
519    }
520
521    pub(crate) fn decode_compact_entity_batch_into_graph(
522        &self,
523        entity: &str,
524        rows: Vec<teaql_core::CompactRow>,
525        root: &EntityRuntimeState,
526        graph: &mut EntityGraphBuilder,
527    ) -> Result<(), teaql_core::EntityError> {
528        self.entity_graph_decoders
529            .decode_compact_batch(entity, rows, root, graph)
530    }
531
532    pub(crate) fn decode_compact_entity_option_into_graph(
533        &self,
534        entity: &str,
535        rows: Vec<teaql_core::CompactRow>,
536        root: &EntityRuntimeState,
537        graph: &mut EntityGraphBuilder,
538        owner_entity: &str,
539        owner_id: u64,
540        relation: &str,
541    ) -> Result<(), teaql_core::EntityError> {
542        self.entity_graph_decoders.decode_compact_option(
543            entity,
544            rows,
545            root,
546            graph,
547            owner_entity,
548            owner_id,
549            relation,
550        )
551    }
552
553    pub fn with_entity_data_service_behavior_registry(
554        mut self,
555        registry: impl EntityDataServiceBehaviorRegistry + 'static,
556    ) -> Self {
557        self.entity_data_service_behavior_registry = Some(Box::new(registry));
558        self
559    }
560
561    pub fn set_entity_data_service_behavior_registry(
562        &mut self,
563        registry: impl EntityDataServiceBehaviorRegistry + 'static,
564    ) {
565        self.entity_data_service_behavior_registry = Some(Box::new(registry));
566    }
567
568    pub fn with_request_policy(mut self, policy: impl RequestPolicy + 'static) -> Self {
569        self.request_policy = Some(Box::new(policy));
570        self
571    }
572
573    pub fn set_request_policy(&mut self, policy: impl RequestPolicy + 'static) {
574        self.request_policy = Some(Box::new(policy));
575    }
576
577    pub fn clear_request_policy(&mut self) {
578        self.request_policy = None;
579    }
580
581    pub fn with_checker_registry(mut self, registry: impl CheckerRegistry + 'static) -> Self {
582        self.checker_registry = Some(Box::new(registry));
583        self
584    }
585
586    pub fn set_checker_registry(&mut self, registry: impl CheckerRegistry + 'static) {
587        self.checker_registry = Some(Box::new(registry));
588    }
589
590    pub(crate) fn with_event_sink(mut self, sink: impl RawAuditEventSink + 'static) -> Self {
591        self.event_sink = Some(Box::new(sink));
592        self
593    }
594
595    pub(crate) fn set_event_sink(&mut self, sink: impl RawAuditEventSink + 'static) {
596        self.event_sink = Some(Box::new(sink));
597    }
598
599    pub fn with_custom_event_sink(
600        mut self,
601        sink: impl crate::SafeAuditEventSink + 'static,
602    ) -> Self {
603        self.custom_event_sink = Some(Box::new(sink));
604        self
605    }
606
607    pub fn set_custom_event_sink(&mut self, sink: impl crate::SafeAuditEventSink + 'static) {
608        self.custom_event_sink = Some(Box::new(sink));
609    }
610
611    pub fn with_internal_id_generator(
612        mut self,
613        generator: impl InternalIdGenerator + 'static,
614    ) -> Self {
615        self.internal_id_generator = Some(Box::new(generator));
616        self
617    }
618
619    pub fn set_internal_id_generator(&mut self, generator: impl InternalIdGenerator + 'static) {
620        self.internal_id_generator = Some(Box::new(generator));
621    }
622
623    pub fn with_schema_provider(mut self, provider: impl SchemaProvider + 'static) -> Self {
624        self.schema_provider = Some(Box::new(provider));
625        self
626    }
627
628    pub fn set_schema_provider(&mut self, provider: impl SchemaProvider + 'static) {
629        self.schema_provider = Some(Box::new(provider));
630    }
631
632    pub async fn ensure_schema(&self) -> Result<(), RuntimeError> {
633        let provider = self
634            .schema_provider
635            .as_ref()
636            .ok_or_else(|| RuntimeError::Schema("missing schema provider".to_owned()))?;
637        let invocation = SchemaInvocation { _context_owned: () };
638        provider.ensure_schema(self, &invocation).await?;
639        GENERATED_SCHEMA_BOOTSTRAP_MODE
640            .scope((), async {
641                for bootstrap in &self.generated_schema_bootstraps {
642                    bootstrap(self).await?;
643                }
644                Ok::<(), RuntimeError>(())
645            })
646            .await?;
647        Ok(())
648    }
649
650    pub(crate) fn is_generated_schema_bootstrap(&self) -> bool {
651        GENERATED_SCHEMA_BOOTSTRAP_MODE
652            .try_with(|_| true)
653            .unwrap_or(false)
654    }
655
656    pub(crate) fn set_generated_schema_bootstraps(
657        &mut self,
658        bootstraps: Vec<GeneratedSchemaBootstrap>,
659    ) {
660        self.generated_schema_bootstraps = bootstraps;
661    }
662
663    #[doc(hidden)]
664    pub fn initialize_generated_bootstrap_entity<E: teaql_core::Entity>(
665        &self,
666        entity: &mut E,
667        entity_name: &str,
668        fixed_id: u64,
669    ) -> Result<(), RuntimeError> {
670        let generator = self.internal_id_generator.as_ref().ok_or_else(|| {
671            RuntimeError::IdGeneration("missing internal ID generator".to_owned())
672        })?;
673        generator.ensure_floor(entity_name, fixed_id)?;
674        entity.mark_as_new();
675        Ok(())
676    }
677
678    pub fn with_language(mut self, language: Language) -> Self {
679        self.language = language;
680        self
681    }
682
683    pub fn set_language(&mut self, language: Language) {
684        self.language = language;
685    }
686
687    pub fn with_i18n_catalog(mut self, catalog: Arc<crate::I18nCatalog>) -> Self {
688        self.i18n_catalog = catalog;
689        self
690    }
691
692    pub fn set_i18n_catalog(&mut self, catalog: Arc<crate::I18nCatalog>) {
693        self.i18n_catalog = catalog;
694    }
695
696    pub fn language(&self) -> Language {
697        self.language
698    }
699
700    pub fn set_language_code(&mut self, code: &str) -> Result<(), RuntimeError> {
701        let Some(language) = Language::from_code(code) else {
702            return Err(RuntimeError::UnsupportedLocale(code.to_owned()));
703        };
704        self.language = language;
705        Ok(())
706    }
707
708    pub fn set_locale_code(&mut self, code: &str) -> Result<(), RuntimeError> {
709        self.set_language_code(code)
710    }
711
712    pub fn generate_id(&self, entity: &str) -> Result<Option<u64>, RuntimeError> {
713        self.internal_id_generator
714            .as_ref()
715            .map(|generator| generator.generate_id(entity))
716            .transpose()
717    }
718
719    pub fn next_id(&self, entity: &str) -> Result<u64, RuntimeError> {
720        match self.generate_id(entity)? {
721            Some(id) => Ok(id),
722            None => local_id_generator().generate_id(entity),
723        }
724    }
725
726    pub fn entity(&self, name: &str) -> Option<&EntityDescriptor> {
727        self.metadata
728            .as_ref()
729            .and_then(|metadata| metadata.entity(name))
730    }
731
732    pub fn all_entities(&self) -> Vec<&EntityDescriptor> {
733        self.metadata
734            .as_ref()
735            .map(|metadata| metadata.all_entities())
736            .unwrap_or_default()
737    }
738
739    pub fn require_entity(&self, name: &str) -> Result<&EntityDescriptor, RuntimeError> {
740        self.entity(name)
741            .ok_or_else(|| RuntimeError::MissingEntity(name.to_owned()))
742    }
743
744    pub fn insert_resource<T>(&mut self, resource: T)
745    where
746        T: Send + Sync + 'static,
747    {
748        self.typed_resources
749            .insert(TypeId::of::<T>(), Box::new(resource));
750    }
751
752    pub fn get_resource<T>(&self) -> Option<&T>
753    where
754        T: Send + Sync + 'static,
755    {
756        self.typed_resources
757            .get(&TypeId::of::<T>())
758            .and_then(|value| value.downcast_ref::<T>())
759    }
760
761    pub fn require_resource<T>(&self) -> Result<&T, ContextError>
762    where
763        T: Send + Sync + 'static,
764    {
765        self.get_resource::<T>()
766            .ok_or(ContextError::MissingTypedResource(
767                std::any::type_name::<T>(),
768            ))
769    }
770
771    pub fn insert_named_resource<T>(&mut self, name: impl Into<String>, resource: T)
772    where
773        T: Send + Sync + 'static,
774    {
775        self.named_resources.insert(name.into(), Box::new(resource));
776    }
777
778    pub fn get_named_resource<T>(&self, name: &str) -> Option<&T>
779    where
780        T: Send + Sync + 'static,
781    {
782        self.named_resources
783            .get(name)
784            .and_then(|value| value.downcast_ref::<T>())
785    }
786
787    pub fn require_named_resource<T>(&self, name: &str) -> Result<&T, ContextError>
788    where
789        T: Send + Sync + 'static,
790    {
791        self.get_named_resource::<T>(name)
792            .ok_or_else(|| ContextError::MissingResource(name.to_owned()))
793    }
794
795    pub fn put_local(&mut self, key: impl Into<String>, value: impl Into<Value>) {
796        self.locals.insert(key.into(), value.into());
797    }
798
799    pub fn local(&self, key: &str) -> Option<&Value> {
800        self.locals.get(key)
801    }
802
803    pub fn remove_local(&mut self, key: &str) -> Option<Value> {
804        self.locals.remove(key)
805    }
806
807    pub fn has_entity_data_service(&self, entity: &str) -> bool {
808        let in_registry = self
809            .entity_registry
810            .as_ref()
811            .map(|registry| registry.contains(entity))
812            .unwrap_or(false);
813        in_registry || self.entity(entity).is_some()
814    }
815
816    pub fn entity_data_service_behavior(
817        &self,
818        entity: &str,
819    ) -> Option<std::sync::Arc<dyn EntityDataServiceBehavior>> {
820        self.entity_data_service_behavior_registry
821            .as_ref()
822            .and_then(|registry| registry.behavior(entity))
823    }
824
825    pub fn has_checker(&self, entity: &str) -> bool {
826        self.checker_registry
827            .as_ref()
828            .and_then(|registry| registry.checker(entity))
829            .is_some()
830    }
831
832    /// One deterministic clock value for every Fix executed by the current
833    /// graph save. The task-local scope keeps concurrent saves on one context
834    /// isolated; standalone checker calls receive their own current value.
835    pub fn fix_time(&self) -> teaql_core::time::Timestamp {
836        crate::entity_save::current_graph_fix_time()
837    }
838
839    pub fn record_fix_evidence(&self, evidence: FixEvidence) {
840        crate::entity_save::record_graph_fix_evidence(evidence);
841    }
842
843    pub(crate) fn replace_last_fix_evidence(&self, evidence: Vec<FixEvidence>) {
844        *self.last_fix_evidence.lock().unwrap() = evidence;
845    }
846
847    pub fn last_fix_evidence(&self) -> Vec<FixEvidence> {
848        self.last_fix_evidence.lock().unwrap().clone()
849    }
850
851    pub fn check_and_fix_values(
852        &self,
853        entity: &str,
854        values: &mut crate::EntityValues,
855    ) -> Result<(), RuntimeError> {
856        self.check_and_fix_values_at(entity, values, &ObjectLocation::root())
857    }
858
859    pub fn check_and_fix_values_at(
860        &self,
861        entity: &str,
862        values: &mut crate::EntityValues,
863        location: &ObjectLocation,
864    ) -> Result<(), RuntimeError> {
865        let status = CheckObjectStatus::from_values(values);
866        let checker = self
867            .checker_registry
868            .as_ref()
869            .and_then(|registry| registry.checker(entity));
870        let mut results = CheckResults::new();
871        if let Some(checker) = checker {
872            checker.check_and_fix(self, values, location, &mut results);
873        }
874
875        // Keep runtime validation aligned with the schema generated from the
876        // same metadata. Custom checkers get the first chance to supply or fix
877        // a value; afterwards every NOT NULL property must be present on a
878        // create, and an update must not explicitly clear one.
879        if let Some(descriptor) = self
880            .metadata
881            .as_ref()
882            .and_then(|metadata| metadata.entity(entity))
883        {
884            for property in descriptor
885                .properties
886                .iter()
887                // The optimistic-lock version is runtime-managed. Insert
888                // preparation assigns its initial value after check/fix, so a
889                // create caller must never be required to provide it.
890                .filter(|property| !property.nullable && !property.is_version)
891            {
892                let missing = !values.contains_key(&property.name);
893                let null = matches!(values.get(&property.name), Some(Value::Null));
894                let property_location = location.clone().member(&property.name);
895                let already_reported = results.iter().any(|result| {
896                    result.rule == crate::CheckRule::Required
897                        && result.location == property_location
898                });
899                if ((status.is_create() && missing) || null) && !already_reported {
900                    results.push(CheckResult::required(property_location));
901                }
902            }
903        }
904        if results.is_empty() {
905            return Ok(());
906        }
907        self.translate_check_results(&mut results);
908        Err(RuntimeError::Check(results))
909    }
910
911    pub fn translate_check_results(&self, results: &mut CheckResults) {
912        for result in results {
913            if result.message.is_none() {
914                result.message = Some(
915                    self.i18n_catalog
916                        .translate_check_result(self.language, result),
917                );
918            }
919        }
920    }
921
922    pub fn send_event(&self, mut event: RawAuditEvent) -> Result<(), RuntimeError> {
923        if self.is_generated_schema_bootstrap()
924            && matches!(
925                event.kind,
926                crate::RawAuditEventKind::Created | crate::RawAuditEventKind::Updated
927            )
928        {
929            let reason = event
930                .trace_chain
931                .last()
932                .map(|node| node.comment.clone())
933                .unwrap_or_else(|| "generated runtime bootstrap".to_owned());
934            let resulting_version = event
935                .new_values
936                .as_ref()
937                .and_then(|values| values.get("version"))
938                .or_else(|| event.values.get("version"))
939                .and_then(teaql_core::Value::try_i64);
940            let occurred_at_millis = std::time::SystemTime::now()
941                .duration_since(std::time::UNIX_EPOCH)
942                .unwrap_or_default()
943                .as_millis() as u64;
944            event.bootstrap_audit = Some(crate::BootstrapAuditIdentity {
945                actor: "teaql-generated-bootstrap".to_owned(),
946                category: "runtime-bootstrap".to_owned(),
947                reason,
948                resulting_version,
949                occurred_at_millis,
950            });
951        }
952        let scope = self.start_runtime_operation(
953            crate::RuntimeOperation::new("audit", format!("{}.event", event.entity))
954                .attribute("teaql.entity.type", event.entity.clone()),
955        );
956        let result = self.send_event_inner(event);
957        match &result {
958            Ok(()) => scope.success(std::collections::BTreeMap::new()),
959            Err(_) => scope.failure("audit_error"),
960        }
961        result
962    }
963
964    fn send_event_inner(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
965        if let Some(sink) = self.event_sink.as_ref() {
966            sink.on_event(self, &event)?;
967        }
968        if let Some(sink) = self.custom_event_sink.as_ref() {
969            let (mask_fields, max_len) = self
970                .metadata
971                .as_ref()
972                .and_then(|metadata| metadata.entity(&event.entity))
973                .map(|desc| (desc.audit_mask_fields.clone(), desc.audit_value_max_len))
974                .unwrap_or_else(|| (vec![], None));
975
976            let safe_event = event.build_safe_event(&mask_fields, max_len);
977            sink.on_safe_event(self, &safe_event)?;
978        }
979
980        crate::log_formatter::LogManager::write_audit_log(&event);
981
982        Ok(())
983    }
984
985    pub async fn get_in_store(&self, key: &str) -> Option<Value> {
986        let store = self.get_resource::<Box<dyn DataStore>>()?;
987        store.get(key).await
988    }
989
990    pub async fn put_in_store(
991        &self,
992        key: &str,
993        value: impl Into<Value>,
994        timeout_seconds: Option<u64>,
995    ) {
996        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
997            store.put(key, value.into(), timeout_seconds).await;
998        }
999    }
1000
1001    pub async fn clear_in_store(&self, key: &str) {
1002        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
1003            store.remove(key).await;
1004        }
1005    }
1006}
1007
1008#[cfg(test)]
1009mod sql_log_option_tests {
1010    use super::*;
1011
1012    #[test]
1013    fn diagnostic_sql_log_is_enabled_by_default_with_independent_switches() {
1014        let mut context = UserContext::default();
1015        assert_eq!(context.sql_log_options(), SqlLogOptions::all());
1016        assert!(context.sql_logs().is_empty());
1017
1018        context.disable_select_sql_log();
1019        assert_eq!(context.sql_log_options(), SqlLogOptions::mutation_only());
1020
1021        context.enable_select_sql_log();
1022        context.disable_mutation_sql_log();
1023        assert_eq!(context.sql_log_options(), SqlLogOptions::select_only());
1024
1025        context.disable_sql_log();
1026        assert_eq!(context.sql_log_options(), SqlLogOptions::disabled());
1027    }
1028
1029    #[test]
1030    fn disabled_sql_log_rejects_executor_metadata_before_recording() {
1031        let mut context = UserContext::default();
1032        context.disable_sql_log();
1033        let now = SystemTime::now();
1034        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1035            backend: "sql".to_owned(),
1036            operation: teaql_data_service::DataServiceOperation::Query,
1037            started_at: now,
1038            ended_at: now,
1039            affected_rows: None,
1040            result_count: Some(1),
1041            trace_chain: Vec::new(),
1042            comment: Some("disabled log test".to_owned()),
1043            backend_request_id: None,
1044            parameterized_query: Some("SELECT id FROM sample WHERE id = $1".to_owned()),
1045            params: vec![Value::I64(1)],
1046            debug_query: Some("SELECT id FROM sample WHERE id = 1".to_owned()),
1047        });
1048        assert!(context.sql_logs().is_empty());
1049    }
1050
1051    #[test]
1052    fn metadata_log_retains_structured_intent_sql_forms_and_multilevel_trace() {
1053        let context = UserContext::default();
1054        let now = SystemTime::now();
1055        let query_trace = vec![
1056            teaql_core::TraceNode::typed(
1057                teaql_core::TraceKind::Comment,
1058                "School",
1059                None,
1060                "what: load school graph",
1061            ),
1062            teaql_core::TraceNode::typed(
1063                teaql_core::TraceKind::Purpose,
1064                "School",
1065                None,
1066                "why: render school details",
1067            ),
1068            teaql_core::TraceNode::typed(
1069                teaql_core::TraceKind::Relation,
1070                "platform",
1071                None,
1072                "School.platform",
1073            ),
1074            teaql_core::TraceNode::typed(
1075                teaql_core::TraceKind::Relation,
1076                "organization",
1077                None,
1078                "Platform.organization",
1079            ),
1080            teaql_core::TraceNode::typed(
1081                teaql_core::TraceKind::Relation,
1082                "region",
1083                None,
1084                "Organization.region",
1085            ),
1086        ];
1087        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1088            backend: "sqlite".to_owned(),
1089            operation: teaql_data_service::DataServiceOperation::Query,
1090            started_at: now,
1091            ended_at: now,
1092            affected_rows: None,
1093            result_count: Some(2),
1094            trace_chain: query_trace.clone(),
1095            comment: None,
1096            backend_request_id: None,
1097            parameterized_query: Some("SELECT name FROM school_data WHERE id = ?".to_owned()),
1098            params: vec![Value::I64(7)],
1099            debug_query: Some("SELECT name FROM school_data WHERE id = 7".to_owned()),
1100        });
1101
1102        let query = context.sql_logs().pop().expect("query log");
1103        assert_eq!(query.comment.as_deref(), Some("what: load school graph"));
1104        assert_eq!(query.purpose.as_deref(), Some("why: render school details"));
1105        assert_eq!(query.audit_reason, None);
1106        assert_eq!(
1107            query
1108                .trace_path
1109                .iter()
1110                .map(|node| node.kind)
1111                .collect::<Vec<_>>(),
1112            vec![
1113                teaql_core::TraceKind::Operation,
1114                teaql_core::TraceKind::Request,
1115                teaql_core::TraceKind::Relation,
1116                teaql_core::TraceKind::Relation,
1117                teaql_core::TraceKind::Relation,
1118                teaql_core::TraceKind::Provider,
1119                teaql_core::TraceKind::Sql,
1120            ]
1121        );
1122        assert_eq!(query.trace_path[0].entity_type, "School");
1123        assert_eq!(query.trace_path[5].entity_type, "sqlite");
1124        assert_eq!(query.trace_path[6].entity_type, "select");
1125        assert_eq!(query.sql, "SELECT name FROM school_data WHERE id = ?");
1126        assert_eq!(query.params, vec![Value::I64(7)]);
1127        assert_eq!(query.debug_sql, "SELECT name FROM school_data WHERE id = 7");
1128        assert_eq!(query.result_count, Some(2));
1129
1130        let mutation_trace = vec![teaql_core::TraceNode::typed(
1131            teaql_core::TraceKind::AuditReason,
1132            "School",
1133            Some(7),
1134            "correct school name",
1135        )];
1136        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1137            backend: "sqlite".to_owned(),
1138            operation: teaql_data_service::DataServiceOperation::Update,
1139            started_at: now,
1140            ended_at: now,
1141            affected_rows: Some(1),
1142            result_count: None,
1143            trace_chain: mutation_trace.clone(),
1144            comment: None,
1145            backend_request_id: None,
1146            parameterized_query: Some("UPDATE school_data SET name = ? WHERE id = ?".to_owned()),
1147            params: vec![Value::from("Academy"), Value::I64(7)],
1148            debug_query: Some("UPDATE school_data SET name = 'Academy' WHERE id = 7".to_owned()),
1149        });
1150        let mutation = context.sql_logs().pop().expect("mutation log");
1151        assert_eq!(mutation.comment, None);
1152        assert_eq!(mutation.purpose, None);
1153        assert_eq!(
1154            mutation.audit_reason.as_deref(),
1155            Some("correct school name")
1156        );
1157        assert_eq!(
1158            mutation
1159                .trace_path
1160                .iter()
1161                .map(|node| node.kind)
1162                .collect::<Vec<_>>(),
1163            vec![
1164                teaql_core::TraceKind::Operation,
1165                teaql_core::TraceKind::Entity,
1166                teaql_core::TraceKind::Provider,
1167                teaql_core::TraceKind::Sql,
1168            ]
1169        );
1170        assert_eq!(mutation.affected_rows, Some(1));
1171    }
1172}
1173
1174#[cfg(test)]
1175mod entity_runtime_state_tests {
1176    use super::*;
1177    use crate::EntityKey;
1178
1179    #[test]
1180    fn reused_user_context_returns_independent_mutation_ledgers() {
1181        let context = UserContext::default();
1182        let first = context.entity_runtime_state();
1183        let key = EntityKey::new("School", 1_u64);
1184        first.set(key.clone(), "name", "First");
1185
1186        let second = context.entity_runtime_state();
1187
1188        assert_eq!(first.changed_field_names(&key).len(), 1);
1189        assert!(second.changed_field_names(&key).is_empty());
1190    }
1191}