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