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                && std::time::Instant::now() > *exp
279            {
280                return None;
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(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    #[allow(clippy::too_many_arguments)] // Mirrors the generated decoder contract.
501    pub(crate) fn decode_compact_entity_list_into_graph(
502        &self,
503        entity: &str,
504        rows: Vec<teaql_core::CompactRow>,
505        root: &EntityRuntimeState,
506        graph: &mut EntityGraphBuilder,
507        owner_entity: &str,
508        owner_id: u64,
509        relation: &str,
510    ) -> Result<(), teaql_core::EntityError> {
511        self.entity_graph_decoders.decode_compact_list(
512            entity,
513            rows,
514            root,
515            graph,
516            owner_entity,
517            owner_id,
518            relation,
519        )
520    }
521
522    pub(crate) fn decode_compact_entity_batch_into_graph(
523        &self,
524        entity: &str,
525        rows: Vec<teaql_core::CompactRow>,
526        root: &EntityRuntimeState,
527        graph: &mut EntityGraphBuilder,
528    ) -> Result<(), teaql_core::EntityError> {
529        self.entity_graph_decoders
530            .decode_compact_batch(entity, rows, root, graph)
531    }
532
533    #[allow(clippy::too_many_arguments)] // Mirrors the generated decoder contract.
534    pub(crate) fn decode_compact_entity_option_into_graph(
535        &self,
536        entity: &str,
537        rows: Vec<teaql_core::CompactRow>,
538        root: &EntityRuntimeState,
539        graph: &mut EntityGraphBuilder,
540        owner_entity: &str,
541        owner_id: u64,
542        relation: &str,
543    ) -> Result<(), teaql_core::EntityError> {
544        self.entity_graph_decoders.decode_compact_option(
545            entity,
546            rows,
547            root,
548            graph,
549            owner_entity,
550            owner_id,
551            relation,
552        )
553    }
554
555    pub fn with_entity_data_service_behavior_registry(
556        mut self,
557        registry: impl EntityDataServiceBehaviorRegistry + 'static,
558    ) -> Self {
559        self.entity_data_service_behavior_registry = Some(Box::new(registry));
560        self
561    }
562
563    pub fn set_entity_data_service_behavior_registry(
564        &mut self,
565        registry: impl EntityDataServiceBehaviorRegistry + 'static,
566    ) {
567        self.entity_data_service_behavior_registry = Some(Box::new(registry));
568    }
569
570    pub fn with_request_policy(mut self, policy: impl RequestPolicy + 'static) -> Self {
571        self.request_policy = Some(Box::new(policy));
572        self
573    }
574
575    pub fn set_request_policy(&mut self, policy: impl RequestPolicy + 'static) {
576        self.request_policy = Some(Box::new(policy));
577    }
578
579    pub fn clear_request_policy(&mut self) {
580        self.request_policy = None;
581    }
582
583    pub fn with_checker_registry(mut self, registry: impl CheckerRegistry + 'static) -> Self {
584        self.checker_registry = Some(Box::new(registry));
585        self
586    }
587
588    pub fn set_checker_registry(&mut self, registry: impl CheckerRegistry + 'static) {
589        self.checker_registry = Some(Box::new(registry));
590    }
591
592    #[cfg(test)]
593    pub(crate) fn with_event_sink(mut self, sink: impl RawAuditEventSink + 'static) -> Self {
594        self.event_sink = Some(Box::new(sink));
595        self
596    }
597
598    pub(crate) fn set_event_sink(&mut self, sink: impl RawAuditEventSink + 'static) {
599        self.event_sink = Some(Box::new(sink));
600    }
601
602    pub fn with_custom_event_sink(
603        mut self,
604        sink: impl crate::SafeAuditEventSink + 'static,
605    ) -> Self {
606        self.custom_event_sink = Some(Box::new(sink));
607        self
608    }
609
610    pub fn set_custom_event_sink(&mut self, sink: impl crate::SafeAuditEventSink + 'static) {
611        self.custom_event_sink = Some(Box::new(sink));
612    }
613
614    pub fn with_internal_id_generator(
615        mut self,
616        generator: impl InternalIdGenerator + 'static,
617    ) -> Self {
618        self.internal_id_generator = Some(Box::new(generator));
619        self
620    }
621
622    pub fn set_internal_id_generator(&mut self, generator: impl InternalIdGenerator + 'static) {
623        self.internal_id_generator = Some(Box::new(generator));
624    }
625
626    pub fn with_schema_provider(mut self, provider: impl SchemaProvider + 'static) -> Self {
627        self.schema_provider = Some(Box::new(provider));
628        self
629    }
630
631    pub fn set_schema_provider(&mut self, provider: impl SchemaProvider + 'static) {
632        self.schema_provider = Some(Box::new(provider));
633    }
634
635    pub async fn ensure_schema(&self) -> Result<(), RuntimeError> {
636        let provider = self
637            .schema_provider
638            .as_ref()
639            .ok_or_else(|| RuntimeError::Schema("missing schema provider".to_owned()))?;
640        let invocation = SchemaInvocation { _context_owned: () };
641        provider.ensure_schema(self, &invocation).await?;
642        GENERATED_SCHEMA_BOOTSTRAP_MODE
643            .scope((), async {
644                for bootstrap in &self.generated_schema_bootstraps {
645                    bootstrap(self).await?;
646                }
647                Ok::<(), RuntimeError>(())
648            })
649            .await?;
650        Ok(())
651    }
652
653    pub(crate) fn is_generated_schema_bootstrap(&self) -> bool {
654        GENERATED_SCHEMA_BOOTSTRAP_MODE
655            .try_with(|_| true)
656            .unwrap_or(false)
657    }
658
659    pub(crate) fn set_generated_schema_bootstraps(
660        &mut self,
661        bootstraps: Vec<GeneratedSchemaBootstrap>,
662    ) {
663        self.generated_schema_bootstraps = bootstraps;
664    }
665
666    #[doc(hidden)]
667    pub fn initialize_generated_bootstrap_entity<E: teaql_core::Entity>(
668        &self,
669        entity: &mut E,
670        entity_name: &str,
671        fixed_id: u64,
672    ) -> Result<(), RuntimeError> {
673        let generator = self.internal_id_generator.as_ref().ok_or_else(|| {
674            RuntimeError::IdGeneration("missing internal ID generator".to_owned())
675        })?;
676        generator.ensure_floor(entity_name, fixed_id)?;
677        entity.mark_as_new();
678        Ok(())
679    }
680
681    pub fn with_language(mut self, language: Language) -> Self {
682        self.language = language;
683        self
684    }
685
686    pub fn set_language(&mut self, language: Language) {
687        self.language = language;
688    }
689
690    pub fn with_i18n_catalog(mut self, catalog: Arc<crate::I18nCatalog>) -> Self {
691        self.i18n_catalog = catalog;
692        self
693    }
694
695    pub fn set_i18n_catalog(&mut self, catalog: Arc<crate::I18nCatalog>) {
696        self.i18n_catalog = catalog;
697    }
698
699    pub fn language(&self) -> Language {
700        self.language
701    }
702
703    pub fn set_language_code(&mut self, code: &str) -> Result<(), RuntimeError> {
704        let Some(language) = Language::from_code(code) else {
705            return Err(RuntimeError::UnsupportedLocale(code.to_owned()));
706        };
707        self.language = language;
708        Ok(())
709    }
710
711    pub fn set_locale_code(&mut self, code: &str) -> Result<(), RuntimeError> {
712        self.set_language_code(code)
713    }
714
715    pub fn generate_id(&self, entity: &str) -> Result<Option<u64>, RuntimeError> {
716        self.internal_id_generator
717            .as_ref()
718            .map(|generator| generator.generate_id(entity))
719            .transpose()
720    }
721
722    pub fn next_id(&self, entity: &str) -> Result<u64, RuntimeError> {
723        match self.generate_id(entity)? {
724            Some(id) => Ok(id),
725            None => local_id_generator().generate_id(entity),
726        }
727    }
728
729    pub fn entity(&self, name: &str) -> Option<&EntityDescriptor> {
730        self.metadata
731            .as_ref()
732            .and_then(|metadata| metadata.entity(name))
733    }
734
735    pub fn all_entities(&self) -> Vec<&EntityDescriptor> {
736        self.metadata
737            .as_ref()
738            .map(|metadata| metadata.all_entities())
739            .unwrap_or_default()
740    }
741
742    pub fn require_entity(&self, name: &str) -> Result<&EntityDescriptor, RuntimeError> {
743        self.entity(name)
744            .ok_or_else(|| RuntimeError::MissingEntity(name.to_owned()))
745    }
746
747    pub fn insert_resource<T>(&mut self, resource: T)
748    where
749        T: Send + Sync + 'static,
750    {
751        self.typed_resources
752            .insert(TypeId::of::<T>(), Box::new(resource));
753    }
754
755    pub fn get_resource<T>(&self) -> Option<&T>
756    where
757        T: Send + Sync + 'static,
758    {
759        self.typed_resources
760            .get(&TypeId::of::<T>())
761            .and_then(|value| value.downcast_ref::<T>())
762    }
763
764    pub fn require_resource<T>(&self) -> Result<&T, ContextError>
765    where
766        T: Send + Sync + 'static,
767    {
768        self.get_resource::<T>()
769            .ok_or(ContextError::MissingTypedResource(
770                std::any::type_name::<T>(),
771            ))
772    }
773
774    pub fn insert_named_resource<T>(&mut self, name: impl Into<String>, resource: T)
775    where
776        T: Send + Sync + 'static,
777    {
778        self.named_resources.insert(name.into(), Box::new(resource));
779    }
780
781    pub fn get_named_resource<T>(&self, name: &str) -> Option<&T>
782    where
783        T: Send + Sync + 'static,
784    {
785        self.named_resources
786            .get(name)
787            .and_then(|value| value.downcast_ref::<T>())
788    }
789
790    pub fn require_named_resource<T>(&self, name: &str) -> Result<&T, ContextError>
791    where
792        T: Send + Sync + 'static,
793    {
794        self.get_named_resource::<T>(name)
795            .ok_or_else(|| ContextError::MissingResource(name.to_owned()))
796    }
797
798    pub fn put_local(&mut self, key: impl Into<String>, value: impl Into<Value>) {
799        self.locals.insert(key.into(), value.into());
800    }
801
802    pub fn local(&self, key: &str) -> Option<&Value> {
803        self.locals.get(key)
804    }
805
806    pub fn remove_local(&mut self, key: &str) -> Option<Value> {
807        self.locals.remove(key)
808    }
809
810    pub fn has_entity_data_service(&self, entity: &str) -> bool {
811        let in_registry = self
812            .entity_registry
813            .as_ref()
814            .map(|registry| registry.contains(entity))
815            .unwrap_or(false);
816        in_registry || self.entity(entity).is_some()
817    }
818
819    pub fn entity_data_service_behavior(
820        &self,
821        entity: &str,
822    ) -> Option<std::sync::Arc<dyn EntityDataServiceBehavior>> {
823        self.entity_data_service_behavior_registry
824            .as_ref()
825            .and_then(|registry| registry.behavior(entity))
826    }
827
828    pub fn has_checker(&self, entity: &str) -> bool {
829        self.checker_registry
830            .as_ref()
831            .and_then(|registry| registry.checker(entity))
832            .is_some()
833    }
834
835    /// One deterministic clock value for every Fix executed by the current
836    /// graph save. The task-local scope keeps concurrent saves on one context
837    /// isolated; standalone checker calls receive their own current value.
838    pub fn fix_time(&self) -> teaql_core::time::Timestamp {
839        crate::entity_save::current_graph_fix_time()
840    }
841
842    pub fn record_fix_evidence(&self, evidence: FixEvidence) {
843        crate::entity_save::record_graph_fix_evidence(evidence);
844    }
845
846    pub(crate) fn replace_last_fix_evidence(&self, evidence: Vec<FixEvidence>) {
847        *self.last_fix_evidence.lock().unwrap() = evidence;
848    }
849
850    pub fn last_fix_evidence(&self) -> Vec<FixEvidence> {
851        self.last_fix_evidence.lock().unwrap().clone()
852    }
853
854    pub fn check_and_fix_values(
855        &self,
856        entity: &str,
857        values: &mut crate::EntityValues,
858    ) -> Result<(), RuntimeError> {
859        self.check_and_fix_values_at(entity, values, &ObjectLocation::root())
860    }
861
862    pub fn check_and_fix_values_at(
863        &self,
864        entity: &str,
865        values: &mut crate::EntityValues,
866        location: &ObjectLocation,
867    ) -> Result<(), RuntimeError> {
868        let status = CheckObjectStatus::from_values(values);
869        let checker = self
870            .checker_registry
871            .as_ref()
872            .and_then(|registry| registry.checker(entity));
873        let mut results = CheckResults::new();
874        if let Some(checker) = checker {
875            checker.check_and_fix(self, values, location, &mut results);
876        }
877
878        // Custom checkers get the first chance to supply or fix a value.
879        self.collect_required_property_results(entity, values, status, location, &mut results);
880        if results.is_empty() {
881            return Ok(());
882        }
883        self.translate_check_results(&mut results);
884        Err(RuntimeError::Check(results))
885    }
886
887    /// Validate the sparse values that a ledger CREATE will actually insert.
888    /// This is intentionally metadata-only: the complete typed graph already
889    /// ran Checker/Fix once, and running it again on sparse changes would
890    /// misclassify unchanged loaded fields and repeat Fix side effects.
891    pub(crate) fn validate_required_create_payload(
892        &self,
893        entity: &str,
894        values: &crate::EntityValues,
895        location: &ObjectLocation,
896    ) -> Result<(), RuntimeError> {
897        let mut results = CheckResults::new();
898        self.collect_required_property_results(
899            entity,
900            values,
901            CheckObjectStatus::Create,
902            location,
903            &mut results,
904        );
905        if results.is_empty() {
906            return Ok(());
907        }
908        self.translate_check_results(&mut results);
909        Err(RuntimeError::Check(results))
910    }
911
912    fn collect_required_property_results(
913        &self,
914        entity: &str,
915        values: &crate::EntityValues,
916        status: CheckObjectStatus,
917        location: &ObjectLocation,
918        results: &mut CheckResults,
919    ) {
920        // Keep runtime validation aligned with the schema generated from the
921        // same metadata. Every NOT NULL property must be present on a create,
922        // and an update must not explicitly clear one.
923        if let Some(descriptor) = self
924            .metadata
925            .as_ref()
926            .and_then(|metadata| metadata.entity(entity))
927        {
928            for property in descriptor
929                .properties
930                .iter()
931                // The optimistic-lock version is runtime-managed. Insert
932                // preparation assigns its initial value after check/fix, so a
933                // create caller must never be required to provide it.
934                .filter(|property| !property.nullable && !property.is_version)
935            {
936                let missing = !values.contains_key(&property.name);
937                let null = matches!(values.get(&property.name), Some(Value::Null));
938                let property_location = location.clone().member(&property.name);
939                let already_reported = results.iter().any(|result| {
940                    result.rule == crate::CheckRule::Required
941                        && result.location == property_location
942                });
943                if ((status.is_create() && missing) || null) && !already_reported {
944                    results.push(CheckResult::required(property_location));
945                }
946            }
947        }
948    }
949
950    pub fn translate_check_results(&self, results: &mut CheckResults) {
951        for result in results {
952            if result.message.is_none() {
953                result.message = Some(
954                    self.i18n_catalog
955                        .translate_check_result(self.language, result),
956                );
957            }
958        }
959    }
960
961    pub fn send_event(&self, mut event: RawAuditEvent) -> Result<(), RuntimeError> {
962        if self.is_generated_schema_bootstrap()
963            && matches!(
964                event.kind,
965                crate::RawAuditEventKind::Created | crate::RawAuditEventKind::Updated
966            )
967        {
968            let reason = event
969                .trace_chain
970                .last()
971                .map(|node| node.comment.clone())
972                .unwrap_or_else(|| "generated runtime bootstrap".to_owned());
973            let resulting_version = event
974                .new_values
975                .as_ref()
976                .and_then(|values| values.get("version"))
977                .or_else(|| event.values.get("version"))
978                .and_then(teaql_core::Value::try_i64);
979            let occurred_at_millis = std::time::SystemTime::now()
980                .duration_since(std::time::UNIX_EPOCH)
981                .unwrap_or_default()
982                .as_millis() as u64;
983            event.bootstrap_audit = Some(crate::BootstrapAuditIdentity {
984                actor: "teaql-generated-bootstrap".to_owned(),
985                category: "runtime-bootstrap".to_owned(),
986                reason,
987                resulting_version,
988                occurred_at_millis,
989            });
990        }
991        let scope = self.start_runtime_operation(
992            crate::RuntimeOperation::new("audit", format!("{}.event", event.entity))
993                .attribute("teaql.entity.type", event.entity.clone()),
994        );
995        let result = self.send_event_inner(event);
996        match &result {
997            Ok(()) => scope.success(std::collections::BTreeMap::new()),
998            Err(_) => scope.failure("audit_error"),
999        }
1000        result
1001    }
1002
1003    fn send_event_inner(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
1004        if let Some(sink) = self.event_sink.as_ref() {
1005            sink.on_event(self, &event)?;
1006        }
1007        if let Some(sink) = self.custom_event_sink.as_ref() {
1008            let (mask_fields, max_len) = self
1009                .metadata
1010                .as_ref()
1011                .and_then(|metadata| metadata.entity(&event.entity))
1012                .map(|desc| (desc.audit_mask_fields.clone(), desc.audit_value_max_len))
1013                .unwrap_or_else(|| (vec![], None));
1014
1015            let safe_event = event.build_safe_event(&mask_fields, max_len);
1016            sink.on_safe_event(self, &safe_event)?;
1017        }
1018
1019        crate::log_formatter::LogManager::write_audit_log(&event);
1020
1021        Ok(())
1022    }
1023
1024    pub async fn get_in_store(&self, key: &str) -> Option<Value> {
1025        let store = self.get_resource::<Box<dyn DataStore>>()?;
1026        store.get(key).await
1027    }
1028
1029    pub async fn put_in_store(
1030        &self,
1031        key: &str,
1032        value: impl Into<Value>,
1033        timeout_seconds: Option<u64>,
1034    ) {
1035        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
1036            store.put(key, value.into(), timeout_seconds).await;
1037        }
1038    }
1039
1040    pub async fn clear_in_store(&self, key: &str) {
1041        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
1042            store.remove(key).await;
1043        }
1044    }
1045}
1046
1047#[cfg(test)]
1048mod sql_log_option_tests {
1049    use super::*;
1050
1051    #[test]
1052    fn diagnostic_sql_log_is_enabled_by_default_with_independent_switches() {
1053        let mut context = UserContext::default();
1054        assert_eq!(context.sql_log_options(), SqlLogOptions::all());
1055        assert!(context.sql_logs().is_empty());
1056
1057        context.disable_select_sql_log();
1058        assert_eq!(context.sql_log_options(), SqlLogOptions::mutation_only());
1059
1060        context.enable_select_sql_log();
1061        context.disable_mutation_sql_log();
1062        assert_eq!(context.sql_log_options(), SqlLogOptions::select_only());
1063
1064        context.disable_sql_log();
1065        assert_eq!(context.sql_log_options(), SqlLogOptions::disabled());
1066    }
1067
1068    #[test]
1069    fn disabled_sql_log_rejects_executor_metadata_before_recording() {
1070        let mut context = UserContext::default();
1071        context.disable_sql_log();
1072        let now = SystemTime::now();
1073        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1074            backend: "sql".to_owned(),
1075            operation: teaql_data_service::DataServiceOperation::Query,
1076            started_at: now,
1077            ended_at: now,
1078            affected_rows: None,
1079            result_count: Some(1),
1080            trace_chain: Vec::new(),
1081            comment: Some("disabled log test".to_owned()),
1082            backend_request_id: None,
1083            parameterized_query: Some("SELECT id FROM sample WHERE id = $1".to_owned()),
1084            params: vec![Value::I64(1)],
1085            debug_query: Some("SELECT id FROM sample WHERE id = 1".to_owned()),
1086        });
1087        assert!(context.sql_logs().is_empty());
1088    }
1089
1090    #[test]
1091    fn default_sql_telemetry_never_retains_bound_values_or_copy_paste_sql() {
1092        let mut context = UserContext::default();
1093        let buffer = UnifiedLogBuffer::default();
1094        context.insert_resource(buffer.clone());
1095        let now = SystemTime::now();
1096        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1097            backend: "sqlite".to_owned(),
1098            operation: teaql_data_service::DataServiceOperation::Query,
1099            started_at: now,
1100            ended_at: now,
1101            affected_rows: None,
1102            result_count: Some(1),
1103            trace_chain: Vec::new(),
1104            comment: Some("what: load an account".to_owned()),
1105            backend_request_id: None,
1106            parameterized_query: Some("SELECT id FROM account WHERE secret = ?".to_owned()),
1107            params: vec![Value::from("private-value-0123")],
1108            debug_query: Some(
1109                "SELECT id FROM account WHERE secret = 'private-value-0123'".to_owned(),
1110            ),
1111        });
1112        let entry = context.sql_logs().pop().expect("safe SQL log");
1113        assert_eq!(entry.sql, "SELECT id FROM account WHERE secret = ?");
1114        assert_eq!(entry.result_count, Some(1));
1115        assert!(entry.params.is_empty());
1116        assert!(entry.debug_sql.is_empty());
1117        assert!(entry.pretty_sql.is_empty());
1118        let buffered = buffer.entries.lock().unwrap();
1119        let LogPayload::Sql(buffered_sql) = &buffered[0].payload else {
1120            panic!("expected SQL log");
1121        };
1122        assert_eq!(buffered_sql, &entry);
1123        let ordinary_text = crate::log_formatter::LogFormatter::format_sql_log(
1124            &crate::log_formatter::HumanReaderFormatter,
1125            &entry.trace_path,
1126            &entry,
1127        );
1128        assert!(!ordinary_text.contains("private-value-0123"));
1129        assert!(!ordinary_text.contains("Debug SQL:"));
1130        let mut diagnostic_entry = entry.clone();
1131        diagnostic_entry.params = vec![Value::from("private-value-0123")];
1132        diagnostic_entry.debug_sql =
1133            "SELECT id FROM account WHERE secret = 'private-value-0123'".to_owned();
1134        let diagnostic_text = crate::log_formatter::LogFormatter::format_sql_log(
1135            &crate::log_formatter::HumanReaderFormatter,
1136            &diagnostic_entry.trace_path,
1137            &diagnostic_entry,
1138        );
1139        assert!(diagnostic_text.contains("Debug SQL:"));
1140        assert!(diagnostic_text.contains("private-value-0123"));
1141    }
1142
1143    #[test]
1144    fn metadata_log_retains_structured_intent_sql_forms_and_multilevel_trace() {
1145        let context = UserContext::default();
1146        let now = SystemTime::now();
1147        let query_trace = vec![
1148            teaql_core::TraceNode::typed(
1149                teaql_core::TraceKind::Comment,
1150                "School",
1151                None,
1152                "what: load school graph",
1153            ),
1154            teaql_core::TraceNode::typed(
1155                teaql_core::TraceKind::Purpose,
1156                "School",
1157                None,
1158                "why: render school details",
1159            ),
1160            teaql_core::TraceNode::typed(
1161                teaql_core::TraceKind::Relation,
1162                "platform",
1163                None,
1164                "School.platform",
1165            ),
1166            teaql_core::TraceNode::typed(
1167                teaql_core::TraceKind::Relation,
1168                "organization",
1169                None,
1170                "Platform.organization",
1171            ),
1172            teaql_core::TraceNode::typed(
1173                teaql_core::TraceKind::Relation,
1174                "region",
1175                None,
1176                "Organization.region",
1177            ),
1178        ];
1179        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1180            backend: "sqlite".to_owned(),
1181            operation: teaql_data_service::DataServiceOperation::Query,
1182            started_at: now,
1183            ended_at: now,
1184            affected_rows: None,
1185            result_count: Some(2),
1186            trace_chain: query_trace.clone(),
1187            comment: None,
1188            backend_request_id: None,
1189            parameterized_query: Some("SELECT name FROM school_data WHERE id = ?".to_owned()),
1190            params: vec![Value::I64(7)],
1191            debug_query: Some("SELECT name FROM school_data WHERE id = 7".to_owned()),
1192        });
1193
1194        let query = context.sql_logs().pop().expect("query log");
1195        assert_eq!(query.comment.as_deref(), Some("what: load school graph"));
1196        assert_eq!(query.purpose.as_deref(), Some("why: render school details"));
1197        assert_eq!(query.audit_reason, None);
1198        assert_eq!(
1199            query
1200                .trace_path
1201                .iter()
1202                .map(|node| node.kind)
1203                .collect::<Vec<_>>(),
1204            vec![
1205                teaql_core::TraceKind::Operation,
1206                teaql_core::TraceKind::Request,
1207                teaql_core::TraceKind::Relation,
1208                teaql_core::TraceKind::Relation,
1209                teaql_core::TraceKind::Relation,
1210                teaql_core::TraceKind::Provider,
1211                teaql_core::TraceKind::Sql,
1212            ]
1213        );
1214        assert_eq!(query.trace_path[0].entity_type, "School");
1215        assert_eq!(query.trace_path[5].entity_type, "sqlite");
1216        assert_eq!(query.trace_path[6].entity_type, "select");
1217        assert_eq!(query.sql, "SELECT name FROM school_data WHERE id = ?");
1218        assert!(query.params.is_empty());
1219        assert!(query.debug_sql.is_empty());
1220        assert_eq!(query.result_count, Some(2));
1221
1222        let mutation_trace = vec![teaql_core::TraceNode::typed(
1223            teaql_core::TraceKind::AuditReason,
1224            "School",
1225            Some(7),
1226            "correct school name",
1227        )];
1228        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1229            backend: "sqlite".to_owned(),
1230            operation: teaql_data_service::DataServiceOperation::Update,
1231            started_at: now,
1232            ended_at: now,
1233            affected_rows: Some(1),
1234            result_count: None,
1235            trace_chain: mutation_trace.clone(),
1236            comment: None,
1237            backend_request_id: None,
1238            parameterized_query: Some("UPDATE school_data SET name = ? WHERE id = ?".to_owned()),
1239            params: vec![Value::from("Academy"), Value::I64(7)],
1240            debug_query: Some("UPDATE school_data SET name = 'Academy' WHERE id = 7".to_owned()),
1241        });
1242        let mutation = context.sql_logs().pop().expect("mutation log");
1243        assert_eq!(mutation.comment, None);
1244        assert_eq!(mutation.purpose, None);
1245        assert_eq!(
1246            mutation.audit_reason.as_deref(),
1247            Some("correct school name")
1248        );
1249        assert_eq!(
1250            mutation
1251                .trace_path
1252                .iter()
1253                .map(|node| node.kind)
1254                .collect::<Vec<_>>(),
1255            vec![
1256                teaql_core::TraceKind::Operation,
1257                teaql_core::TraceKind::Entity,
1258                teaql_core::TraceKind::Provider,
1259                teaql_core::TraceKind::Sql,
1260            ]
1261        );
1262        assert_eq!(mutation.affected_rows, Some(1));
1263        assert!(mutation.params.is_empty());
1264        assert!(mutation.debug_sql.is_empty());
1265    }
1266}
1267
1268#[cfg(test)]
1269mod entity_runtime_state_tests {
1270    use super::*;
1271    use crate::EntityKey;
1272
1273    #[test]
1274    fn reused_user_context_returns_independent_mutation_ledgers() {
1275        let context = UserContext::default();
1276        let first = context.entity_runtime_state();
1277        let key = EntityKey::new("School", 1_u64);
1278        first.set(key.clone(), "name", "First");
1279
1280        let second = context.entity_runtime_state();
1281
1282        assert_eq!(first.changed_field_names(&key).len(), 1);
1283        assert!(second.changed_field_names(&key).is_empty());
1284    }
1285}