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 if std::time::Instant::now() > *exp {
279 return None;
280 }
281 }
282 return Some(val.clone());
283 }
284 None
285 }
286
287 async fn put(&self, key: &str, value: Value, timeout_seconds: Option<u64>) {
288 let mut lock = self.cache.write().unwrap();
289 let expires_at = timeout_seconds
290 .map(|secs| std::time::Instant::now() + std::time::Duration::from_secs(secs));
291 lock.insert(key.to_string(), (value, expires_at));
292 }
293
294 async fn remove(&self, key: &str) {
295 let mut lock = self.cache.write().unwrap();
296 lock.remove(key);
297 }
298}
299
300impl UserContext {
301 pub fn new() -> Self {
302 Self::default()
303 }
304
305 pub fn with_active_root(mut self, entity_type: impl Into<String>, id: u64) -> Self {
306 let entity_type = crate::canonical_id_space_entity(&entity_type.into());
307 assert!(
308 !entity_type.trim().is_empty(),
309 "active root entity type is required"
310 );
311 assert!(id > 0, "active root id must be positive");
312 self.active_root
313 .set(ContextEntityRef { entity_type, id })
314 .expect("active root may only be assigned once");
315 self
316 }
317
318 #[doc(hidden)]
319 pub fn set_generated_bootstrap_active_root(
320 &self,
321 entity_type: impl Into<String>,
322 id: u64,
323 ) -> Result<(), RuntimeError> {
324 let entity_type = crate::canonical_id_space_entity(&entity_type.into());
325 if entity_type.trim().is_empty() || id == 0 {
326 return Err(RuntimeError::Schema(
327 "invalid generated active root".to_owned(),
328 ));
329 }
330 match self.active_root.get() {
331 Some(existing) if existing.entity_type == entity_type && existing.id == id => Ok(()),
332 Some(existing) => Err(RuntimeError::Schema(format!(
333 "active root already set to {}:{}",
334 existing.entity_type, existing.id
335 ))),
336 None => self
337 .active_root
338 .set(ContextEntityRef { entity_type, id })
339 .map_err(|_| RuntimeError::Schema("active root initialization raced".to_owned())),
340 }
341 }
342
343 pub fn require_active_root(
344 &self,
345 expected_entity_type: &str,
346 ) -> Result<&ContextEntityRef, ContextRootError> {
347 let canonical_expected = crate::canonical_id_space_entity(expected_entity_type);
348 match self.active_root.get() {
349 Some(root) if root.entity_type == canonical_expected => Ok(root),
350 actual_root => Err(ContextRootError {
351 expected_entity_type: canonical_expected,
352 actual_root: actual_root.cloned(),
353 }),
354 }
355 }
356
357 pub(crate) fn active_root_ref(&self) -> Option<&ContextEntityRef> {
358 self.active_root.get()
359 }
360
361 pub fn with_runtime_telemetry(mut self, telemetry: Arc<dyn crate::RuntimeTelemetry>) -> Self {
362 self.runtime_telemetry = telemetry;
363 self
364 }
365
366 pub fn set_runtime_telemetry(&mut self, telemetry: Arc<dyn crate::RuntimeTelemetry>) {
367 self.runtime_telemetry = telemetry;
368 }
369
370 pub fn runtime_telemetry(&self) -> &Arc<dyn crate::RuntimeTelemetry> {
371 &self.runtime_telemetry
372 }
373
374 pub(crate) fn runtime_telemetry_is_noop(&self) -> bool {
375 self.runtime_telemetry.is_noop()
376 }
377
378 pub fn start_runtime_operation(
379 &self,
380 operation: crate::RuntimeOperation,
381 ) -> crate::FailOpenRuntimeTelemetryScope {
382 crate::start_runtime_operation(&self.runtime_telemetry, operation)
383 }
384
385 pub fn user_identifier(&self) -> Option<&str> {
386 self.user_identifier.as_deref()
387 }
388
389 pub fn set_user_identifier(&mut self, user_identifier: impl Into<String>) {
390 self.user_identifier = Some(user_identifier.into());
391 }
392
393 pub fn with_user_identifier(mut self, user_identifier: impl Into<String>) -> Self {
394 self.user_identifier = Some(user_identifier.into());
395 self
396 }
397
398 pub fn set_user_identifier_option(&mut self, user_identifier: Option<String>) {
399 self.user_identifier = user_identifier;
400 }
401
402 pub fn with_user_identifier_option(mut self, user_identifier: Option<String>) -> Self {
403 self.user_identifier = user_identifier;
404 self
405 }
406
407 pub fn timezone(&self) -> Option<&str> {
408 self.timezone.as_deref()
409 }
410
411 pub fn set_timezone(&mut self, timezone: impl Into<String>) {
412 self.timezone = Some(timezone.into());
413 }
414
415 pub fn with_timezone(mut self, timezone: impl Into<String>) -> Self {
416 self.timezone = Some(timezone.into());
417 self
418 }
419
420 pub fn trace_id(&self) -> &str {
421 &self.trace_id
422 }
423
424 pub fn set_trace_id(&mut self, trace_id: impl Into<String>) {
425 self.trace_id = trace_id.into();
426 }
427
428 pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
429 self.trace_id = trace_id.into();
430 self
431 }
432
433 pub fn with_module(mut self, module: crate::RuntimeModule) -> Self {
434 module.apply_to(&mut self);
435 self
436 }
437
438 pub fn entity_runtime_state(&self) -> EntityRuntimeState {
439 EntityRuntimeState::fresh_with_shared_graph(&self.entity_runtime_state)
442 }
443
444 pub fn initial_graphs(&self) -> &[GraphNode] {
445 &self.initial_graphs
446 }
447
448 pub fn set_initial_graphs(&mut self, graphs: Vec<GraphNode>) {
449 self.initial_graphs = graphs;
450 }
451
452 pub fn root_graphs(&self) -> &[GraphNode] {
453 &self.root_graphs
454 }
455
456 pub fn set_root_graphs(&mut self, graphs: Vec<GraphNode>) {
457 self.root_graphs = graphs;
458 }
459
460 pub fn with_metadata(mut self, metadata: impl MetadataStore + 'static) -> Self {
461 self.metadata = Some(Box::new(metadata));
462 self
463 }
464
465 pub fn set_metadata(&mut self, metadata: impl MetadataStore + 'static) {
466 self.metadata = Some(Box::new(metadata));
467 }
468
469 pub fn with_entity_registry(mut self, registry: impl EntityRegistry + 'static) -> Self {
470 self.entity_registry = Some(Box::new(registry));
471 self
472 }
473
474 pub fn set_entity_registry(&mut self, registry: impl EntityRegistry + 'static) {
475 self.entity_registry = Some(Box::new(registry));
476 }
477
478 pub fn set_entity_graph_decoder_registry(
479 &mut self,
480 registry: InMemoryEntityGraphDecoderRegistry,
481 ) {
482 self.entity_graph_decoders = registry;
483 }
484
485 pub(crate) fn has_entity_graph_decoder(&self, entity: &str) -> bool {
486 self.entity_graph_decoders.contains(entity)
487 }
488
489 pub(crate) fn decode_compact_entity_into_graph(
490 &self,
491 entity: &str,
492 row: teaql_core::CompactRow,
493 root: &EntityRuntimeState,
494 graph: &mut EntityGraphBuilder,
495 ) -> Result<(), teaql_core::EntityError> {
496 self.entity_graph_decoders
497 .decode_compact(entity, row, root, graph)
498 }
499
500 pub(crate) fn decode_compact_entity_list_into_graph(
501 &self,
502 entity: &str,
503 rows: Vec<teaql_core::CompactRow>,
504 root: &EntityRuntimeState,
505 graph: &mut EntityGraphBuilder,
506 owner_entity: &str,
507 owner_id: u64,
508 relation: &str,
509 ) -> Result<(), teaql_core::EntityError> {
510 self.entity_graph_decoders.decode_compact_list(
511 entity,
512 rows,
513 root,
514 graph,
515 owner_entity,
516 owner_id,
517 relation,
518 )
519 }
520
521 pub(crate) fn decode_compact_entity_batch_into_graph(
522 &self,
523 entity: &str,
524 rows: Vec<teaql_core::CompactRow>,
525 root: &EntityRuntimeState,
526 graph: &mut EntityGraphBuilder,
527 ) -> Result<(), teaql_core::EntityError> {
528 self.entity_graph_decoders
529 .decode_compact_batch(entity, rows, root, graph)
530 }
531
532 pub(crate) fn decode_compact_entity_option_into_graph(
533 &self,
534 entity: &str,
535 rows: Vec<teaql_core::CompactRow>,
536 root: &EntityRuntimeState,
537 graph: &mut EntityGraphBuilder,
538 owner_entity: &str,
539 owner_id: u64,
540 relation: &str,
541 ) -> Result<(), teaql_core::EntityError> {
542 self.entity_graph_decoders.decode_compact_option(
543 entity,
544 rows,
545 root,
546 graph,
547 owner_entity,
548 owner_id,
549 relation,
550 )
551 }
552
553 pub fn with_entity_data_service_behavior_registry(
554 mut self,
555 registry: impl EntityDataServiceBehaviorRegistry + 'static,
556 ) -> Self {
557 self.entity_data_service_behavior_registry = Some(Box::new(registry));
558 self
559 }
560
561 pub fn set_entity_data_service_behavior_registry(
562 &mut self,
563 registry: impl EntityDataServiceBehaviorRegistry + 'static,
564 ) {
565 self.entity_data_service_behavior_registry = Some(Box::new(registry));
566 }
567
568 pub fn with_request_policy(mut self, policy: impl RequestPolicy + 'static) -> Self {
569 self.request_policy = Some(Box::new(policy));
570 self
571 }
572
573 pub fn set_request_policy(&mut self, policy: impl RequestPolicy + 'static) {
574 self.request_policy = Some(Box::new(policy));
575 }
576
577 pub fn clear_request_policy(&mut self) {
578 self.request_policy = None;
579 }
580
581 pub fn with_checker_registry(mut self, registry: impl CheckerRegistry + 'static) -> Self {
582 self.checker_registry = Some(Box::new(registry));
583 self
584 }
585
586 pub fn set_checker_registry(&mut self, registry: impl CheckerRegistry + 'static) {
587 self.checker_registry = Some(Box::new(registry));
588 }
589
590 pub(crate) fn with_event_sink(mut self, sink: impl RawAuditEventSink + 'static) -> Self {
591 self.event_sink = Some(Box::new(sink));
592 self
593 }
594
595 pub(crate) fn set_event_sink(&mut self, sink: impl RawAuditEventSink + 'static) {
596 self.event_sink = Some(Box::new(sink));
597 }
598
599 pub fn with_custom_event_sink(
600 mut self,
601 sink: impl crate::SafeAuditEventSink + 'static,
602 ) -> Self {
603 self.custom_event_sink = Some(Box::new(sink));
604 self
605 }
606
607 pub fn set_custom_event_sink(&mut self, sink: impl crate::SafeAuditEventSink + 'static) {
608 self.custom_event_sink = Some(Box::new(sink));
609 }
610
611 pub fn with_internal_id_generator(
612 mut self,
613 generator: impl InternalIdGenerator + 'static,
614 ) -> Self {
615 self.internal_id_generator = Some(Box::new(generator));
616 self
617 }
618
619 pub fn set_internal_id_generator(&mut self, generator: impl InternalIdGenerator + 'static) {
620 self.internal_id_generator = Some(Box::new(generator));
621 }
622
623 pub fn with_schema_provider(mut self, provider: impl SchemaProvider + 'static) -> Self {
624 self.schema_provider = Some(Box::new(provider));
625 self
626 }
627
628 pub fn set_schema_provider(&mut self, provider: impl SchemaProvider + 'static) {
629 self.schema_provider = Some(Box::new(provider));
630 }
631
632 pub async fn ensure_schema(&self) -> Result<(), RuntimeError> {
633 let provider = self
634 .schema_provider
635 .as_ref()
636 .ok_or_else(|| RuntimeError::Schema("missing schema provider".to_owned()))?;
637 let invocation = SchemaInvocation { _context_owned: () };
638 provider.ensure_schema(self, &invocation).await?;
639 GENERATED_SCHEMA_BOOTSTRAP_MODE
640 .scope((), async {
641 for bootstrap in &self.generated_schema_bootstraps {
642 bootstrap(self).await?;
643 }
644 Ok::<(), RuntimeError>(())
645 })
646 .await?;
647 Ok(())
648 }
649
650 pub(crate) fn is_generated_schema_bootstrap(&self) -> bool {
651 GENERATED_SCHEMA_BOOTSTRAP_MODE
652 .try_with(|_| true)
653 .unwrap_or(false)
654 }
655
656 pub(crate) fn set_generated_schema_bootstraps(
657 &mut self,
658 bootstraps: Vec<GeneratedSchemaBootstrap>,
659 ) {
660 self.generated_schema_bootstraps = bootstraps;
661 }
662
663 #[doc(hidden)]
664 pub fn initialize_generated_bootstrap_entity<E: teaql_core::Entity>(
665 &self,
666 entity: &mut E,
667 entity_name: &str,
668 fixed_id: u64,
669 ) -> Result<(), RuntimeError> {
670 let generator = self.internal_id_generator.as_ref().ok_or_else(|| {
671 RuntimeError::IdGeneration("missing internal ID generator".to_owned())
672 })?;
673 generator.ensure_floor(entity_name, fixed_id)?;
674 entity.mark_as_new();
675 Ok(())
676 }
677
678 pub fn with_language(mut self, language: Language) -> Self {
679 self.language = language;
680 self
681 }
682
683 pub fn set_language(&mut self, language: Language) {
684 self.language = language;
685 }
686
687 pub fn with_i18n_catalog(mut self, catalog: Arc<crate::I18nCatalog>) -> Self {
688 self.i18n_catalog = catalog;
689 self
690 }
691
692 pub fn set_i18n_catalog(&mut self, catalog: Arc<crate::I18nCatalog>) {
693 self.i18n_catalog = catalog;
694 }
695
696 pub fn language(&self) -> Language {
697 self.language
698 }
699
700 pub fn set_language_code(&mut self, code: &str) -> Result<(), RuntimeError> {
701 let Some(language) = Language::from_code(code) else {
702 return Err(RuntimeError::UnsupportedLocale(code.to_owned()));
703 };
704 self.language = language;
705 Ok(())
706 }
707
708 pub fn set_locale_code(&mut self, code: &str) -> Result<(), RuntimeError> {
709 self.set_language_code(code)
710 }
711
712 pub fn generate_id(&self, entity: &str) -> Result<Option<u64>, RuntimeError> {
713 self.internal_id_generator
714 .as_ref()
715 .map(|generator| generator.generate_id(entity))
716 .transpose()
717 }
718
719 pub fn next_id(&self, entity: &str) -> Result<u64, RuntimeError> {
720 match self.generate_id(entity)? {
721 Some(id) => Ok(id),
722 None => local_id_generator().generate_id(entity),
723 }
724 }
725
726 pub fn entity(&self, name: &str) -> Option<&EntityDescriptor> {
727 self.metadata
728 .as_ref()
729 .and_then(|metadata| metadata.entity(name))
730 }
731
732 pub fn all_entities(&self) -> Vec<&EntityDescriptor> {
733 self.metadata
734 .as_ref()
735 .map(|metadata| metadata.all_entities())
736 .unwrap_or_default()
737 }
738
739 pub fn require_entity(&self, name: &str) -> Result<&EntityDescriptor, RuntimeError> {
740 self.entity(name)
741 .ok_or_else(|| RuntimeError::MissingEntity(name.to_owned()))
742 }
743
744 pub fn insert_resource<T>(&mut self, resource: T)
745 where
746 T: Send + Sync + 'static,
747 {
748 self.typed_resources
749 .insert(TypeId::of::<T>(), Box::new(resource));
750 }
751
752 pub fn get_resource<T>(&self) -> Option<&T>
753 where
754 T: Send + Sync + 'static,
755 {
756 self.typed_resources
757 .get(&TypeId::of::<T>())
758 .and_then(|value| value.downcast_ref::<T>())
759 }
760
761 pub fn require_resource<T>(&self) -> Result<&T, ContextError>
762 where
763 T: Send + Sync + 'static,
764 {
765 self.get_resource::<T>()
766 .ok_or(ContextError::MissingTypedResource(
767 std::any::type_name::<T>(),
768 ))
769 }
770
771 pub fn insert_named_resource<T>(&mut self, name: impl Into<String>, resource: T)
772 where
773 T: Send + Sync + 'static,
774 {
775 self.named_resources.insert(name.into(), Box::new(resource));
776 }
777
778 pub fn get_named_resource<T>(&self, name: &str) -> Option<&T>
779 where
780 T: Send + Sync + 'static,
781 {
782 self.named_resources
783 .get(name)
784 .and_then(|value| value.downcast_ref::<T>())
785 }
786
787 pub fn require_named_resource<T>(&self, name: &str) -> Result<&T, ContextError>
788 where
789 T: Send + Sync + 'static,
790 {
791 self.get_named_resource::<T>(name)
792 .ok_or_else(|| ContextError::MissingResource(name.to_owned()))
793 }
794
795 pub fn put_local(&mut self, key: impl Into<String>, value: impl Into<Value>) {
796 self.locals.insert(key.into(), value.into());
797 }
798
799 pub fn local(&self, key: &str) -> Option<&Value> {
800 self.locals.get(key)
801 }
802
803 pub fn remove_local(&mut self, key: &str) -> Option<Value> {
804 self.locals.remove(key)
805 }
806
807 pub fn has_entity_data_service(&self, entity: &str) -> bool {
808 let in_registry = self
809 .entity_registry
810 .as_ref()
811 .map(|registry| registry.contains(entity))
812 .unwrap_or(false);
813 in_registry || self.entity(entity).is_some()
814 }
815
816 pub fn entity_data_service_behavior(
817 &self,
818 entity: &str,
819 ) -> Option<std::sync::Arc<dyn EntityDataServiceBehavior>> {
820 self.entity_data_service_behavior_registry
821 .as_ref()
822 .and_then(|registry| registry.behavior(entity))
823 }
824
825 pub fn has_checker(&self, entity: &str) -> bool {
826 self.checker_registry
827 .as_ref()
828 .and_then(|registry| registry.checker(entity))
829 .is_some()
830 }
831
832 pub fn fix_time(&self) -> teaql_core::time::Timestamp {
836 crate::entity_save::current_graph_fix_time()
837 }
838
839 pub fn record_fix_evidence(&self, evidence: FixEvidence) {
840 crate::entity_save::record_graph_fix_evidence(evidence);
841 }
842
843 pub(crate) fn replace_last_fix_evidence(&self, evidence: Vec<FixEvidence>) {
844 *self.last_fix_evidence.lock().unwrap() = evidence;
845 }
846
847 pub fn last_fix_evidence(&self) -> Vec<FixEvidence> {
848 self.last_fix_evidence.lock().unwrap().clone()
849 }
850
851 pub fn check_and_fix_values(
852 &self,
853 entity: &str,
854 values: &mut crate::EntityValues,
855 ) -> Result<(), RuntimeError> {
856 self.check_and_fix_values_at(entity, values, &ObjectLocation::root())
857 }
858
859 pub fn check_and_fix_values_at(
860 &self,
861 entity: &str,
862 values: &mut crate::EntityValues,
863 location: &ObjectLocation,
864 ) -> Result<(), RuntimeError> {
865 let status = CheckObjectStatus::from_values(values);
866 let checker = self
867 .checker_registry
868 .as_ref()
869 .and_then(|registry| registry.checker(entity));
870 let mut results = CheckResults::new();
871 if let Some(checker) = checker {
872 checker.check_and_fix(self, values, location, &mut results);
873 }
874
875 if let Some(descriptor) = self
880 .metadata
881 .as_ref()
882 .and_then(|metadata| metadata.entity(entity))
883 {
884 for property in descriptor
885 .properties
886 .iter()
887 .filter(|property| !property.nullable && !property.is_version)
891 {
892 let missing = !values.contains_key(&property.name);
893 let null = matches!(values.get(&property.name), Some(Value::Null));
894 let property_location = location.clone().member(&property.name);
895 let already_reported = results.iter().any(|result| {
896 result.rule == crate::CheckRule::Required
897 && result.location == property_location
898 });
899 if ((status.is_create() && missing) || null) && !already_reported {
900 results.push(CheckResult::required(property_location));
901 }
902 }
903 }
904 if results.is_empty() {
905 return Ok(());
906 }
907 self.translate_check_results(&mut results);
908 Err(RuntimeError::Check(results))
909 }
910
911 pub fn translate_check_results(&self, results: &mut CheckResults) {
912 for result in results {
913 if result.message.is_none() {
914 result.message = Some(
915 self.i18n_catalog
916 .translate_check_result(self.language, result),
917 );
918 }
919 }
920 }
921
922 pub fn send_event(&self, mut event: RawAuditEvent) -> Result<(), RuntimeError> {
923 if self.is_generated_schema_bootstrap()
924 && matches!(
925 event.kind,
926 crate::RawAuditEventKind::Created | crate::RawAuditEventKind::Updated
927 )
928 {
929 let reason = event
930 .trace_chain
931 .last()
932 .map(|node| node.comment.clone())
933 .unwrap_or_else(|| "generated runtime bootstrap".to_owned());
934 let resulting_version = event
935 .new_values
936 .as_ref()
937 .and_then(|values| values.get("version"))
938 .or_else(|| event.values.get("version"))
939 .and_then(teaql_core::Value::try_i64);
940 let occurred_at_millis = std::time::SystemTime::now()
941 .duration_since(std::time::UNIX_EPOCH)
942 .unwrap_or_default()
943 .as_millis() as u64;
944 event.bootstrap_audit = Some(crate::BootstrapAuditIdentity {
945 actor: "teaql-generated-bootstrap".to_owned(),
946 category: "runtime-bootstrap".to_owned(),
947 reason,
948 resulting_version,
949 occurred_at_millis,
950 });
951 }
952 let scope = self.start_runtime_operation(
953 crate::RuntimeOperation::new("audit", format!("{}.event", event.entity))
954 .attribute("teaql.entity.type", event.entity.clone()),
955 );
956 let result = self.send_event_inner(event);
957 match &result {
958 Ok(()) => scope.success(std::collections::BTreeMap::new()),
959 Err(_) => scope.failure("audit_error"),
960 }
961 result
962 }
963
964 fn send_event_inner(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
965 if let Some(sink) = self.event_sink.as_ref() {
966 sink.on_event(self, &event)?;
967 }
968 if let Some(sink) = self.custom_event_sink.as_ref() {
969 let (mask_fields, max_len) = self
970 .metadata
971 .as_ref()
972 .and_then(|metadata| metadata.entity(&event.entity))
973 .map(|desc| (desc.audit_mask_fields.clone(), desc.audit_value_max_len))
974 .unwrap_or_else(|| (vec![], None));
975
976 let safe_event = event.build_safe_event(&mask_fields, max_len);
977 sink.on_safe_event(self, &safe_event)?;
978 }
979
980 crate::log_formatter::LogManager::write_audit_log(&event);
981
982 Ok(())
983 }
984
985 pub async fn get_in_store(&self, key: &str) -> Option<Value> {
986 let store = self.get_resource::<Box<dyn DataStore>>()?;
987 store.get(key).await
988 }
989
990 pub async fn put_in_store(
991 &self,
992 key: &str,
993 value: impl Into<Value>,
994 timeout_seconds: Option<u64>,
995 ) {
996 if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
997 store.put(key, value.into(), timeout_seconds).await;
998 }
999 }
1000
1001 pub async fn clear_in_store(&self, key: &str) {
1002 if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
1003 store.remove(key).await;
1004 }
1005 }
1006}
1007
1008#[cfg(test)]
1009mod sql_log_option_tests {
1010 use super::*;
1011
1012 #[test]
1013 fn diagnostic_sql_log_is_enabled_by_default_with_independent_switches() {
1014 let mut context = UserContext::default();
1015 assert_eq!(context.sql_log_options(), SqlLogOptions::all());
1016 assert!(context.sql_logs().is_empty());
1017
1018 context.disable_select_sql_log();
1019 assert_eq!(context.sql_log_options(), SqlLogOptions::mutation_only());
1020
1021 context.enable_select_sql_log();
1022 context.disable_mutation_sql_log();
1023 assert_eq!(context.sql_log_options(), SqlLogOptions::select_only());
1024
1025 context.disable_sql_log();
1026 assert_eq!(context.sql_log_options(), SqlLogOptions::disabled());
1027 }
1028
1029 #[test]
1030 fn disabled_sql_log_rejects_executor_metadata_before_recording() {
1031 let mut context = UserContext::default();
1032 context.disable_sql_log();
1033 let now = SystemTime::now();
1034 context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1035 backend: "sql".to_owned(),
1036 operation: teaql_data_service::DataServiceOperation::Query,
1037 started_at: now,
1038 ended_at: now,
1039 affected_rows: None,
1040 result_count: Some(1),
1041 trace_chain: Vec::new(),
1042 comment: Some("disabled log test".to_owned()),
1043 backend_request_id: None,
1044 parameterized_query: Some("SELECT id FROM sample WHERE id = $1".to_owned()),
1045 params: vec![Value::I64(1)],
1046 debug_query: Some("SELECT id FROM sample WHERE id = 1".to_owned()),
1047 });
1048 assert!(context.sql_logs().is_empty());
1049 }
1050
1051 #[test]
1052 fn metadata_log_retains_structured_intent_sql_forms_and_multilevel_trace() {
1053 let context = UserContext::default();
1054 let now = SystemTime::now();
1055 let query_trace = vec![
1056 teaql_core::TraceNode::typed(
1057 teaql_core::TraceKind::Comment,
1058 "School",
1059 None,
1060 "what: load school graph",
1061 ),
1062 teaql_core::TraceNode::typed(
1063 teaql_core::TraceKind::Purpose,
1064 "School",
1065 None,
1066 "why: render school details",
1067 ),
1068 teaql_core::TraceNode::typed(
1069 teaql_core::TraceKind::Relation,
1070 "platform",
1071 None,
1072 "School.platform",
1073 ),
1074 teaql_core::TraceNode::typed(
1075 teaql_core::TraceKind::Relation,
1076 "organization",
1077 None,
1078 "Platform.organization",
1079 ),
1080 teaql_core::TraceNode::typed(
1081 teaql_core::TraceKind::Relation,
1082 "region",
1083 None,
1084 "Organization.region",
1085 ),
1086 ];
1087 context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1088 backend: "sqlite".to_owned(),
1089 operation: teaql_data_service::DataServiceOperation::Query,
1090 started_at: now,
1091 ended_at: now,
1092 affected_rows: None,
1093 result_count: Some(2),
1094 trace_chain: query_trace.clone(),
1095 comment: None,
1096 backend_request_id: None,
1097 parameterized_query: Some("SELECT name FROM school_data WHERE id = ?".to_owned()),
1098 params: vec![Value::I64(7)],
1099 debug_query: Some("SELECT name FROM school_data WHERE id = 7".to_owned()),
1100 });
1101
1102 let query = context.sql_logs().pop().expect("query log");
1103 assert_eq!(query.comment.as_deref(), Some("what: load school graph"));
1104 assert_eq!(query.purpose.as_deref(), Some("why: render school details"));
1105 assert_eq!(query.audit_reason, None);
1106 assert_eq!(
1107 query
1108 .trace_path
1109 .iter()
1110 .map(|node| node.kind)
1111 .collect::<Vec<_>>(),
1112 vec![
1113 teaql_core::TraceKind::Operation,
1114 teaql_core::TraceKind::Request,
1115 teaql_core::TraceKind::Relation,
1116 teaql_core::TraceKind::Relation,
1117 teaql_core::TraceKind::Relation,
1118 teaql_core::TraceKind::Provider,
1119 teaql_core::TraceKind::Sql,
1120 ]
1121 );
1122 assert_eq!(query.trace_path[0].entity_type, "School");
1123 assert_eq!(query.trace_path[5].entity_type, "sqlite");
1124 assert_eq!(query.trace_path[6].entity_type, "select");
1125 assert_eq!(query.sql, "SELECT name FROM school_data WHERE id = ?");
1126 assert_eq!(query.params, vec![Value::I64(7)]);
1127 assert_eq!(query.debug_sql, "SELECT name FROM school_data WHERE id = 7");
1128 assert_eq!(query.result_count, Some(2));
1129
1130 let mutation_trace = vec![teaql_core::TraceNode::typed(
1131 teaql_core::TraceKind::AuditReason,
1132 "School",
1133 Some(7),
1134 "correct school name",
1135 )];
1136 context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
1137 backend: "sqlite".to_owned(),
1138 operation: teaql_data_service::DataServiceOperation::Update,
1139 started_at: now,
1140 ended_at: now,
1141 affected_rows: Some(1),
1142 result_count: None,
1143 trace_chain: mutation_trace.clone(),
1144 comment: None,
1145 backend_request_id: None,
1146 parameterized_query: Some("UPDATE school_data SET name = ? WHERE id = ?".to_owned()),
1147 params: vec![Value::from("Academy"), Value::I64(7)],
1148 debug_query: Some("UPDATE school_data SET name = 'Academy' WHERE id = 7".to_owned()),
1149 });
1150 let mutation = context.sql_logs().pop().expect("mutation log");
1151 assert_eq!(mutation.comment, None);
1152 assert_eq!(mutation.purpose, None);
1153 assert_eq!(
1154 mutation.audit_reason.as_deref(),
1155 Some("correct school name")
1156 );
1157 assert_eq!(
1158 mutation
1159 .trace_path
1160 .iter()
1161 .map(|node| node.kind)
1162 .collect::<Vec<_>>(),
1163 vec![
1164 teaql_core::TraceKind::Operation,
1165 teaql_core::TraceKind::Entity,
1166 teaql_core::TraceKind::Provider,
1167 teaql_core::TraceKind::Sql,
1168 ]
1169 );
1170 assert_eq!(mutation.affected_rows, Some(1));
1171 }
1172}
1173
1174#[cfg(test)]
1175mod entity_runtime_state_tests {
1176 use super::*;
1177 use crate::EntityKey;
1178
1179 #[test]
1180 fn reused_user_context_returns_independent_mutation_ledgers() {
1181 let context = UserContext::default();
1182 let first = context.entity_runtime_state();
1183 let key = EntityKey::new("School", 1_u64);
1184 first.set(key.clone(), "name", "First");
1185
1186 let second = context.entity_runtime_state();
1187
1188 assert_eq!(first.changed_field_names(&key).len(), 1);
1189 assert!(second.changed_field_names(&key).is_empty());
1190 }
1191}