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
83pub 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 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 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)] 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)] 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 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 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 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 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 .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_eq!(entry.parameter_count(), 1);
1116 assert_eq!(entry.params, vec![Value::Null]);
1117 assert!(entry.debug_sql.is_empty());
1118 assert!(entry.pretty_sql.is_empty());
1119 let buffered = buffer.entries.lock().unwrap();
1120 let LogPayload::Sql(buffered_sql) = &buffered[0].payload else {
1121 panic!("expected SQL log");
1122 };
1123 assert_eq!(buffered_sql, &entry);
1124 let ordinary_text = crate::log_formatter::LogFormatter::format_sql_log(
1125 &crate::log_formatter::HumanReaderFormatter,
1126 &entry.trace_path,
1127 &entry,
1128 );
1129 assert!(!ordinary_text.contains("private-value-0123"));
1130 assert!(!ordinary_text.contains("Debug SQL:"));
1131 let mut diagnostic_entry = entry.clone();
1132 diagnostic_entry.params = vec![Value::from("private-value-0123")];
1133 diagnostic_entry.debug_sql =
1134 "SELECT id FROM account WHERE secret = 'private-value-0123'".to_owned();
1135 let diagnostic_text = crate::log_formatter::LogFormatter::format_sql_log(
1136 &crate::log_formatter::HumanReaderFormatter,
1137 &diagnostic_entry.trace_path,
1138 &diagnostic_entry,
1139 );
1140 assert!(diagnostic_text.contains("Debug SQL:"));
1141 assert!(diagnostic_text.contains("private-value-0123"));
1142 }
1143
1144 #[test]
1145 fn metadata_log_retains_structured_intent_sql_forms_and_multilevel_trace() {
1146 let context = UserContext::default();
1147 let now = SystemTime::now();
1148 let query_trace = vec![
1149 teaql_core::TraceNode::typed(
1150 teaql_core::TraceKind::Comment,
1151 "School",
1152 None,
1153 "what: load school graph",
1154 ),
1155 teaql_core::TraceNode::typed(
1156 teaql_core::TraceKind::Purpose,
1157 "School",
1158 None,
1159 "why: render school details",
1160 ),
1161 teaql_core::TraceNode::typed(
1162 teaql_core::TraceKind::Relation,
1163 "platform",
1164 None,
1165 "School.platform",
1166 ),
1167 teaql_core::TraceNode::typed(
1168 teaql_core::TraceKind::Relation,
1169 "organization",
1170 None,
1171 "Platform.organization",
1172 ),
1173 teaql_core::TraceNode::typed(
1174 teaql_core::TraceKind::Relation,
1175 "region",
1176 None,
1177 "Organization.region",
1178 ),
1179 ];
1180 context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1181 backend: "sqlite".to_owned(),
1182 operation: teaql_data_service::DataServiceOperation::Query,
1183 started_at: now,
1184 ended_at: now,
1185 affected_rows: None,
1186 result_count: Some(2),
1187 trace_chain: query_trace.clone(),
1188 comment: None,
1189 backend_request_id: None,
1190 parameterized_query: Some("SELECT name FROM school_data WHERE id = ?".to_owned()),
1191 params: vec![Value::I64(7)],
1192 debug_query: Some("SELECT name FROM school_data WHERE id = 7".to_owned()),
1193 });
1194
1195 let query = context.sql_logs().pop().expect("query log");
1196 assert_eq!(query.comment.as_deref(), Some("what: load school graph"));
1197 assert_eq!(query.purpose.as_deref(), Some("why: render school details"));
1198 assert_eq!(query.audit_reason, None);
1199 assert_eq!(
1200 query
1201 .trace_path
1202 .iter()
1203 .map(|node| node.kind)
1204 .collect::<Vec<_>>(),
1205 vec![
1206 teaql_core::TraceKind::Operation,
1207 teaql_core::TraceKind::Request,
1208 teaql_core::TraceKind::Relation,
1209 teaql_core::TraceKind::Relation,
1210 teaql_core::TraceKind::Relation,
1211 teaql_core::TraceKind::Provider,
1212 teaql_core::TraceKind::Sql,
1213 ]
1214 );
1215 assert_eq!(query.trace_path[0].entity_type, "School");
1216 assert_eq!(query.trace_path[5].entity_type, "sqlite");
1217 assert_eq!(query.trace_path[6].entity_type, "select");
1218 assert_eq!(query.sql, "SELECT name FROM school_data WHERE id = ?");
1219 assert_eq!(query.parameter_count(), 1);
1220 assert_eq!(query.params, vec![Value::Null]);
1221 assert!(query.debug_sql.is_empty());
1222 assert_eq!(query.result_count, Some(2));
1223
1224 let mutation_trace = vec![teaql_core::TraceNode::typed(
1225 teaql_core::TraceKind::AuditReason,
1226 "School",
1227 Some(7),
1228 "correct school name",
1229 )];
1230 context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1231 backend: "sqlite".to_owned(),
1232 operation: teaql_data_service::DataServiceOperation::Update,
1233 started_at: now,
1234 ended_at: now,
1235 affected_rows: Some(1),
1236 result_count: None,
1237 trace_chain: mutation_trace.clone(),
1238 comment: None,
1239 backend_request_id: None,
1240 parameterized_query: Some("UPDATE school_data SET name = ? WHERE id = ?".to_owned()),
1241 params: vec![Value::from("Academy"), Value::I64(7)],
1242 debug_query: Some("UPDATE school_data SET name = 'Academy' WHERE id = 7".to_owned()),
1243 });
1244 let mutation = context.sql_logs().pop().expect("mutation log");
1245 assert_eq!(mutation.parameter_count(), 2);
1246 assert_eq!(mutation.params, vec![Value::Null, Value::Null]);
1247 assert_eq!(mutation.comment, None);
1248 assert_eq!(mutation.purpose, None);
1249 assert_eq!(
1250 mutation.audit_reason.as_deref(),
1251 Some("correct school name")
1252 );
1253 assert_eq!(
1254 mutation
1255 .trace_path
1256 .iter()
1257 .map(|node| node.kind)
1258 .collect::<Vec<_>>(),
1259 vec![
1260 teaql_core::TraceKind::Operation,
1261 teaql_core::TraceKind::Entity,
1262 teaql_core::TraceKind::Provider,
1263 teaql_core::TraceKind::Sql,
1264 ]
1265 );
1266 assert_eq!(mutation.affected_rows, Some(1));
1267 assert!(mutation.debug_sql.is_empty());
1268 }
1269}
1270
1271#[cfg(test)]
1272mod entity_runtime_state_tests {
1273 use super::*;
1274 use crate::EntityKey;
1275
1276 #[test]
1277 fn reused_user_context_returns_independent_mutation_ledgers() {
1278 let context = UserContext::default();
1279 let first = context.entity_runtime_state();
1280 let key = EntityKey::new("School", 1_u64);
1281 first.set(key.clone(), "name", "First");
1282
1283 let second = context.entity_runtime_state();
1284
1285 assert_eq!(first.changed_field_names(&key).len(), 1);
1286 assert!(second.changed_field_names(&key).is_empty());
1287 }
1288}