1#![allow(warnings)]
2extern crate self as teaql_runtime;
3mod checker;
4mod context;
5mod data_service;
6mod entity_runtime;
7pub mod entity_save;
8mod entity_status;
9mod error;
10mod event;
11pub mod generated_support;
12mod graph;
13mod i18n;
14mod id;
15pub mod inmemory_engine;
16mod language;
17pub mod log_formatter;
18mod memory;
19mod registry;
20mod telemetry;
21#[cfg(feature = "opentelemetry")]
22mod telemetry_opentelemetry;
23
24pub use context::{
25 ContextEntityRef, ContextRootError, ContinuousPageCursor, ContinuousPageCursorStore, DataStore,
26 InMemoryContinuousPageCursorStore, InMemoryDataStore, InfoLogEntry, LogPayload,
27 RemoteLockProvider, SchemaProvider, SqlLogEntry, SqlLogOperation, SqlLogOptions,
28 UnifiedLogBuffer, UnifiedLogEntry, UserContext,
29};
30pub use data_service::{
31 AggregationCacheBackend, EntityDataService, GraphTransactionBoundary, InMemoryAggregationCache,
32 RelationLoadPlan,
33};
34pub use entity_runtime::{
35 ChangeSetStack, EntityChangeSet, EntityGraphBuilder, EntityKey, EntityRoot, LedgerEntity,
36 RootContext,
37};
38pub use entity_save::{AuditedSaveExt, graph_node_from_entity, save_audited_ledger_entity};
39pub use entity_status::{EntityAction, EntityStatus};
40pub use error::{ContextError, DataServiceError, RuntimeError};
41pub use event::{
42 EntityPropertyChange, InMemoryRawAuditEventSink, RawAuditEvent, RawAuditEventKind,
43 RawAuditEventSink, SafeAuditEvent, SafeAuditEventSink, SafeAuditField,
44};
45pub use generated_support::*;
46pub use graph::{
47 EntityValues, GraphMutationBatch, GraphMutationKind, GraphMutationPlan, GraphMutationPlanItem,
48 GraphNode, GraphOperation, ScopedCommentNode, TraceScopeToken, sorted_update_fields,
49};
50pub use i18n::I18nCatalog;
51pub(crate) use id::local_id_generator;
52pub use id::{AtomicCounterIdGenerator, InternalIdGenerator, SnowflakeIdGenerator};
53pub use inmemory_engine::{ExprEvaluator, InMemoryQueryEngine};
54pub use language::{
55 BuiltinTranslator, Language, Locale, MessageTranslator, translate_check_result,
56 translate_location,
57};
58pub(crate) use memory::MemoryDataService;
59pub use registry::{
60 EntityDataServiceBehavior, EntityDataServiceBehaviorRegistry, EntityRegistry,
61 InMemoryEntityDataServiceBehaviorRegistry, InMemoryEntityGraphDecoderRegistry,
62 InMemoryEntityRegistry, InMemoryMetadataStore, MetadataStore, RequestPolicy, RuntimeModule,
63};
64pub use telemetry::{
65 FailOpenRuntimeTelemetryPropagationContext, FailOpenRuntimeTelemetryScope,
66 NoopRuntimeTelemetry, RuntimeAttributeValue, RuntimeOperation, RuntimeTelemetry,
67 RuntimeTelemetryPropagationContext, RuntimeTelemetryScope, extract_runtime_context,
68 runtime_error_category, start_runtime_operation,
69};
70#[cfg(feature = "opentelemetry")]
71pub use telemetry_opentelemetry::OpenTelemetryRuntimeTelemetry;
72
73#[cfg(test)]
74mod tests {
75 use std::collections::{BTreeMap, VecDeque};
76 use std::sync::{Arc, Mutex};
77
78 use super::{
79 AggregationCacheBackend, CHECK_OBJECT_STATUS_FIELD, CheckObjectStatus, CheckResult,
80 CheckResults, CheckRule, Checker, DataServiceError, EntityDataServiceBehavior, EntityRoot,
81 EntityValues, GraphMutationKind, GraphNode, I18nCatalog, InMemoryAggregationCache,
82 InMemoryCheckerRegistry, InMemoryEntityDataServiceBehaviorRegistry, InMemoryEntityRegistry,
83 InMemoryMetadataStore, InternalIdGenerator, Language, MemoryDataService, MetadataStore,
84 ObjectLocation, RawAuditEvent, RawAuditEventKind, RawAuditEventSink, RemoteLockProvider,
85 RequestPolicy, RuntimeError, RuntimeModule, RuntimeOperation, RuntimeTelemetry,
86 RuntimeTelemetryScope, SafeAuditEvent, SafeAuditEventSink, SqlLogOperation, SqlLogOptions,
87 TypedChecker, TypedEntityChecker, UserContext, translate_check_result,
88 };
89 use crate::data_service::RuntimeDataService;
90 use teaql_core::{
91 Aggregate, AggregateFunction, BinaryOp, DataType, Decimal, DeleteCommand, Entity,
92 EntityDescriptor, EntityError, Expr, GeneratedValues, InsertCommand, OrderBy,
93 PropertyDescriptor, Record, RecoverCommand, RelationAggregate, SelectQuery, TeaqlEntity,
94 UpdateCommand, Value,
95 };
96 use teaql_data_service::{
97 DataServiceCapabilities, DataServiceExecutor, DataServiceOperation, ExecutionMetadata,
98 MutationExecutor, MutationRequest, MutationResult, QueryExecutor, QueryRequest,
99 QueryResult,
100 };
101 use teaql_macros::TeaqlEntity as DeriveTeaqlEntity;
102 use teaql_sql::{
103 CompiledQuery, DatabaseKind, SqlCompileError, SqlDialect, quote_identifier_if_needed,
104 };
105
106 const ORDER_DEFAULT_PROJECTION: &str = "id, version, name";
107
108 #[derive(Debug, Default, Clone, Copy)]
109 struct PostgresDialect;
110
111 impl SqlDialect for PostgresDialect {
112 fn kind(&self) -> DatabaseKind {
113 DatabaseKind::PostgreSql
114 }
115
116 fn quote_ident(&self, ident: &str) -> String {
117 quote_identifier_if_needed(ident, '"')
118 }
119
120 fn placeholder(&self, index: usize) -> String {
121 format!("${index}")
122 }
123
124 fn schema_type_sql(
125 &self,
126 data_type: DataType,
127 _property: &PropertyDescriptor,
128 ) -> Result<&'static str, SqlCompileError> {
129 match data_type {
130 DataType::Bool => Ok("BOOLEAN"),
131 DataType::I64 | DataType::U64 => Ok("BIGINT"),
132 DataType::F64 => Ok("DOUBLE PRECISION"),
133 DataType::Decimal => Ok("NUMERIC"),
134 DataType::Text => Ok("VARCHAR(255)"),
135 DataType::LargeText => Ok("TEXT"),
136 DataType::Json => Ok("JSONB"),
137 DataType::Date => Ok("DATE"),
138 DataType::Timestamp => Ok("TIMESTAMPTZ"),
139 }
140 }
141 }
142
143 fn entity() -> EntityDescriptor {
144 EntityDescriptor::new("Order")
145 .table_name("orders")
146 .property(
147 PropertyDescriptor::new("id", DataType::U64)
148 .column_name("id")
149 .id()
150 .not_null(),
151 )
152 .property(
153 PropertyDescriptor::new("version", DataType::I64)
154 .column_name("version")
155 .version()
156 .not_null(),
157 )
158 .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
159 .relation(
160 teaql_core::RelationDescriptor::new("lines", "OrderLine")
161 .local_key("id")
162 .foreign_key("order_id")
163 .many(),
164 )
165 }
166
167 fn line_entity() -> EntityDescriptor {
168 EntityDescriptor::new("OrderLine")
169 .table_name("orderline")
170 .property(
171 PropertyDescriptor::new("id", DataType::U64)
172 .column_name("id")
173 .id()
174 .not_null(),
175 )
176 .property(
177 PropertyDescriptor::new("version", DataType::I64)
178 .column_name("version")
179 .version(),
180 )
181 .property(
182 PropertyDescriptor::new("order_id", DataType::U64)
183 .column_name("order_id")
184 .not_null(),
185 )
186 .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
187 .property(
188 PropertyDescriptor::new("product_id", DataType::U64)
189 .column_name("product_id")
190 .not_null(),
191 )
192 .relation(
193 teaql_core::RelationDescriptor::new("product", "Product")
194 .local_key("product_id")
195 .foreign_key("id"),
196 )
197 }
198
199 fn product_entity() -> EntityDescriptor {
200 EntityDescriptor::new("Product")
201 .table_name("product")
202 .property(
203 PropertyDescriptor::new("id", DataType::U64)
204 .column_name("id")
205 .id()
206 .not_null(),
207 )
208 .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
209 }
210
211 #[derive(Debug, Default)]
212 struct StubExecutor {
213 affected: u64,
214 rows: Vec<Record>,
215 }
216
217 #[derive(Debug, Default)]
218 struct QueueExecutor {
219 affected: u64,
220 rows: Mutex<VecDeque<Vec<Record>>>,
221 queries: Mutex<Vec<String>>,
222 }
223
224 #[derive(Debug, Default)]
225 struct CapturingQueryExecutor {
226 rows: Vec<Record>,
227 queries: Mutex<Vec<SelectQuery>>,
228 }
229
230 struct OrderBehavior;
231
232 #[allow(dead_code)]
233 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
234 #[teaql(entity = "CatalogProduct", table = "catalog_product")]
235 struct CatalogProductRow {
236 #[teaql(id)]
237 id: u64,
238 name: String,
239 }
240
241 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
242 #[teaql(entity = "OrderAggregate", table = "orders")]
243 struct OrderAggregateDynamic {
244 #[teaql(id)]
245 id: u64,
246 #[teaql(dynamic)]
247 dynamic: BTreeMap<String, Value>,
248 }
249
250 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
251 #[teaql(entity = "Product", table = "product")]
252 struct ProductEntityRow {
253 #[teaql(id)]
254 id: u64,
255 name: String,
256 }
257
258 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
259 #[teaql(entity = "OrderLine", table = "orderline")]
260 struct OrderLineEntityRow {
261 #[teaql(id)]
262 id: u64,
263 #[teaql(column = "order_id")]
264 order_id: u64,
265 name: String,
266 #[teaql(column = "product_id")]
267 product_id: u64,
268 #[teaql(relation(target = "Product", local_key = "product_id", foreign_key = "id"))]
269 product: Option<ProductEntityRow>,
270 }
271
272 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
273 #[teaql(entity = "OrderLine", table = "orderline")]
274 struct ProductLineEntityRow {
275 #[teaql(id)]
276 id: u64,
277 #[teaql(column = "order_id")]
278 order_id: u64,
279 name: String,
280 #[teaql(column = "product_id")]
281 product_id: u64,
282 }
283
284 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
285 #[teaql(entity = "Product", table = "product")]
286 struct ProductWithLinesEntityRow {
287 #[teaql(id)]
288 id: u64,
289 name: String,
290 #[teaql(relation(
291 target = "OrderLine",
292 local_key = "id",
293 foreign_key = "product_id",
294 many
295 ))]
296 lines: teaql_core::SmartList<ProductLineEntityRow>,
297 }
298
299 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
300 #[teaql(entity = "OrderLine", table = "orderline")]
301 struct OrderLineWithProductEntityRow {
302 #[teaql(id)]
303 id: u64,
304 #[teaql(column = "order_id")]
305 order_id: u64,
306 name: String,
307 #[teaql(column = "product_id")]
308 product_id: u64,
309 #[teaql(relation(target = "Product", local_key = "product_id", foreign_key = "id"))]
310 product: Option<ProductWithLinesEntityRow>,
311 }
312
313 #[derive(Debug, DeriveTeaqlEntity)]
314 #[teaql(entity = "FlatVendor", table = "flat_vendor")]
315 struct FlatVendorRow {
316 #[teaql(id)]
317 id: u64,
318 name: String,
319 #[teaql(skip)]
320 root: EntityRoot,
321 }
322
323 #[derive(Debug, DeriveTeaqlEntity)]
324 #[teaql(entity = "FlatTrip", table = "flat_trip")]
325 struct FlatTripRow {
326 #[teaql(id)]
327 id: u64,
328 vendor_id: u64,
329 #[teaql(relation(target = "FlatVendor", local_key = "vendor_id", foreign_key = "id"))]
330 vendor: Option<FlatVendorRow>,
331 #[teaql(skip)]
332 root: EntityRoot,
333 }
334
335 impl FlatTripRow {
336 fn vendor(&self) -> Option<&FlatVendorRow> {
337 self.vendor
338 .as_ref()
339 .or_else(|| self.root.resolve_entity::<FlatVendorRow>(self.vendor_id))
340 }
341 }
342
343 #[derive(Clone, Debug, DeriveTeaqlEntity)]
344 #[teaql(entity = "FlatFleet", table = "flat_fleet")]
345 struct FlatFleetRow {
346 #[teaql(id)]
347 id: u64,
348 #[teaql(relation(
349 target = "FlatFleetTrip",
350 local_key = "id",
351 foreign_key = "fleet_id",
352 many
353 ))]
354 trip_list: teaql_core::SmartList<FlatFleetTripRow>,
355 #[teaql(skip)]
356 root: EntityRoot,
357 }
358
359 impl FlatFleetRow {
360 fn trip_list(&self) -> &teaql_core::SmartList<FlatFleetTripRow> {
361 if self.trip_list.is_loaded {
362 &self.trip_list
363 } else {
364 self.root
365 .resolve_relation_list(Self::ENTITY_NAME, self.id, "trip_list")
366 .unwrap_or(&self.trip_list)
367 }
368 }
369
370 fn trip_list_mut(&mut self) -> &mut teaql_core::SmartList<FlatFleetTripRow> {
371 if !self.trip_list.is_loaded {
372 if let Some(loaded) = self
373 .root
374 .resolve_relation_list(Self::ENTITY_NAME, self.id, "trip_list")
375 .cloned()
376 {
377 self.trip_list = loaded;
378 }
379 }
380 &mut self.trip_list
381 }
382 }
383
384 #[derive(Clone, Debug, DeriveTeaqlEntity)]
385 #[teaql(entity = "FlatFleetTrip", table = "flat_fleet_trip")]
386 struct FlatFleetTripRow {
387 #[teaql(id)]
388 id: u64,
389 fleet_id: u64,
390 name: String,
391 #[teaql(skip)]
392 root: EntityRoot,
393 }
394
395 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
396 #[teaql(entity = "Order", table = "orders")]
397 struct OrderAggregateRow {
398 #[teaql(id)]
399 id: u64,
400 #[teaql(version)]
401 version: i64,
402 name: String,
403 #[teaql(relation(target = "OrderLine", local_key = "id", foreign_key = "order_id", many))]
404 lines: teaql_core::SmartList<OrderLineEntityRow>,
405 }
406
407 #[derive(Debug, Clone, PartialEq, DeriveTeaqlEntity)]
408 #[teaql(entity = "Order", table = "orders")]
409 struct Order {
410 #[teaql(id)]
411 id: u64,
412 #[teaql(version)]
413 version: i64,
414 name: String,
415 }
416
417 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
418 #[teaql(entity = "Product", table = "product")]
419 struct TypedGraphProduct {
420 #[teaql(id)]
421 id: u64,
422 name: String,
423 }
424
425 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
426 #[teaql(entity = "OrderLine", table = "orderline")]
427 struct TypedGraphLine {
428 #[teaql(id)]
429 id: u64,
430 #[teaql(column = "order_id")]
431 order_id: Option<u64>,
432 name: String,
433 #[teaql(column = "product_id")]
434 product_id: Option<u64>,
435 #[teaql(relation(target = "Product", local_key = "product_id", foreign_key = "id"))]
436 product: Option<TypedGraphProduct>,
437 }
438
439 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
440 #[teaql(entity = "Order", table = "orders")]
441 struct TypedGraphOrder {
442 #[teaql(id)]
443 id: u64,
444 #[teaql(version)]
445 version: i64,
446 name: String,
447 #[teaql(relation(target = "OrderLine", local_key = "id", foreign_key = "order_id", many))]
448 lines: teaql_core::SmartList<TypedGraphLine>,
449 }
450
451 #[derive(Debug, PartialEq, Eq)]
452 struct OrderEntity {
453 id: u64,
454 version: i64,
455 name: String,
456 }
457
458 impl teaql_core::TeaqlEntity for OrderEntity {
459 const ENTITY_NAME: &'static str = "Order";
460
461 fn entity_descriptor() -> EntityDescriptor {
462 entity()
463 }
464 }
465
466 impl Entity for OrderEntity {
467 fn from_compact_row(row: teaql_core::CompactRow) -> Result<Self, EntityError> {
468 let record = row.into_map();
469 let id = match record.get("id") {
470 Some(Value::U64(v)) => *v,
471 Some(Value::I64(v)) if *v >= 0 => *v as u64,
472 other => {
473 return Err(EntityError::new(
474 "Order",
475 format!("invalid id field: {other:?}"),
476 ));
477 }
478 };
479 let version = match record.get("version") {
480 Some(Value::I64(v)) => *v,
481 other => {
482 return Err(EntityError::new(
483 "Order",
484 format!("invalid version field: {other:?}"),
485 ));
486 }
487 };
488 let name = match record.get("name") {
489 Some(Value::Text(v)) => v.clone(),
490 other => {
491 return Err(EntityError::new(
492 "Order",
493 format!("invalid name field: {other:?}"),
494 ));
495 }
496 };
497 Ok(Self { id, version, name })
498 }
499
500 fn into_values(self) -> teaql_core::MutationValues {
501 Record::from([
502 (String::from("id"), Value::U64(self.id)),
503 (String::from("version"), Value::I64(self.version)),
504 (String::from("name"), Value::Text(self.name)),
505 ]).into()
506 }
507 }
508
509 #[derive(Debug)]
510 struct StubError;
511
512 struct RecordingRuntimeTelemetry(Arc<Mutex<Vec<String>>>);
513
514 impl RuntimeTelemetry for RecordingRuntimeTelemetry {
515 fn start(&self, operation: RuntimeOperation) -> Box<dyn RuntimeTelemetryScope> {
516 self.0
517 .lock()
518 .unwrap()
519 .push(format!("start:{}", operation.family));
520 Box::new(RecordingRuntimeTelemetryScope(self.0.clone()))
521 }
522 }
523
524 struct RecordingRuntimeTelemetryScope(Arc<Mutex<Vec<String>>>);
525
526 impl RuntimeTelemetryScope for RecordingRuntimeTelemetryScope {
527 fn success(&mut self, _attributes: BTreeMap<String, crate::RuntimeAttributeValue>) {
528 self.0.lock().unwrap().push("success".to_owned());
529 }
530
531 fn failure(&mut self, _error_type: &str) {
532 self.0.lock().unwrap().push("failure".to_owned());
533 }
534 }
535
536 impl std::fmt::Display for StubError {
537 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
538 write!(f, "stub error")
539 }
540 }
541
542 impl std::error::Error for StubError {}
543
544 impl DataServiceExecutor for StubExecutor {
545 type Error = StubError;
546
547 fn capabilities(&self) -> DataServiceCapabilities {
548 DataServiceCapabilities::default()
549 }
550 }
551
552 impl QueryExecutor for StubExecutor {
553 async fn query(&self, _request: QueryRequest) -> Result<QueryResult, Self::Error> {
554 Ok(QueryResult {
555 rows: self.rows.clone().into_iter().map(teaql_core::CompactRow::from_map).collect(),
556 metadata: ExecutionMetadata {
557 debug_query: None,
558 backend: "stub".to_owned(),
559 operation: DataServiceOperation::Query,
560 started_at: std::time::SystemTime::now(),
561 ended_at: std::time::SystemTime::now(),
562 affected_rows: None,
563 result_count: Some(self.rows.len()),
564 trace_chain: Vec::new(),
565 comment: None,
566 backend_request_id: None,
567 parameterized_query: None,
568 params: Vec::new(),
569 },
570 })
571 }
572 }
573
574 impl MutationExecutor for StubExecutor {
575 async fn mutate(&self, _request: MutationRequest) -> Result<MutationResult, Self::Error> {
576 Ok(MutationResult {
577 affected_rows: self.affected,
578 generated_values: GeneratedValues::new(),
579 persisted_snapshot: None,
580 metadata: ExecutionMetadata {
581 debug_query: None,
582 backend: "stub".to_owned(),
583 operation: DataServiceOperation::Update,
584 started_at: std::time::SystemTime::now(),
585 ended_at: std::time::SystemTime::now(),
586 affected_rows: Some(self.affected),
587 result_count: None,
588 trace_chain: Vec::new(),
589 comment: None,
590 backend_request_id: None,
591 parameterized_query: None,
592 params: Vec::new(),
593 },
594 })
595 }
596 }
597
598 impl DataServiceExecutor for CapturingQueryExecutor {
599 type Error = StubError;
600
601 fn capabilities(&self) -> DataServiceCapabilities {
602 DataServiceCapabilities::default()
603 }
604 }
605
606 impl QueryExecutor for CapturingQueryExecutor {
607 async fn query(&self, request: QueryRequest) -> Result<QueryResult, Self::Error> {
608 self.queries.lock().unwrap().push(request.query);
609 Ok(QueryResult {
610 rows: self.rows.clone().into_iter().map(teaql_core::CompactRow::from_map).collect(),
611 metadata: ExecutionMetadata {
612 debug_query: None,
613 backend: "capture".to_owned(),
614 operation: DataServiceOperation::Query,
615 started_at: std::time::SystemTime::now(),
616 ended_at: std::time::SystemTime::now(),
617 affected_rows: None,
618 result_count: Some(self.rows.len()),
619 trace_chain: Vec::new(),
620 comment: None,
621 backend_request_id: None,
622 parameterized_query: None,
623 params: Vec::new(),
624 },
625 })
626 }
627 }
628
629 impl MutationExecutor for CapturingQueryExecutor {
630 async fn mutate(&self, _request: MutationRequest) -> Result<MutationResult, Self::Error> {
631 unreachable!("relation query test does not mutate")
632 }
633 }
634
635 impl DataServiceExecutor for QueueExecutor {
636 type Error = StubError;
637
638 fn capabilities(&self) -> DataServiceCapabilities {
639 DataServiceCapabilities::default()
640 }
641 }
642
643 impl QueryExecutor for QueueExecutor {
644 async fn query(&self, request: QueryRequest) -> Result<QueryResult, Self::Error> {
645 let sql_approx = format!("SELECT ... FROM {} ...", request.query.entity);
646 self.queries.lock().unwrap().push(sql_approx);
647 Ok(QueryResult {
648 rows: self.rows.lock().unwrap().pop_front().unwrap_or_default()
649 .into_iter().map(teaql_core::CompactRow::from_map).collect(),
650 metadata: ExecutionMetadata {
651 debug_query: None,
652 backend: "queue".to_owned(),
653 operation: DataServiceOperation::Query,
654 started_at: std::time::SystemTime::now(),
655 ended_at: std::time::SystemTime::now(),
656 affected_rows: None,
657 result_count: Some(0),
658 trace_chain: Vec::new(),
659 comment: None,
660 backend_request_id: None,
661 parameterized_query: None,
662 params: Vec::new(),
663 },
664 })
665 }
666 }
667
668 impl MutationExecutor for QueueExecutor {
669 async fn mutate(&self, _request: MutationRequest) -> Result<MutationResult, Self::Error> {
670 Ok(MutationResult {
671 affected_rows: self.affected,
672 generated_values: GeneratedValues::new(),
673 persisted_snapshot: None,
674 metadata: ExecutionMetadata {
675 debug_query: None,
676 backend: "queue".to_owned(),
677 operation: DataServiceOperation::Update,
678 started_at: std::time::SystemTime::now(),
679 ended_at: std::time::SystemTime::now(),
680 affected_rows: Some(self.affected),
681 result_count: None,
682 trace_chain: Vec::new(),
683 comment: None,
684 backend_request_id: None,
685 parameterized_query: None,
686 params: Vec::new(),
687 },
688 })
689 }
690 }
691
692 impl EntityDataServiceBehavior for OrderBehavior {
693 fn before_select(
694 &self,
695 _ctx: &UserContext,
696 query: &mut teaql_core::SelectQuery,
697 ) -> Result<(), RuntimeError> {
698 query.filter = Some(Expr::eq("version", 1_i64));
699 Ok(())
700 }
701
702 fn before_insert(
703 &self,
704 _ctx: &UserContext,
705 command: &mut InsertCommand,
706 ) -> Result<(), RuntimeError> {
707 command
708 .values
709 .entry("version".to_owned())
710 .or_insert(Value::I64(1));
711 Ok(())
712 }
713
714 fn relation_loads(&self, _ctx: &UserContext) -> Vec<String> {
715 vec!["lines".to_owned()]
716 }
717 }
718
719 struct ContextAwareOrderBehavior;
720 struct TenantRequestPolicy;
721 struct OrderChecker;
722 struct TypedOrderChecker;
723 #[derive(Clone)]
724 struct RecordingEventSink {
725 events: Arc<Mutex<Vec<RawAuditEvent>>>,
726 }
727 #[derive(Clone)]
728 struct RecordingSafeEventSink {
729 events: Arc<Mutex<Vec<SafeAuditEvent>>>,
730 }
731
732 impl EntityDataServiceBehavior for ContextAwareOrderBehavior {
733 fn before_insert(
734 &self,
735 context: &UserContext,
736 command: &mut InsertCommand,
737 ) -> Result<(), RuntimeError> {
738 let tenant = context
739 .get_named_resource::<String>("tenant")
740 .cloned()
741 .ok_or_else(|| RuntimeError::Behavior("missing tenant resource".to_owned()))?;
742 let version = *context
743 .get_named_resource::<i64>("initial_version")
744 .ok_or_else(|| {
745 RuntimeError::Behavior("missing initial_version resource".to_owned())
746 })?;
747 let trace_id = match context.local("trace_id") {
748 Some(Value::Text(value)) => value.clone(),
749 other => {
750 return Err(RuntimeError::Behavior(format!(
751 "missing trace_id local, got {other:?}"
752 )));
753 }
754 };
755
756 command
757 .values
758 .entry("name".to_owned())
759 .or_insert(Value::Text(format!("{tenant}:{trace_id}")));
760 command
761 .values
762 .entry("version".to_owned())
763 .or_insert(Value::I64(version));
764 Ok(())
765 }
766 }
767
768 impl RequestPolicy for TenantRequestPolicy {
769 fn enforce_select(
770 &self,
771 context: &UserContext,
772 query: &mut SelectQuery,
773 ) -> Result<(), RuntimeError> {
774 if query.entity == "Order" {
775 let tenant_id = context
776 .get_named_resource::<u64>("tenant_id")
777 .copied()
778 .ok_or_else(|| RuntimeError::Policy("missing tenant_id".to_owned()))?;
779 query.filter = Some(match query.filter.take() {
780 Some(filter) => filter.and_expr(Expr::eq("id", tenant_id)),
781 None => Expr::eq("id", tenant_id),
782 });
783 }
784 Ok(())
785 }
786
787 fn enforce_insert(
788 &self,
789 context: &UserContext,
790 command: &mut InsertCommand,
791 ) -> Result<(), RuntimeError> {
792 if command.entity == "Order" {
793 let tenant_id = context
794 .get_named_resource::<u64>("tenant_id")
795 .copied()
796 .ok_or_else(|| RuntimeError::Policy("missing tenant_id".to_owned()))?;
797 command
798 .values
799 .insert("version".to_owned(), Value::I64(tenant_id as i64));
800 }
801 Ok(())
802 }
803 }
804
805 impl Checker for OrderChecker {
806 fn entity(&self) -> &str {
807 "Order"
808 }
809
810 fn check_and_fix(
811 &self,
812 _ctx: &UserContext,
813 values: &mut EntityValues,
814 location: &ObjectLocation,
815 results: &mut CheckResults,
816 ) {
817 let status = CheckObjectStatus::from_values(values);
818 if status.is_create() {
819 self.required(values, "name", location, results);
820 values.entry("version".to_owned()).or_insert(Value::I64(1));
821 }
822 if status.is_update()
823 && values.get("name") == Some(&Value::Text("graph-update".to_owned()))
824 {
825 values.insert(
826 "name".to_owned(),
827 Value::Text("graph-update-checked".to_owned()),
828 );
829 }
830 self.min_string_length(values, "name", 3, location, results);
831 }
832 }
833
834 impl TypedChecker<Order> for TypedOrderChecker {
835 fn check_and_fix_typed(
836 &self,
837 _ctx: &UserContext,
838 entity: &mut Order,
839 status: CheckObjectStatus,
840 location: &ObjectLocation,
841 results: &mut CheckResults,
842 ) {
843 if status.is_create() {
844 if entity.name.is_empty() {
845 results.push(CheckResult::required(location.clone().member("name")));
846 }
847 }
848 if entity.name.chars().count() < 3 {
849 results.push(CheckResult::min_str(
850 location.clone().member("name"),
851 3,
852 entity.name.clone(),
853 ));
854 }
855 if entity.name == "fix" {
856 entity.name = "fixed".to_owned();
857 }
858 }
859 }
860
861 impl RawAuditEventSink for RecordingEventSink {
862 fn on_event(&self, _ctx: &UserContext, event: &RawAuditEvent) -> Result<(), RuntimeError> {
863 self.events.lock().unwrap().push(event.clone());
864 Ok(())
865 }
866 }
867
868 impl SafeAuditEventSink for RecordingSafeEventSink {
869 fn on_safe_event(
870 &self,
871 _ctx: &UserContext,
872 event: &SafeAuditEvent,
873 ) -> Result<(), RuntimeError> {
874 self.events.lock().unwrap().push(event.clone());
875 Ok(())
876 }
877 }
878
879 struct FixedIdGenerator(u64);
880
881 impl InternalIdGenerator for FixedIdGenerator {
882 fn generate_id(&self, _entity: &str) -> Result<u64, RuntimeError> {
883 Ok(self.0)
884 }
885 }
886
887 struct SequentialIdGenerator {
888 next: Mutex<u64>,
889 }
890
891 impl SequentialIdGenerator {
892 fn new(next: u64) -> Self {
893 Self {
894 next: Mutex::new(next),
895 }
896 }
897 }
898
899 impl InternalIdGenerator for SequentialIdGenerator {
900 fn generate_id(&self, _entity: &str) -> Result<u64, RuntimeError> {
901 let mut next = self
902 .next
903 .lock()
904 .map_err(|err| RuntimeError::IdGeneration(err.to_string()))?;
905 let id = *next;
906 *next += 1;
907 Ok(id)
908 }
909 }
910
911 #[tokio::test]
912 async fn metadata_store_registers_entities() {
913 let store = InMemoryMetadataStore::new().with_entity(entity());
914 assert!(store.entity("Order").is_some());
915 }
916
917 #[tokio::test]
918 async fn runtime_module_registers_descriptor_into_context() {
919 let context = UserContext::new().with_module(RuntimeModule::new().descriptor(entity()));
920 assert!(context.entity("Order").is_some());
921 assert!(context.has_entity_data_service("Order"));
922 }
923
924 #[tokio::test]
925 async fn runtime_module_registers_derived_entity_and_behavior() {
926 let context = UserContext::new().with_module(
927 RuntimeModule::new().entity_with_behavior::<CatalogProductRow, _>(OrderBehavior),
928 );
929 assert!(context.entity("CatalogProduct").is_some());
930 assert!(context.has_entity_data_service("CatalogProduct"));
931 assert!(
932 context
933 .entity_data_service_behavior("CatalogProduct")
934 .is_some()
935 );
936 }
937
938 #[tokio::test]
939 async fn module_macro_registers_multiple_entities() {
940 let context = UserContext::new().with_module(crate::module!(CatalogProductRow));
941 assert!(context.entity("CatalogProduct").is_some());
942 assert!(context.has_entity_data_service("CatalogProduct"));
943 }
944
945 #[tokio::test]
946 async fn module_macro_registers_entity_behavior_pairs() {
947 let context =
948 UserContext::new().with_module(crate::module!(CatalogProductRow => OrderBehavior));
949 assert!(context.entity("CatalogProduct").is_some());
950 assert!(
951 context
952 .entity_data_service_behavior("CatalogProduct")
953 .is_some()
954 );
955 }
956
957 #[tokio::test]
958 async fn data_service_returns_optimistic_lock_conflict() {
959 let store = InMemoryMetadataStore::new().with_entity(entity());
960 let executor = StubExecutor {
961 affected: 0,
962 rows: Vec::new(),
963 };
964 let repo = RuntimeDataService::new(&store, &executor);
965
966 let err = repo
967 .update(
968 &UpdateCommand::new("Order", 1_u64)
969 .expected_version(3)
970 .value("name", "next"),
971 )
972 .await
973 .unwrap_err();
974
975 match err {
976 DataServiceError::Runtime(RuntimeError::OptimisticLockConflict { .. }) => {}
977 other => panic!("unexpected error: {other}"),
978 }
979 }
980
981 #[tokio::test]
982 async fn user_context_indexes_resources_and_locals() {
983 let mut context =
984 UserContext::new().with_metadata(InMemoryMetadataStore::new().with_entity(entity()));
985 context.insert_resource::<u64>(42);
986 context.insert_named_resource("tenant", String::from("acme"));
987 context.put_local("trace_id", "req-1");
988
989 assert!(context.entity("Order").is_some());
990 assert_eq!(context.get_resource::<u64>(), Some(&42));
991 assert_eq!(
992 context.get_named_resource::<String>("tenant"),
993 Some(&String::from("acme"))
994 );
995 assert_eq!(
996 context.local("trace_id"),
997 Some(&Value::Text("req-1".to_owned()))
998 );
999 }
1000
1001 #[tokio::test]
1002 async fn user_context_builds_context_data_service() {
1003 let telemetry_events = Arc::new(Mutex::new(Vec::new()));
1004 let mut context = UserContext::new()
1005 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1006 .with_runtime_telemetry(Arc::new(RecordingRuntimeTelemetry(
1007 telemetry_events.clone(),
1008 )));
1009 context.insert_resource(PostgresDialect);
1010 context.insert_resource(StubExecutor {
1011 affected: 1,
1012 rows: Vec::new(),
1013 });
1014
1015 let repo = context.data_service_internal::<StubExecutor>().unwrap();
1016 let affected = repo
1017 .update(
1018 &UpdateCommand::new("Order", 1_u64)
1019 .expected_version(3)
1020 .value("name", "next"),
1021 )
1022 .await
1023 .unwrap();
1024
1025 assert_eq!(affected, 1);
1026 assert_eq!(
1027 telemetry_events.lock().unwrap().as_slice(),
1028 ["start:mutation", "start:provider", "success", "success"]
1029 );
1030 }
1031
1032 #[tokio::test]
1033 async fn user_context_resolves_entity_data_service_by_entity_type() {
1034 let mut context = UserContext::new()
1035 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1036 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1037 context.insert_resource(PostgresDialect);
1038 context.insert_resource(StubExecutor {
1039 affected: 1,
1040 rows: Vec::new(),
1041 });
1042
1043 let repo = context
1044 .entity_data_service::<StubExecutor>("Order")
1045 .unwrap();
1046 assert_eq!(repo.entity(), "Order");
1047 assert_eq!(repo.select().entity, "Order");
1048
1049 let affected = repo
1050 .insert_internal(
1051 &repo
1052 .insert_command()
1053 .value("id", 1_u64)
1054 .value("version", 1_i64)
1055 .value("name", "n"),
1056 )
1057 .await
1058 .unwrap();
1059 assert_eq!(affected, 1);
1060 }
1061
1062 #[tokio::test]
1063 async fn entity_data_service_applies_behavior_hooks() {
1064 let mut context = UserContext::new()
1065 .with_metadata(
1066 InMemoryMetadataStore::new()
1067 .with_entity(entity())
1068 .with_entity(line_entity())
1069 .with_entity(product_entity()),
1070 )
1071 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1072 .with_entity_data_service_behavior_registry(
1073 InMemoryEntityDataServiceBehaviorRegistry::new()
1074 .with_behavior("Order", OrderBehavior),
1075 );
1076 context.insert_resource(PostgresDialect);
1077 context.insert_resource(StubExecutor {
1078 affected: 1,
1079 rows: Vec::new(),
1080 });
1081
1082 let repo = context
1083 .entity_data_service::<StubExecutor>("Order")
1084 .unwrap();
1085
1086 let insert = repo.insert_command().value("id", 1_u64).value("name", "n");
1090 let affected = repo.insert_internal(&insert).await.unwrap();
1091 assert_eq!(affected, 1);
1092 assert_eq!(repo.relation_loads(), vec!["lines".to_owned()]);
1093 }
1094
1095 #[tokio::test]
1096 async fn entity_data_service_applies_request_policy_after_behavior_hooks() {
1097 let mut context = UserContext::new()
1098 .with_metadata(
1099 InMemoryMetadataStore::new()
1100 .with_entity(entity())
1101 .with_entity(line_entity())
1102 .with_entity(product_entity()),
1103 )
1104 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1105 .with_entity_data_service_behavior_registry(
1106 InMemoryEntityDataServiceBehaviorRegistry::new()
1107 .with_behavior("Order", OrderBehavior),
1108 )
1109 .with_request_policy(TenantRequestPolicy);
1110 context.insert_named_resource("tenant_id", 9_u64);
1111 context.insert_resource(PostgresDialect);
1112 context.insert_resource(StubExecutor {
1113 affected: 1,
1114 rows: Vec::new(),
1115 });
1116
1117 let repo = context
1118 .entity_data_service::<StubExecutor>("Order")
1119 .unwrap();
1120
1121 let insert = repo.insert_command().value("id", 1_u64).value("name", "n");
1126 let command = repo.prepare_insert_command(&insert).unwrap();
1127 assert_eq!(command.values.get("version"), Some(&Value::I64(9)));
1128 }
1129
1130 #[tokio::test]
1131 async fn entity_data_service_prepares_insert_command_with_generated_id() {
1132 let mut context = UserContext::new()
1133 .with_metadata(
1134 InMemoryMetadataStore::new()
1135 .with_entity(entity())
1136 .with_entity(line_entity())
1137 .with_entity(product_entity()),
1138 )
1139 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1140 .with_entity_data_service_behavior_registry(
1141 InMemoryEntityDataServiceBehaviorRegistry::new()
1142 .with_behavior("Order", OrderBehavior),
1143 )
1144 .with_internal_id_generator(FixedIdGenerator(42));
1145 context.insert_resource(PostgresDialect);
1146 context.insert_resource(StubExecutor {
1147 affected: 1,
1148 rows: Vec::new(),
1149 });
1150
1151 let repo = context
1152 .entity_data_service::<StubExecutor>("Order")
1153 .unwrap();
1154
1155 let prepared = repo
1156 .prepare_insert_command(&repo.insert_command().value("id", 0_u64).value("name", "n"))
1157 .unwrap();
1158
1159 assert_eq!(prepared.values.get("id"), Some(&Value::U64(42)));
1160 assert_eq!(prepared.values.get("version"), Some(&Value::I64(1)));
1161 assert_eq!(
1162 prepared.values.get("name"),
1163 Some(&Value::Text("n".to_owned()))
1164 );
1165
1166 let prepared_zero_version = repo
1167 .prepare_insert_command(
1168 &repo
1169 .insert_command()
1170 .value("id", 0_u64)
1171 .value("version", 0_i64)
1172 .value("name", "zero-version"),
1173 )
1174 .unwrap();
1175 assert_eq!(
1176 prepared_zero_version.values.get("version"),
1177 Some(&Value::I64(1))
1178 );
1179 }
1180
1181 #[tokio::test]
1182 async fn custom_user_context_can_drive_insert_preparation() {
1183 let mut context = UserContext::new()
1184 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1185 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1186 .with_entity_data_service_behavior_registry(
1187 InMemoryEntityDataServiceBehaviorRegistry::new()
1188 .with_behavior("Order", ContextAwareOrderBehavior),
1189 )
1190 .with_internal_id_generator(FixedIdGenerator(99));
1191 context.insert_named_resource("tenant", String::from("acme"));
1192 context.insert_named_resource("initial_version", 7_i64);
1193 context.put_local("trace_id", "req-9");
1194 context.insert_resource(PostgresDialect);
1195 context.insert_resource(StubExecutor {
1196 affected: 1,
1197 rows: Vec::new(),
1198 });
1199
1200 let repo = context
1201 .entity_data_service::<StubExecutor>("Order")
1202 .unwrap();
1203 let prepared = repo.prepare_insert_command(&repo.insert_command()).unwrap();
1204
1205 assert_eq!(prepared.values.get("id"), Some(&Value::U64(99)));
1206 assert_eq!(prepared.values.get("version"), Some(&Value::I64(7)));
1207 assert_eq!(
1208 prepared.values.get("name"),
1209 Some(&Value::Text("acme:req-9".to_owned()))
1210 );
1211 }
1212
1213 #[tokio::test]
1214 async fn checker_registry_validates_and_fixes_insert_commands() {
1215 let mut context = UserContext::new()
1216 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1217 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1218 .with_checker_registry(InMemoryCheckerRegistry::new().with_checker(OrderChecker))
1219 .with_internal_id_generator(FixedIdGenerator(77));
1220 context.insert_resource(PostgresDialect);
1221 context.insert_resource(StubExecutor {
1222 affected: 1,
1223 rows: Vec::new(),
1224 });
1225
1226 let repo = context
1227 .entity_data_service::<StubExecutor>("Order")
1228 .unwrap();
1229 let prepared = repo
1230 .prepare_insert_command(&repo.insert_command().value("name", "valid"))
1231 .unwrap();
1232
1233 assert_eq!(prepared.values.get("id"), Some(&Value::U64(77)));
1234 assert_eq!(prepared.values.get("version"), Some(&Value::I64(1)));
1235 assert!(!prepared.values.contains_key(CHECK_OBJECT_STATUS_FIELD));
1236
1237 let error = repo
1238 .prepare_insert_command(&repo.insert_command().value("name", "no"))
1239 .unwrap_err();
1240 match error {
1241 RuntimeError::Check(results) => {
1242 assert_eq!(results.len(), 1);
1243 assert_eq!(results[0].location.to_string(), "name");
1244 }
1245 other => panic!("unexpected checker error: {other:?}"),
1246 }
1247 }
1248
1249 #[test]
1250 fn metadata_not_null_constraints_are_checked_without_a_custom_checker() {
1251 let context = UserContext::new().with_metadata(
1252 InMemoryMetadataStore::new().with_entity(
1253 EntityDescriptor::new("School")
1254 .property(PropertyDescriptor::new("id", DataType::U64).id().not_null())
1255 .property(PropertyDescriptor::new("contact_phone", DataType::Text).not_null()),
1256 ),
1257 );
1258 let mut values = EntityValues::from(Record::from([
1259 ("id".to_owned(), Value::U64(1)),
1260 (
1261 CHECK_OBJECT_STATUS_FIELD.to_owned(),
1262 Value::from(CheckObjectStatus::Create),
1263 ),
1264 ]));
1265
1266 let error = context
1267 .check_and_fix_values("School", &mut values)
1268 .unwrap_err();
1269
1270 match error {
1271 RuntimeError::Check(results) => {
1272 assert_eq!(results.len(), 1);
1273 assert_eq!(results[0].rule, CheckRule::Required);
1274 assert_eq!(results[0].location.to_string(), "contact_phone");
1275 }
1276 other => panic!("unexpected validation error: {other:?}"),
1277 }
1278 }
1279
1280 #[test]
1281 fn metadata_not_null_constraints_allow_omitted_fields_on_partial_update() {
1282 let context = UserContext::new().with_metadata(
1283 InMemoryMetadataStore::new().with_entity(
1284 EntityDescriptor::new("School")
1285 .property(PropertyDescriptor::new("id", DataType::U64).id().not_null())
1286 .property(PropertyDescriptor::new("contact_phone", DataType::Text).not_null()),
1287 ),
1288 );
1289 let mut values = EntityValues::from(Record::from([
1290 ("id".to_owned(), Value::U64(1)),
1291 (
1292 CHECK_OBJECT_STATUS_FIELD.to_owned(),
1293 Value::from(CheckObjectStatus::Update),
1294 ),
1295 ]));
1296
1297 context.check_and_fix_values("School", &mut values).unwrap();
1298
1299 values.insert("contact_phone".to_owned(), Value::Null);
1300 assert!(matches!(
1301 context.check_and_fix_values("School", &mut values),
1302 Err(RuntimeError::Check(_))
1303 ));
1304 }
1305
1306 #[tokio::test]
1307 async fn typed_checker_validates_and_fixes_derived_entities_without_record_access() {
1308 let mut context = UserContext::new()
1309 .with_metadata(InMemoryMetadataStore::new().with_entity(Order::entity_descriptor()))
1310 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1311 .with_checker_registry(
1312 InMemoryCheckerRegistry::new()
1313 .with_checker(TypedEntityChecker::<Order, _>::new(TypedOrderChecker)),
1314 )
1315 .with_internal_id_generator(FixedIdGenerator(79));
1316 context.insert_resource(PostgresDialect);
1317 context.insert_resource(StubExecutor {
1318 affected: 1,
1319 rows: Vec::new(),
1320 });
1321
1322 let repo = context
1323 .entity_data_service::<StubExecutor>("Order")
1324 .unwrap();
1325 let prepared = repo
1326 .prepare_insert_command(
1327 &repo
1328 .insert_command()
1329 .value("name", "fix")
1330 .value("version", 1_i64),
1331 )
1332 .unwrap();
1333 assert_eq!(
1334 prepared.values.get("name"),
1335 Some(&Value::Text("fixed".to_owned()))
1336 );
1337 assert_eq!(prepared.values.get("id"), Some(&Value::U64(79)));
1338 assert!(!prepared.values.contains_key(CHECK_OBJECT_STATUS_FIELD));
1339
1340 let error = repo
1341 .prepare_insert_command(&repo.insert_command().value("version", 1_i64))
1342 .unwrap_err();
1343 match error {
1344 RuntimeError::Check(results) => {
1345 assert!(
1346 results
1347 .iter()
1348 .any(|result| result.rule == CheckRule::Required
1349 && result.location.to_string() == "name")
1350 );
1351 }
1352 other => panic!("unexpected typed checker error: {other:?}"),
1353 }
1354 }
1355
1356 #[tokio::test]
1357 async fn checker_registry_reports_nested_create_locations_and_fixes_records() {
1358 let context = UserContext::new()
1359 .with_checker_registry(InMemoryCheckerRegistry::new().with_checker(OrderChecker));
1360
1361 let mut child = EntityValues::from(Record::from([
1362 (String::from("id"), Value::U64(10)),
1363 (
1364 String::from(CHECK_OBJECT_STATUS_FIELD),
1365 Value::from(CheckObjectStatus::Create),
1366 ),
1367 ]));
1368 let error = context
1369 .check_and_fix_values_at(
1370 "Order",
1371 &mut child,
1372 &ObjectLocation::hash_root("lines").element(0),
1373 )
1374 .unwrap_err();
1375
1376 assert_eq!(child.get("version"), Some(&Value::I64(1)));
1377 match error {
1378 RuntimeError::Check(results) => {
1379 assert_eq!(results.len(), 1);
1380 assert_eq!(results[0].rule, CheckRule::Required);
1381 assert_eq!(results[0].location.to_string(), "lines[0].name");
1382 }
1383 other => panic!("unexpected checker error: {other:?}"),
1384 }
1385
1386 child.insert("name".to_owned(), Value::Text("valid child".to_owned()));
1387 context
1388 .check_and_fix_values_at(
1389 "Order",
1390 &mut child,
1391 &ObjectLocation::hash_root("lines").element(0),
1392 )
1393 .unwrap();
1394 }
1395
1396 #[tokio::test]
1397 async fn built_in_language_translators_cover_fifteen_languages() {
1398 assert_eq!(Language::ALL.len(), 15);
1399 let results = [
1400 super::CheckResult::required(ObjectLocation::hash_root("name")),
1401 super::CheckResult::min(ObjectLocation::hash_root("age"), 18_i64, 12_i64),
1402 super::CheckResult::max(ObjectLocation::hash_root("age"), 65_i64, 70_i64),
1403 super::CheckResult::min_str(ObjectLocation::hash_root("name"), 2, "x"),
1404 super::CheckResult::max_str(ObjectLocation::hash_root("name"), 8, "too long name"),
1405 ];
1406 let messages = Language::ALL
1407 .iter()
1408 .flat_map(|language| {
1409 results
1410 .iter()
1411 .map(|result| translate_check_result(*language, result))
1412 })
1413 .collect::<Vec<_>>();
1414
1415 assert_eq!(messages.len(), 75);
1416 assert!(messages.iter().all(|message| !message.is_empty()));
1417 assert!(messages.iter().all(|message| !message.contains('{')));
1418 assert!(messages.iter().any(|message| message.contains("required")));
1419 assert!(messages.iter().any(|message| message.contains("å¿…å¡«")));
1420 assert!(
1421 messages
1422 .iter()
1423 .any(|message| message.contains("obligatoire"))
1424 );
1425 assert_eq!(Language::from_code("zh-CN"), Some(Language::Chinese));
1426 assert_eq!(
1427 Language::from_code("zh-TW"),
1428 Some(Language::TraditionalChinese)
1429 );
1430 }
1431
1432 #[tokio::test]
1433 async fn user_context_language_switch_translates_checker_errors() {
1434 let mut context = UserContext::new()
1435 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1436 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1437 .with_checker_registry(InMemoryCheckerRegistry::new().with_checker(OrderChecker))
1438 .with_internal_id_generator(FixedIdGenerator(77))
1439 .with_language(Language::Chinese);
1440 context.insert_resource(PostgresDialect);
1441 context.insert_resource(StubExecutor {
1442 affected: 1,
1443 rows: Vec::new(),
1444 });
1445
1446 let repo = context
1447 .entity_data_service::<StubExecutor>("Order")
1448 .unwrap();
1449 let error = repo
1450 .prepare_insert_command(&repo.insert_command())
1451 .unwrap_err();
1452 match error {
1453 RuntimeError::Check(results) => {
1454 assert_eq!(results.len(), 1);
1455 assert!(
1456 results[0]
1457 .message
1458 .as_ref()
1459 .is_some_and(|message| message.contains("å¿…å¡«"))
1460 );
1461 }
1462 other => panic!("unexpected checker error: {other:?}"),
1463 }
1464
1465 let mut context = UserContext::new().with_language(Language::English);
1466 context.set_language_code("es").unwrap();
1467 assert_eq!(context.language(), Language::Spanish);
1468 assert!(context.set_locale_code("invalid-code").is_err());
1469 assert_eq!(context.language(), Language::Spanish);
1470
1471 let catalog = I18nCatalog::from_json(
1472 r#"{
1473 "schema":"teaql.i18n/v1",
1474 "defaultLocale":"en",
1475 "locales":{
1476 "en":{"messages":{"checker.required":"EN {location}"},"vocabulary":{}},
1477 "es":{"messages":{"checker.required":"ES {location}"},"vocabulary":{}}
1478 }
1479 }"#,
1480 )
1481 .unwrap();
1482 context.set_i18n_catalog(Arc::new(catalog));
1483 let mut results = vec![super::CheckResult::required(ObjectLocation::hash_root(
1484 "name",
1485 ))];
1486 context.translate_check_results(&mut results);
1487 assert_eq!(results[0].message.as_deref(), Some("ES Name"));
1488 }
1489
1490 #[tokio::test]
1491 async fn user_context_event_sink_receives_data_service_mutation_events() {
1492 let events = Arc::new(Mutex::new(Vec::new()));
1493 let safe_events = Arc::new(Mutex::new(Vec::new()));
1494 let mut context = UserContext::new()
1495 .with_metadata(
1496 InMemoryMetadataStore::new()
1497 .with_entity(entity().audit_mask_fields(vec!["name".to_owned()])),
1498 )
1499 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1500 .with_internal_id_generator(FixedIdGenerator(88))
1501 .with_event_sink(RecordingEventSink {
1502 events: events.clone(),
1503 })
1504 .with_custom_event_sink(RecordingSafeEventSink {
1505 events: safe_events.clone(),
1506 });
1507 context.insert_resource(PostgresDialect);
1508 context.insert_resource(StubExecutor {
1509 affected: 1,
1510 rows: vec![Record::from([
1511 ("id".to_owned(), Value::U64(88)),
1512 ("version".to_owned(), Value::I64(1)),
1513 ("name".to_owned(), Value::Text("old".to_owned())),
1514 ])],
1515 });
1516
1517 let repo = context
1518 .entity_data_service::<StubExecutor>("Order")
1519 .unwrap();
1520 repo.insert_internal(&repo.insert_command().value("name", "created"))
1521 .await
1522 .unwrap();
1523 repo.update_internal(
1524 &repo
1525 .update_command(88_u64)
1526 .expected_version(1)
1527 .value("name", "updated"),
1528 )
1529 .await
1530 .unwrap();
1531 repo.delete_internal(&repo.delete_command(88_u64).expected_version(2))
1532 .await
1533 .unwrap();
1534 repo.recover_internal(&repo.recover_command(88_u64, -3))
1535 .await
1536 .unwrap();
1537
1538 let events = events.lock().unwrap();
1539 assert_eq!(events.len(), 4);
1540 assert_eq!(events[0].kind, RawAuditEventKind::Created);
1541 assert_eq!(events[0].entity, "Order");
1542 assert_eq!(events[0].values.get("id"), Some(&Value::U64(88)));
1543 assert_eq!(events[1].kind, RawAuditEventKind::Updated);
1544 assert_eq!(events[1].values.get("id"), Some(&Value::U64(88)));
1545 assert_eq!(events[1].values.get("version"), Some(&Value::I64(2)));
1546 assert_eq!(events[1].updated_fields, vec!["name".to_owned()]);
1547 assert_eq!(
1548 events[1]
1549 .old_values
1550 .as_ref()
1551 .and_then(|values| values.get("name")),
1552 None );
1554 assert_eq!(
1555 events[1]
1556 .new_values
1557 .as_ref()
1558 .and_then(|values| values.get("name")),
1559 Some(&Value::Text("updated".to_owned()))
1560 );
1561 assert_eq!(events[1].changes.len(), 1);
1562 assert_eq!(events[1].changes[0].field, "name");
1563 assert_eq!(
1564 events[1].changes[0].old_value,
1565 None );
1567 assert_eq!(
1568 events[1].changes[0].new_value,
1569 Some(Value::Text("updated".to_owned()))
1570 );
1571 assert_eq!(events[2].kind, RawAuditEventKind::Deleted);
1572 assert!(events[2].old_values.is_none()); assert!(events[2].new_values.is_none());
1574 assert_eq!(events[3].kind, RawAuditEventKind::Recovered);
1575 assert_eq!(
1576 events[3]
1577 .old_values
1578 .as_ref()
1579 .and_then(|values| values.get("version")),
1580 None );
1582 assert_eq!(
1583 events[3]
1584 .new_values
1585 .as_ref()
1586 .and_then(|values| values.get("version")),
1587 Some(&Value::I64(4))
1588 );
1589 assert_eq!(events[3].changes[0].field, "version");
1590 drop(events);
1591
1592 let safe_events = safe_events.lock().unwrap();
1593 assert_eq!(safe_events.len(), 4);
1594 assert_eq!(safe_events[0].kind, RawAuditEventKind::Created);
1595 let name = safe_events[0]
1596 .fields
1597 .iter()
1598 .find(|field| field.name == "name")
1599 .expect("application audit event should contain the changed name field");
1600 assert!(name.masked);
1601 assert_ne!(name.value.as_deref(), Some("created"));
1602 }
1603
1604 #[tokio::test]
1605 async fn entity_data_service_builds_relation_plans() {
1606 let mut context = UserContext::new()
1607 .with_metadata(
1608 InMemoryMetadataStore::new()
1609 .with_entity(entity())
1610 .with_entity(line_entity())
1611 .with_entity(product_entity()),
1612 )
1613 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1614 .with_entity_data_service_behavior_registry(
1615 InMemoryEntityDataServiceBehaviorRegistry::new()
1616 .with_behavior("Order", OrderBehavior),
1617 );
1618 context.insert_resource(PostgresDialect);
1619 context.insert_resource(StubExecutor {
1620 affected: 1,
1621 rows: Vec::new(),
1622 });
1623
1624 let repo = context
1625 .entity_data_service::<StubExecutor>("Order")
1626 .unwrap();
1627 let plans = repo.relation_plans().unwrap();
1628
1629 assert_eq!(plans.len(), 1);
1630 assert_eq!(plans[0].relation_name, "lines");
1631 assert_eq!(plans[0].target_entity, "OrderLine");
1632 assert_eq!(plans[0].local_key, "id");
1633 assert_eq!(plans[0].foreign_key, "order_id");
1634 assert!(plans[0].many);
1635 }
1636
1637 #[tokio::test]
1638 async fn entity_data_service_builds_relation_query_from_parent_rows() {
1639 let mut context = UserContext::new()
1640 .with_metadata(
1641 InMemoryMetadataStore::new()
1642 .with_entity(entity())
1643 .with_entity(line_entity())
1644 .with_entity(product_entity()),
1645 )
1646 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1647 .with_entity_data_service_behavior_registry(
1648 InMemoryEntityDataServiceBehaviorRegistry::new()
1649 .with_behavior("Order", OrderBehavior),
1650 );
1651 context.insert_resource(PostgresDialect);
1652 context.insert_resource(StubExecutor {
1653 affected: 1,
1654 rows: Vec::new(),
1655 });
1656
1657 let repo = context
1658 .entity_data_service::<StubExecutor>("Order")
1659 .unwrap();
1660 let parent_rows = vec![
1661 teaql_core::CompactRow::from_map(Record::from([(String::from("id"), Value::U64(11))])),
1662 teaql_core::CompactRow::from_map(Record::from([(String::from("id"), Value::U64(12))])),
1663 teaql_core::CompactRow::from_map(Record::from([(String::from("id"), Value::U64(11))])),
1664 ];
1665
1666 let query = repo.relation_query("lines", &parent_rows).unwrap();
1667 let Some(Expr::Binary { right, .. }) = query.filter else {
1668 panic!("relation query should contain an IN filter")
1669 };
1670 let Expr::Value(Value::List(ids)) = *right else {
1671 panic!("relation IN filter should contain identity values")
1672 };
1673 assert_eq!(ids, vec![Value::U64(11), Value::U64(12)]);
1674 }
1679
1680 #[tokio::test]
1681 async fn entity_data_service_enhances_parent_rows_with_relations() {
1682 let telemetry_events = Arc::new(Mutex::new(Vec::new()));
1683 let mut context = UserContext::new()
1684 .with_metadata(
1685 InMemoryMetadataStore::new()
1686 .with_entity(entity())
1687 .with_entity(line_entity())
1688 .with_entity(product_entity()),
1689 )
1690 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1691 .with_entity_data_service_behavior_registry(
1692 InMemoryEntityDataServiceBehaviorRegistry::new()
1693 .with_behavior("Order", OrderBehavior),
1694 )
1695 .with_runtime_telemetry(Arc::new(RecordingRuntimeTelemetry(
1696 telemetry_events.clone(),
1697 )));
1698 context.insert_resource(PostgresDialect);
1699 context.insert_resource(StubExecutor {
1700 affected: 1,
1701 rows: vec![
1702 Record::from([
1703 (String::from("id"), Value::U64(101)),
1704 (String::from("order_id"), Value::U64(11)),
1705 (String::from("name"), Value::Text(String::from("l1"))),
1706 ]),
1707 Record::from([
1708 (String::from("id"), Value::U64(102)),
1709 (String::from("order_id"), Value::U64(11)),
1710 (String::from("name"), Value::Text(String::from("l2"))),
1711 ]),
1712 Record::from([
1713 (String::from("id"), Value::U64(201)),
1714 (String::from("order_id"), Value::U64(12)),
1715 (String::from("name"), Value::Text(String::from("l3"))),
1716 ]),
1717 ],
1718 });
1719
1720 let repo = context
1721 .entity_data_service::<StubExecutor>("Order")
1722 .unwrap();
1723 let mut parents = vec![
1724 teaql_core::CompactRow::from_map(Record::from([(String::from("id"), Value::U64(11))])),
1725 teaql_core::CompactRow::from_map(Record::from([(String::from("id"), Value::U64(12))])),
1726 ];
1727
1728 repo.enhance_relations_internal(&mut parents).await.unwrap();
1729
1730 match parents[0].get("lines") {
1731 Some(Value::List(lines)) => assert_eq!(lines.len(), 2),
1732 other => panic!("unexpected lines payload: {other:?}"),
1733 }
1734 match parents[1].get("lines") {
1735 Some(Value::List(lines)) => assert_eq!(lines.len(), 1),
1736 other => panic!("unexpected lines payload: {other:?}"),
1737 }
1738 assert!(
1739 telemetry_events
1740 .lock()
1741 .unwrap()
1742 .iter()
1743 .any(|event| event == "start:relation_load")
1744 );
1745 }
1746
1747 #[tokio::test]
1748 async fn relation_limit_is_partitioned_per_parent_and_rank_is_internal() {
1749 let mut rows = Vec::new();
1750 for (order_id, first_line_id) in [(11_u64, 101_u64), (12_u64, 201_u64)] {
1751 for rank in 1_u64..=3 {
1752 rows.push(Record::from([
1753 (String::from("id"), Value::U64(first_line_id + rank - 1)),
1754 (String::from("order_id"), Value::U64(order_id)),
1755 (
1756 String::from(teaql_core::PARTITION_RANK_PROPERTY),
1757 Value::U64(rank),
1758 ),
1759 ]));
1760 }
1761 }
1762
1763 let mut context = UserContext::new()
1764 .with_metadata(
1765 InMemoryMetadataStore::new()
1766 .with_entity(entity())
1767 .with_entity(line_entity()),
1768 )
1769 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1770 context.insert_resource(PostgresDialect);
1771 context.insert_resource(CapturingQueryExecutor {
1772 rows,
1773 queries: Mutex::new(Vec::new()),
1774 });
1775
1776 let repo = context
1777 .entity_data_service::<CapturingQueryExecutor>("Order")
1778 .unwrap();
1779 let mut parents = vec![
1780 teaql_core::CompactRow::from_map(Record::from([(String::from("id"), Value::U64(11))])),
1781 teaql_core::CompactRow::from_map(Record::from([(String::from("id"), Value::U64(12))])),
1782 ];
1783 let query = SelectQuery::new("Order").relation_query(
1784 "lines",
1785 SelectQuery::new("OrderLine")
1786 .order_by(OrderBy::desc("id"))
1787 .limit(3),
1788 );
1789
1790 repo.enhance_query_relations_internal(&mut parents, &query)
1791 .await
1792 .unwrap();
1793
1794 let captured = &context
1795 .get_resource::<CapturingQueryExecutor>()
1796 .unwrap()
1797 .queries
1798 .lock()
1799 .unwrap()[0];
1800 assert_eq!(captured.partition_by.as_deref(), Some("order_id"));
1801 assert_eq!(captured.slice.and_then(|slice| slice.limit), Some(3));
1802 for parent in &parents {
1803 let Some(Value::List(lines)) = parent.get("lines") else {
1804 panic!("missing relation lines")
1805 };
1806 assert_eq!(lines.len(), 3);
1807 assert!(lines.iter().all(|line| match line {
1808 Value::Object(line) => !line.contains_key(teaql_core::PARTITION_RANK_PROPERTY),
1809 _ => false,
1810 }));
1811 }
1812 }
1813
1814 #[tokio::test]
1815 async fn relation_enhancement_wraps_inverse_many_relation_as_list() {
1816 let mut context = UserContext::new()
1817 .with_metadata(
1818 InMemoryMetadataStore::new()
1819 .with_entity(OrderLineWithProductEntityRow::entity_descriptor())
1820 .with_entity(ProductWithLinesEntityRow::entity_descriptor()),
1821 )
1822 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("OrderLine"));
1823 context.insert_resource(PostgresDialect);
1824 context.insert_resource(QueueExecutor {
1825 affected: 1,
1826 rows: Mutex::new(VecDeque::from([
1827 vec![Record::from([
1828 (String::from("id"), Value::U64(11)),
1829 (String::from("order_id"), Value::U64(7)),
1830 (String::from("name"), Value::Text(String::from("line"))),
1831 (String::from("product_id"), Value::U64(101)),
1832 ])],
1833 vec![Record::from([
1834 (String::from("id"), Value::U64(101)),
1835 (String::from("name"), Value::Text(String::from("sku"))),
1836 ])],
1837 ])),
1838 queries: Mutex::new(Vec::new()),
1839 });
1840
1841 let repo = context
1842 .entity_data_service::<QueueExecutor>("OrderLine")
1843 .unwrap();
1844 let rows = repo
1845 .fetch_enhanced_entities_internal::<OrderLineWithProductEntityRow>(
1846 &SelectQuery::new("OrderLine").relation("product"),
1847 )
1848 .await
1849 .unwrap();
1850
1851 let product = rows.data[0].product.as_ref().unwrap();
1852 assert_eq!(product.lines.data.len(), 1);
1853 assert_eq!(product.lines.data[0].id, 11);
1854 }
1855
1856 #[tokio::test]
1857 async fn generated_to_one_getter_resolves_from_runtime_module_identity_graph() {
1858 let mut context = RuntimeModule::new()
1859 .entity::<FlatTripRow>()
1860 .entity::<FlatVendorRow>()
1861 .into_context();
1862 context.insert_resource(PostgresDialect);
1863 context.insert_resource(QueueExecutor {
1864 affected: 1,
1865 rows: Mutex::new(VecDeque::from([
1866 vec![Record::from([
1867 (String::from("id"), Value::U64(11)),
1868 (String::from("vendor_id"), Value::U64(101)),
1869 ])],
1870 vec![Record::from([
1871 (String::from("id"), Value::U64(101)),
1872 (String::from("name"), Value::Text(String::from("Acme"))),
1873 ])],
1874 ])),
1875 queries: Mutex::new(Vec::new()),
1876 });
1877
1878 let repo = context
1879 .entity_data_service::<QueueExecutor>("FlatTrip")
1880 .unwrap();
1881 let rows = repo
1882 .fetch_enhanced_entities_internal::<FlatTripRow>(
1883 &SelectQuery::new("FlatTrip").relation("vendor"),
1884 )
1885 .await
1886 .unwrap();
1887
1888 assert!(rows.data[0].vendor.is_none());
1889 assert_eq!(rows.data[0].vendor().unwrap().name, "Acme");
1890 }
1891
1892 #[tokio::test]
1893 async fn generated_to_many_getter_uses_adjacency_and_mutation_copies_on_write() {
1894 let mut context = RuntimeModule::new()
1895 .entity::<FlatFleetRow>()
1896 .entity::<FlatFleetTripRow>()
1897 .into_context();
1898 context.insert_resource(PostgresDialect);
1899 context.insert_resource(QueueExecutor {
1900 affected: 1,
1901 rows: Mutex::new(VecDeque::from([
1902 vec![Record::from([(String::from("id"), Value::U64(7))])],
1903 vec![
1904 Record::from([
1905 (String::from("id"), Value::U64(11)),
1906 (String::from("fleet_id"), Value::U64(7)),
1907 (String::from("name"), Value::Text(String::from("first"))),
1908 ]),
1909 Record::from([
1910 (String::from("id"), Value::U64(12)),
1911 (String::from("fleet_id"), Value::U64(7)),
1912 (String::from("name"), Value::Text(String::from("second"))),
1913 ]),
1914 ],
1915 ])),
1916 queries: Mutex::new(Vec::new()),
1917 });
1918
1919 let repo = context
1920 .entity_data_service::<QueueExecutor>("FlatFleet")
1921 .unwrap();
1922 let mut rows = repo
1923 .fetch_enhanced_entities_internal::<FlatFleetRow>(
1924 &SelectQuery::new("FlatFleet").relation("trip_list"),
1925 )
1926 .await
1927 .unwrap();
1928 let fleet = &mut rows.data[0];
1929
1930 assert!(!fleet.trip_list.is_loaded);
1931 assert_eq!(fleet.trip_list().data.len(), 2);
1932 assert_eq!(fleet.trip_list().data[1].name, "second");
1933 fleet.trip_list_mut().push(FlatFleetTripRow {
1934 id: 13,
1935 fleet_id: 7,
1936 name: "third".to_owned(),
1937 root: EntityRoot::default(),
1938 });
1939 assert!(fleet.trip_list.is_loaded);
1940 assert_eq!(fleet.trip_list().data.len(), 3);
1941 assert_eq!(
1942 fleet
1943 .root
1944 .resolve_relation_list::<FlatFleetTripRow>("FlatFleet", 7, "trip_list")
1945 .unwrap()
1946 .data
1947 .len(),
1948 2
1949 );
1950 }
1951
1952 #[tokio::test]
1953 async fn entity_data_service_fetches_smart_list_of_entities() {
1954 let mut context = UserContext::new()
1955 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1956 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1957 context.insert_resource(PostgresDialect);
1958 context.insert_resource(StubExecutor {
1959 affected: 1,
1960 rows: vec![Record::from([
1961 (String::from("id"), Value::U64(7)),
1962 (String::from("version"), Value::I64(2)),
1963 (String::from("name"), Value::Text(String::from("typed"))),
1964 ])],
1965 });
1966
1967 let repo = context
1968 .entity_data_service::<StubExecutor>("Order")
1969 .unwrap();
1970 let rows = repo
1971 .fetch_entities_internal::<OrderEntity>(&repo.select())
1972 .await
1973 .unwrap();
1974
1975 assert_eq!(rows.len(), 1);
1976 assert_eq!(
1977 rows.first(),
1978 Some(&OrderEntity {
1979 id: 7,
1980 version: 2,
1981 name: String::from("typed"),
1982 })
1983 );
1984 }
1985
1986 #[tokio::test]
1987 async fn entity_data_service_fetches_smart_list_of_derived_entities() {
1988 let mut context = UserContext::new()
1989 .with_metadata(
1990 InMemoryMetadataStore::new().with_entity(CatalogProductRow::entity_descriptor()),
1991 )
1992 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("CatalogProduct"));
1993 context.insert_resource(PostgresDialect);
1994 context.insert_resource(StubExecutor {
1995 affected: 1,
1996 rows: vec![Record::from([
1997 (String::from("id"), Value::U64(9)),
1998 (String::from("name"), Value::Text(String::from("derived"))),
1999 ])],
2000 });
2001
2002 let repo = context
2003 .entity_data_service::<StubExecutor>("CatalogProduct")
2004 .unwrap();
2005 let rows = repo
2006 .fetch_entities_internal::<CatalogProductRow>(&repo.select())
2007 .await
2008 .unwrap();
2009
2010 assert_eq!(rows.len(), 1);
2011 assert_eq!(
2012 rows.first(),
2013 Some(&CatalogProductRow {
2014 id: 9,
2015 name: String::from("derived"),
2016 })
2017 );
2018 }
2019
2020 #[tokio::test]
2021 async fn entity_data_service_collects_dynamic_properties_for_aggregate_output() {
2022 let mut context = UserContext::new()
2023 .with_metadata(
2024 InMemoryMetadataStore::new()
2025 .with_entity(OrderAggregateDynamic::entity_descriptor()),
2026 )
2027 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("OrderAggregate"));
2028 context.insert_resource(PostgresDialect);
2029 context.insert_resource(StubExecutor {
2030 affected: 1,
2031 rows: vec![Record::from([
2032 (String::from("id"), Value::U64(1)),
2033 (String::from("lineCount"), Value::I64(3)),
2034 (String::from("amount"), Value::F64(18.5)),
2035 ])],
2036 });
2037
2038 let repo = context
2039 .entity_data_service::<StubExecutor>("OrderAggregate")
2040 .unwrap();
2041 let rows = repo
2042 .fetch_entities_internal::<OrderAggregateDynamic>(&repo.select())
2043 .await
2044 .unwrap();
2045
2046 assert_eq!(rows.len(), 1);
2047 assert_eq!(rows.data[0].id, 1);
2048 assert_eq!(rows.data[0].dynamic.get("lineCount"), Some(&Value::I64(3)));
2049 assert_eq!(rows.data[0].dynamic.get("amount"), Some(&Value::F64(18.5)));
2050 assert_eq!(
2051 rows.into_vec().into_iter().next().unwrap().into_json(),
2052 serde_json::json!({
2053 "id": 1,
2054 "lineCount": 3,
2055 "amount": 18.5
2056 })
2057 );
2058 }
2059
2060 #[tokio::test]
2061 async fn entity_data_service_executes_relation_aggregates_into_dynamic_properties() {
2062 let executor = QueueExecutor {
2063 affected: 1,
2064 rows: Mutex::new(VecDeque::from([
2065 vec![
2066 Record::from([
2067 (String::from("id"), Value::U64(1)),
2068 (String::from("version"), Value::I64(1)),
2069 (String::from("name"), Value::Text(String::from("first"))),
2070 ]),
2071 Record::from([
2072 (String::from("id"), Value::U64(2)),
2073 (String::from("version"), Value::I64(1)),
2074 (String::from("name"), Value::Text(String::from("second"))),
2075 ]),
2076 ],
2077 vec![Record::from([
2078 (String::from("order_id"), Value::U64(1)),
2079 (String::from("lineCount"), Value::I64(3)),
2080 ])],
2081 ])),
2082 queries: Mutex::new(Vec::new()),
2083 };
2084 let mut context = UserContext::new()
2085 .with_metadata(
2086 InMemoryMetadataStore::new()
2087 .with_entity(entity())
2088 .with_entity(line_entity()),
2089 )
2090 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
2091 context.insert_resource(PostgresDialect);
2092 context.insert_resource(executor);
2093
2094 let repo = context
2095 .entity_data_service::<QueueExecutor>("Order")
2096 .unwrap();
2097 let rows = repo
2098 .fetch_all_with_relation_aggregates_internal(
2099 &repo
2100 .select()
2101 .project("id")
2102 .project("version")
2103 .project("name"),
2104 &[RelationAggregate::new(
2105 "lines",
2106 "lineCount",
2107 SelectQuery::new("OrderLine"),
2108 true,
2109 )],
2110 )
2111 .await
2112 .unwrap();
2113
2114 assert_eq!(rows[0].get("lineCount"), Some(&Value::I64(3)));
2115 assert_eq!(rows[1].get("lineCount"), Some(&Value::U64(0)));
2116
2117 let executor = context.get_resource::<QueueExecutor>().unwrap();
2118 let queries = executor.queries.lock().unwrap();
2119 assert_eq!(queries.len(), 2);
2120 assert_eq!(queries[1], "SELECT ... FROM OrderLine ...");
2121 }
2122
2123 #[tokio::test]
2124 async fn entity_data_service_maps_relation_aggregate_storage_key_to_property_key() {
2125 let mut line = line_entity();
2126 line.properties
2127 .iter_mut()
2128 .find(|property| property.name == "order_id")
2129 .unwrap()
2130 .column_name = "order_ref".to_owned();
2131 let executor = QueueExecutor {
2132 affected: 1,
2133 rows: Mutex::new(VecDeque::from([
2134 vec![Record::from([
2135 (String::from("id"), Value::U64(1)),
2136 (String::from("version"), Value::I64(1)),
2137 (String::from("name"), Value::Text(String::from("first"))),
2138 ])],
2139 vec![Record::from([
2140 (String::from("order_ref"), Value::I64(1)),
2141 (String::from("lineCount"), Value::I64(3)),
2142 ])],
2143 ])),
2144 queries: Mutex::new(Vec::new()),
2145 };
2146 let mut context = UserContext::new()
2147 .with_metadata(
2148 InMemoryMetadataStore::new()
2149 .with_entity(entity())
2150 .with_entity(line),
2151 )
2152 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
2153 context.insert_resource(PostgresDialect);
2154 context.insert_resource(executor);
2155
2156 let repo = context
2157 .entity_data_service::<QueueExecutor>("Order")
2158 .unwrap();
2159 let rows = repo
2160 .fetch_all_with_relation_aggregates_internal(
2161 &repo
2162 .select()
2163 .project("id")
2164 .project("version")
2165 .project("name"),
2166 &[RelationAggregate::new(
2167 "lines",
2168 "lineCount",
2169 SelectQuery::new("OrderLine"),
2170 true,
2171 )],
2172 )
2173 .await
2174 .unwrap();
2175
2176 assert_eq!(rows[0].get("lineCount"), Some(&Value::I64(3)));
2177 let executor = context.get_resource::<QueueExecutor>().unwrap();
2178 assert_eq!(
2179 executor.queries.lock().unwrap()[1],
2180 "SELECT ... FROM OrderLine ..."
2181 );
2182 }
2183
2184 #[tokio::test]
2185 async fn entity_data_service_uses_aggregation_cache_when_resource_is_registered() {
2186 let telemetry_events = Arc::new(Mutex::new(Vec::new()));
2187 let executor = QueueExecutor {
2188 affected: 1,
2189 rows: Mutex::new(VecDeque::from([vec![Record::from([(
2190 String::from("count"),
2191 Value::I64(2),
2192 )])]])),
2193 queries: Mutex::new(Vec::new()),
2194 };
2195 let mut context = UserContext::new()
2196 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
2197 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
2198 .with_runtime_telemetry(Arc::new(RecordingRuntimeTelemetry(
2199 telemetry_events.clone(),
2200 )));
2201 context.insert_resource(PostgresDialect);
2202 context.insert_resource(executor);
2203 context.insert_resource(InMemoryAggregationCache::default());
2204
2205 let repo = context
2206 .entity_data_service::<QueueExecutor>("Order")
2207 .unwrap();
2208 let query = repo
2209 .select()
2210 .count("count")
2211 .enable_aggregation_cache_for(60_000);
2212
2213 let first = repo.fetch_all_internal(&query).await.unwrap();
2214 let second = repo.fetch_all_internal(&query).await.unwrap();
2215
2216 assert_eq!(first, second);
2217 let executor = context.get_resource::<QueueExecutor>().unwrap();
2218 assert_eq!(executor.queries.lock().unwrap().len(), 1);
2219 let events = telemetry_events.lock().unwrap();
2220 assert_eq!(
2221 events
2222 .iter()
2223 .filter(|event| event.as_str() == "start:cache")
2224 .count(),
2225 2
2226 );
2227 assert_eq!(
2228 events
2229 .iter()
2230 .filter(|event| event.as_str() == "start:provider")
2231 .count(),
2232 1
2233 );
2234 }
2235
2236 #[tokio::test]
2237 async fn continuous_page_fetch_uses_id_seek_for_the_next_page() {
2238 let rows = (91_u64..=100)
2239 .rev()
2240 .map(|id| {
2241 Record::from([
2242 (String::from("id"), Value::U64(id)),
2243 (String::from("version"), Value::I64(1)),
2244 (String::from("name"), Value::Text(format!("order-{id}"))),
2245 ])
2246 })
2247 .collect();
2248 let mut context = UserContext::new()
2249 .with_user_identifier("tenant-1:user-1")
2250 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
2251 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
2252 context.insert_resource(PostgresDialect);
2253 context.insert_resource(CapturingQueryExecutor {
2254 rows,
2255 queries: Mutex::new(Vec::new()),
2256 });
2257 let repo = context
2258 .entity_data_service::<CapturingQueryExecutor>("Order")
2259 .unwrap();
2260
2261 let first = SelectQuery::new("Order")
2262 .order_desc("id")
2263 .page(0, 10)
2264 .optimize_for_continuous_page_fetch_with("recent-orders", 60);
2265 repo.fetch_all_internal(&first).await.unwrap();
2266 assert_eq!(
2267 context.continuous_page_plan().as_deref(),
2268 Some("OFFSET_FALLBACK:FIRST_PAGE")
2269 );
2270
2271 let second = SelectQuery::new("Order")
2272 .order_desc("id")
2273 .page(10, 10)
2274 .optimize_for_continuous_page_fetch_with("recent-orders", 60);
2275 repo.fetch_all_internal(&second).await.unwrap();
2276 assert_eq!(
2277 context.continuous_page_plan().as_deref(),
2278 Some("CURSOR_SEEK")
2279 );
2280 assert!(context.continuous_page_cursor_id().is_some());
2281
2282 let captured = context
2283 .get_resource::<CapturingQueryExecutor>()
2284 .unwrap()
2285 .queries
2286 .lock()
2287 .unwrap();
2288 assert_eq!(
2289 captured[1].slice.as_ref().map(|slice| slice.offset),
2290 Some(0)
2291 );
2292 assert!(format!("{:?}", captured[1].filter).contains("Lt"));
2293 assert!(format!("{:?}", captured[1].filter).contains("U64(91)"));
2294 }
2295
2296 #[tokio::test]
2297 async fn aggregation_cache_is_namespaced_and_invalidated_after_write() {
2298 let executor = QueueExecutor {
2299 affected: 1,
2300 rows: Mutex::new(VecDeque::from([
2301 vec![Record::from([(String::from("count"), Value::I64(2))])],
2302 vec![Record::from([(String::from("count"), Value::I64(3))])],
2303 ])),
2304 queries: Mutex::new(Vec::new()),
2305 };
2306 let mut context = UserContext::new()
2307 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
2308 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
2309 context.insert_resource(PostgresDialect);
2310 context.insert_resource(executor);
2311 context.insert_resource(
2312 Arc::new(InMemoryAggregationCache::with_namespace("tenant-a"))
2313 as Arc<dyn AggregationCacheBackend>,
2314 );
2315
2316 let repo = context
2317 .entity_data_service::<QueueExecutor>("Order")
2318 .unwrap();
2319 let query = repo
2320 .select()
2321 .count("count")
2322 .enable_aggregation_cache_for(60_000);
2323
2324 let first = repo.fetch_all_internal(&query).await.unwrap();
2325 let cached = repo.fetch_all_internal(&query).await.unwrap();
2326 repo.insert_internal(
2327 &InsertCommand::new("Order")
2328 .value("id", 9_u64)
2329 .value("version", 1_i64)
2330 .value("name", "new"),
2331 )
2332 .await
2333 .unwrap();
2334 let refreshed = repo.fetch_all_internal(&query).await.unwrap();
2335
2336 assert_eq!(first, cached);
2337 assert_ne!(cached, refreshed);
2338 let executor = context.get_resource::<QueueExecutor>().unwrap();
2339 assert_eq!(executor.queries.lock().unwrap().len(), 2);
2340 }
2341
2342 #[tokio::test]
2343 async fn aggregation_cache_propagates_to_relation_aggregates() {
2344 let parent_rows = vec![
2345 Record::from([
2346 (String::from("id"), Value::U64(1)),
2347 (String::from("version"), Value::I64(1)),
2348 (String::from("name"), Value::Text(String::from("first"))),
2349 ]),
2350 Record::from([
2351 (String::from("id"), Value::U64(2)),
2352 (String::from("version"), Value::I64(1)),
2353 (String::from("name"), Value::Text(String::from("second"))),
2354 ]),
2355 ];
2356 let aggregate_rows = vec![Record::from([
2357 (String::from("order_id"), Value::U64(1)),
2358 (String::from("lineCount"), Value::I64(3)),
2359 ])];
2360 let executor = QueueExecutor {
2361 affected: 1,
2362 rows: Mutex::new(VecDeque::from([parent_rows, aggregate_rows])),
2363 queries: Mutex::new(Vec::new()),
2364 };
2365 let mut context = UserContext::new()
2366 .with_metadata(
2367 InMemoryMetadataStore::new()
2368 .with_entity(entity())
2369 .with_entity(line_entity()),
2370 )
2371 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
2372 context.insert_resource(PostgresDialect);
2373 context.insert_resource(executor);
2374 context.insert_resource(InMemoryAggregationCache::default());
2375
2376 let repo = context
2377 .entity_data_service::<QueueExecutor>("Order")
2378 .unwrap();
2379 let query = repo
2380 .select()
2381 .project("id")
2382 .project("version")
2383 .project("name")
2384 .enable_aggregation_cache_for(60_000)
2385 .propagate_aggregation_cache(60_000);
2386 let aggregate =
2387 RelationAggregate::new("lines", "lineCount", SelectQuery::new("OrderLine"), true);
2388
2389 let first = repo
2390 .fetch_all_with_relation_aggregates_internal(&query, &[aggregate.clone()])
2391 .await
2392 .unwrap();
2393 let second = repo
2394 .fetch_all_with_relation_aggregates_internal(&query, &[aggregate])
2395 .await
2396 .unwrap();
2397
2398 let executor = context.get_resource::<QueueExecutor>().unwrap();
2399 assert_eq!(executor.queries.lock().unwrap().len(), 2);
2400 assert_eq!(first, second);
2401 }
2402
2403 #[tokio::test]
2404 async fn memory_data_service_fetches_smart_list_entities_with_query_features() {
2405 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2406 let data_service = MemoryDataService::new(metadata).with_rows(
2407 "Order",
2408 vec![
2409 Record::from([
2410 (String::from("id"), Value::U64(1)),
2411 (String::from("version"), Value::I64(1)),
2412 (String::from("name"), Value::Text(String::from("alpha"))),
2413 ]),
2414 Record::from([
2415 (String::from("id"), Value::U64(2)),
2416 (String::from("version"), Value::I64(1)),
2417 (String::from("name"), Value::Text(String::from("beta"))),
2418 ]),
2419 Record::from([
2420 (String::from("id"), Value::U64(3)),
2421 (String::from("version"), Value::I64(1)),
2422 (String::from("name"), Value::Text(String::from("gamma"))),
2423 ]),
2424 ],
2425 );
2426
2427 let query = teaql_core::SelectQuery::new("Order")
2428 .filter(Expr::Binary {
2429 left: Box::new(Expr::column("id")),
2430 op: teaql_core::BinaryOp::Gte,
2431 right: Box::new(Expr::value(2_u64)),
2432 })
2433 .order_by(OrderBy::desc("id"))
2434 .limit(1);
2435
2436 let orders = data_service.fetch_entities::<Order>(&query).unwrap();
2437
2438 assert_eq!(orders.ids(), vec![Value::U64(3)]);
2439 assert_eq!(orders.versions(), vec![1]);
2440 assert_eq!(orders.first().unwrap().name, "gamma");
2441 }
2442
2443 #[tokio::test]
2444 async fn memory_data_service_runs_relation_aggregates() {
2445 let metadata = InMemoryMetadataStore::new()
2446 .with_entity(entity())
2447 .with_entity(line_entity());
2448
2449 let data_service = MemoryDataService::new(metadata)
2450 .with_rows(
2451 "Order",
2452 vec![
2453 Record::from([
2454 (String::from("id"), Value::U64(1)),
2455 (String::from("version"), Value::I64(1)),
2456 (String::from("name"), Value::Text(String::from("first"))),
2457 ]),
2458 Record::from([
2459 (String::from("id"), Value::U64(2)),
2460 (String::from("version"), Value::I64(1)),
2461 (String::from("name"), Value::Text(String::from("second"))),
2462 ]),
2463 ],
2464 )
2465 .with_rows(
2466 "OrderLine",
2467 vec![
2468 Record::from([
2469 (String::from("id"), Value::U64(10)),
2470 (String::from("version"), Value::I64(1)),
2471 (String::from("order_id"), Value::U64(1)),
2472 (String::from("name"), Value::Text(String::from("line1"))),
2473 ]),
2474 Record::from([
2475 (String::from("id"), Value::U64(11)),
2476 (String::from("version"), Value::I64(1)),
2477 (String::from("order_id"), Value::U64(1)),
2478 (String::from("name"), Value::Text(String::from("line2"))),
2479 ]),
2480 Record::from([
2481 (String::from("id"), Value::U64(12)),
2482 (String::from("version"), Value::I64(1)),
2483 (String::from("order_id"), Value::U64(2)),
2484 (String::from("name"), Value::Text(String::from("line3"))),
2485 ]),
2486 ],
2487 );
2488
2489 let query = SelectQuery::new("Order").project("id").project("name");
2490 let aggregate =
2491 RelationAggregate::new("lines", "lineCount", SelectQuery::new("OrderLine"), true);
2492
2493 let rows = data_service
2494 .fetch_all_with_relation_aggregates(&query, &[aggregate])
2495 .unwrap();
2496
2497 assert_eq!(rows.len(), 2);
2498
2499 let first_order = rows
2500 .iter()
2501 .find(|r| r.get("id") == Some(&Value::U64(1)))
2502 .unwrap();
2503 assert_eq!(first_order.get("lineCount"), Some(&Value::U64(2)));
2504
2505 let second_order = rows
2506 .iter()
2507 .find(|r| r.get("id") == Some(&Value::U64(2)))
2508 .unwrap();
2509 assert_eq!(second_order.get("lineCount"), Some(&Value::U64(1)));
2510 }
2511
2512 #[tokio::test]
2513 async fn memory_data_service_runs_aggregates() {
2514 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2515 let data_service = MemoryDataService::new(metadata).with_rows(
2516 "Order",
2517 vec![
2518 Record::from([
2519 (String::from("id"), Value::U64(1)),
2520 (String::from("version"), Value::I64(1)),
2521 (String::from("name"), Value::Text(String::from("alpha"))),
2522 ]),
2523 Record::from([
2524 (String::from("id"), Value::U64(2)),
2525 (String::from("version"), Value::I64(2)),
2526 (String::from("name"), Value::Text(String::from("beta"))),
2527 ]),
2528 ],
2529 );
2530
2531 let query = teaql_core::SelectQuery {
2532 hard_limit: 10_000,
2533 entity: String::from("Order"),
2534 projection: Vec::new(),
2535 expr_projection: Vec::new(),
2536 filter: None,
2537 having: None,
2538 order_by: Vec::new(),
2539 slice: None,
2540 partition_by: None,
2541 trace_chain: Vec::new(),
2542 aggregates: vec![
2543 Aggregate {
2544 function: AggregateFunction::Count,
2545 field: String::from("id"),
2546 alias: String::from("count"),
2547 },
2548 Aggregate {
2549 function: AggregateFunction::Sum,
2550 field: String::from("version"),
2551 alias: String::from("versionSum"),
2552 },
2553 ],
2554 group_by: Vec::new(),
2555 relations: Vec::new(),
2556 aggregation_cache: None,
2557 comment: None,
2558 raw_sql: None,
2559 raw_sql_search_criteria: Vec::new(),
2560 dynamic_properties: Vec::new(),
2561 raw_projections: Vec::new(),
2562 object_group_bys: Vec::new(),
2563 search_with_text: None,
2564 child_enhancements: Vec::new(),
2565 stream_config: None,
2566 continuous_page_fetch: None,
2567 };
2568
2569 let rows = data_service.fetch_all(&query).unwrap();
2570
2571 assert_eq!(rows.len(), 1);
2572 assert_eq!(rows[0].get("count"), Some(&Value::U64(2)));
2573 assert_eq!(rows[0].get("versionSum"), Some(&Value::U64(3)));
2574 }
2575
2576 #[tokio::test]
2577 async fn memory_data_service_runs_grouped_aggregates_and_extended_filters() {
2578 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2579 let data_service = MemoryDataService::new(metadata).with_rows(
2580 "Order",
2581 vec![
2582 Record::from([
2583 (String::from("id"), Value::U64(1)),
2584 (String::from("version"), Value::I64(1)),
2585 (String::from("name"), Value::Text(String::from("alpha"))),
2586 ]),
2587 Record::from([
2588 (String::from("id"), Value::U64(2)),
2589 (String::from("version"), Value::I64(2)),
2590 (String::from("name"), Value::Text(String::from("alpha"))),
2591 ]),
2592 Record::from([
2593 (String::from("id"), Value::U64(3)),
2594 (String::from("version"), Value::I64(3)),
2595 (String::from("name"), Value::Text(String::from("tmp-beta"))),
2596 ]),
2597 ],
2598 );
2599
2600 let rows = data_service
2601 .fetch_all(
2602 &teaql_core::SelectQuery::new("Order")
2603 .filter(
2604 Expr::between("version", 1_i64, 3_i64)
2605 .and_expr(Expr::not_like("name", "tmp%"))
2606 .and_expr(Expr::not_in_list("name", vec![Value::from("deleted")])),
2607 )
2608 .group_by("name")
2609 .count("total")
2610 .sum("version", "versionSum"),
2611 )
2612 .unwrap();
2613
2614 assert_eq!(rows.len(), 1);
2615 assert_eq!(
2616 rows[0].get("name"),
2617 Some(&Value::Text(String::from("alpha")))
2618 );
2619 assert_eq!(rows[0].get("total"), Some(&Value::U64(2)));
2620 assert_eq!(rows[0].get("versionSum"), Some(&Value::U64(3)));
2621 }
2622
2623 #[tokio::test]
2624 async fn memory_data_service_runs_extended_aggregates_and_having() {
2625 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2626 let data_service = MemoryDataService::new(metadata).with_rows(
2627 "Order",
2628 vec![
2629 Record::from([
2630 (String::from("id"), Value::U64(1)),
2631 (String::from("version"), Value::I64(1)),
2632 (String::from("name"), Value::Text(String::from("alpha"))),
2633 ]),
2634 Record::from([
2635 (String::from("id"), Value::U64(2)),
2636 (String::from("version"), Value::I64(3)),
2637 (String::from("name"), Value::Text(String::from("alpha"))),
2638 ]),
2639 Record::from([
2640 (String::from("id"), Value::U64(3)),
2641 (String::from("version"), Value::I64(7)),
2642 (String::from("name"), Value::Text(String::from("beta"))),
2643 ]),
2644 ],
2645 );
2646
2647 let rows = data_service
2648 .fetch_all(
2649 &teaql_core::SelectQuery::new("Order")
2650 .group_by("name")
2651 .count("total")
2652 .stddev("version", "stddevVersion")
2653 .var_pop("version", "varPopVersion")
2654 .bit_or("version", "bitOrVersion")
2655 .having(Expr::gt("total", 1_i64)),
2656 )
2657 .unwrap();
2658
2659 assert_eq!(rows.len(), 1);
2660 assert_eq!(
2661 rows[0].get("name"),
2662 Some(&Value::Text(String::from("alpha")))
2663 );
2664 assert_eq!(rows[0].get("total"), Some(&Value::U64(2)));
2665 assert_eq!(
2666 rows[0].get("stddevVersion").map(Value::to_json_value),
2667 Some(serde_json::Value::String(
2668 "1.4142135623730951454746218583".to_owned()
2669 ))
2670 );
2671 assert_eq!(
2672 rows[0].get("varPopVersion"),
2673 Some(&Value::Decimal(Decimal::ONE))
2674 );
2675 assert_eq!(rows[0].get("bitOrVersion"), Some(&Value::I64(3)));
2676 }
2677
2678 #[tokio::test]
2679 async fn memory_data_service_runs_sound_like_filter() {
2680 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2681 let data_service = MemoryDataService::new(metadata).with_rows(
2682 "Order",
2683 vec![
2684 Record::from([
2685 (String::from("id"), Value::U64(1)),
2686 (String::from("version"), Value::I64(1)),
2687 (String::from("name"), Value::Text(String::from("Robert"))),
2688 ]),
2689 Record::from([
2690 (String::from("id"), Value::U64(2)),
2691 (String::from("version"), Value::I64(1)),
2692 (String::from("name"), Value::Text(String::from("Rupert"))),
2693 ]),
2694 Record::from([
2695 (String::from("id"), Value::U64(3)),
2696 (String::from("version"), Value::I64(1)),
2697 (String::from("name"), Value::Text(String::from("Ashcraft"))),
2698 ]),
2699 ],
2700 );
2701
2702 let rows = data_service
2703 .fetch_all(
2704 &teaql_core::SelectQuery::new("Order")
2705 .filter(Expr::sound_like("name", "Robert"))
2706 .order_asc("id"),
2707 )
2708 .unwrap();
2709
2710 assert_eq!(rows.len(), 2);
2711 assert_eq!(rows[0].get("name"), Some(&Value::Text("Robert".to_owned())));
2712 assert_eq!(rows[1].get("name"), Some(&Value::Text("Rupert".to_owned())));
2713 }
2714
2715 #[tokio::test]
2716 async fn memory_data_service_runs_java_style_string_match_filters() {
2717 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2718 let data_service = MemoryDataService::new(metadata).with_rows(
2719 "Order",
2720 vec![
2721 Record::from([
2722 (String::from("id"), Value::U64(1)),
2723 (String::from("version"), Value::I64(1)),
2724 (String::from("name"), Value::Text(String::from("tea-order"))),
2725 ]),
2726 Record::from([
2727 (String::from("id"), Value::U64(2)),
2728 (String::from("version"), Value::I64(1)),
2729 (
2730 String::from("name"),
2731 Value::Text(String::from("coffee-order")),
2732 ),
2733 ]),
2734 Record::from([
2735 (String::from("id"), Value::U64(3)),
2736 (String::from("version"), Value::I64(1)),
2737 (
2738 String::from("name"),
2739 Value::Text(String::from("tea-archived")),
2740 ),
2741 ]),
2742 ],
2743 );
2744
2745 let rows = data_service
2746 .fetch_all(
2747 &teaql_core::SelectQuery::new("Order")
2748 .filter(
2749 Expr::contain("name", "tea")
2750 .and_expr(Expr::begin_with("name", "tea"))
2751 .and_expr(Expr::end_with("name", "order"))
2752 .and_expr(Expr::not_contain("name", "coffee"))
2753 .and_expr(Expr::not_begin_with("name", "archived"))
2754 .and_expr(Expr::not_end_with("name", "draft")),
2755 )
2756 .order_asc("id"),
2757 )
2758 .unwrap();
2759
2760 assert_eq!(rows.len(), 1);
2761 assert_eq!(
2762 rows[0].get("name"),
2763 Some(&Value::Text("tea-order".to_owned()))
2764 );
2765 }
2766
2767 #[tokio::test]
2768 async fn memory_data_service_runs_property_to_property_filters() {
2769 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2770 let data_service = MemoryDataService::new(metadata).with_rows(
2771 "Order",
2772 vec![
2773 Record::from([
2774 (String::from("id"), Value::U64(1)),
2775 (String::from("version"), Value::I64(2)),
2776 (String::from("name"), Value::Text(String::from("keep"))),
2777 ]),
2778 Record::from([
2779 (String::from("id"), Value::U64(2)),
2780 (String::from("version"), Value::I64(1)),
2781 (String::from("name"), Value::Text(String::from("skip"))),
2782 ]),
2783 ],
2784 );
2785
2786 let rows = data_service
2787 .fetch_all(
2788 &teaql_core::SelectQuery::new("Order")
2789 .filter(Expr::compare_columns("version", BinaryOp::Gte, "id"))
2790 .order_asc("id"),
2791 )
2792 .unwrap();
2793
2794 assert_eq!(rows.len(), 1);
2795 assert_eq!(rows[0].get("name"), Some(&Value::Text("keep".to_owned())));
2796 }
2797
2798 #[tokio::test]
2799 async fn memory_data_service_supports_mutations_and_optimistic_locking() {
2800 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2801 let data_service = MemoryDataService::new(metadata);
2802
2803 data_service
2804 .insert(
2805 &InsertCommand::new("Order")
2806 .value("id", 10_u64)
2807 .value("version", 1_i64)
2808 .value("name", "draft"),
2809 )
2810 .unwrap();
2811 data_service
2812 .update(
2813 &UpdateCommand::new("Order", 10_u64)
2814 .expected_version(1)
2815 .value("name", "submitted"),
2816 )
2817 .unwrap();
2818
2819 let row = data_service
2820 .fetch_all(&teaql_core::SelectQuery::new("Order").filter(Expr::eq("id", 10_u64)))
2821 .unwrap()
2822 .pop()
2823 .unwrap();
2824 assert_eq!(
2825 row.get("name"),
2826 Some(&Value::Text(String::from("submitted")))
2827 );
2828 assert_eq!(row.get("version"), Some(&Value::I64(2)));
2829
2830 let conflict = data_service
2831 .update(
2832 &UpdateCommand::new("Order", 10_u64)
2833 .expected_version(1)
2834 .value("name", "stale"),
2835 )
2836 .unwrap_err();
2837 assert!(matches!(
2838 conflict,
2839 DataServiceError::Runtime(RuntimeError::OptimisticLockConflict { .. })
2840 ));
2841
2842 data_service
2843 .delete(&DeleteCommand::new("Order", 10_u64).expected_version(2))
2844 .unwrap();
2845 let row = data_service
2846 .fetch_all(&teaql_core::SelectQuery::new("Order").filter(Expr::eq("id", 10_u64)))
2847 .unwrap()
2848 .pop()
2849 .unwrap();
2850 assert_eq!(row.get("version"), Some(&Value::I64(-3)));
2851
2852 data_service
2853 .recover(&RecoverCommand::new("Order", 10_u64, -3))
2854 .unwrap();
2855 let row = data_service
2856 .fetch_all(&teaql_core::SelectQuery::new("Order").filter(Expr::eq("id", 10_u64)))
2857 .unwrap()
2858 .pop()
2859 .unwrap();
2860 assert_eq!(row.get("version"), Some(&Value::I64(4)));
2861 }
2862
2863 #[tokio::test]
2864 async fn user_context_reports_missing_schema_provider() {
2865 let err = UserContext::new().ensure_schema().await.unwrap_err();
2866 assert!(
2867 matches!(err, RuntimeError::Schema(message) if message == "missing schema provider")
2868 );
2869 }
2870
2871 #[tokio::test]
2872 async fn user_context_stores_and_exposes_user_identifier() {
2873 let mut context = UserContext::new();
2874 let pid = std::process::id();
2875 let thread_id_str = format!("{:?}", std::thread::current().id());
2876 let numeric_thread_id = thread_id_str
2877 .strip_prefix("ThreadId(")
2878 .and_then(|s| s.strip_suffix(")"))
2879 .unwrap_or(&thread_id_str);
2880 let os_user = std::env::var("USER")
2881 .or_else(|_| std::env::var("USERNAME"))
2882 .unwrap_or_else(|_| "main".to_owned());
2883 let expected_default = format!("{os_user}@pid-{pid}.tid-{numeric_thread_id}");
2884 assert_eq!(context.user_identifier(), Some(expected_default.as_str()));
2885
2886 context.set_user_identifier("user-123");
2887 assert_eq!(context.user_identifier(), Some("user-123"));
2888
2889 let ctx2 = UserContext::new().with_user_identifier("user-456");
2890 assert_eq!(ctx2.user_identifier(), Some("user-456"));
2891
2892 let mut ctx3 = UserContext::new();
2893 ctx3.set_user_identifier_option(Some("user-789".to_owned()));
2894 assert_eq!(ctx3.user_identifier(), Some("user-789"));
2895 ctx3.set_user_identifier_option(None);
2896 assert_eq!(ctx3.user_identifier(), None);
2897
2898 let ctx4 = UserContext::new().with_user_identifier_option(Some("user-abc".to_owned()));
2899 assert_eq!(ctx4.user_identifier(), Some("user-abc"));
2900 }
2901
2902 #[test]
2903 fn local_lock_enforces_ownership_timeout_and_lease_expiry() {
2904 let first = UserContext::new();
2905 let second = UserContext::new();
2906 let key = format!("local-lock-{:?}", std::time::SystemTime::now());
2907
2908 assert!(first.try_local_lock(&key, 0, 50));
2909 assert!(!second.try_local_lock(&key, 0, 50));
2910 second.unlock_local(&key);
2911 assert!(!second.try_local_lock(&key, 0, 50));
2912 std::thread::sleep(std::time::Duration::from_millis(60));
2913 assert!(second.try_local_lock(&key, 0, 50));
2914 second.unlock_local(&key);
2915 assert!(first.try_local_lock(&key, 0, 50));
2916 first.unlock_local(&key);
2917 }
2918
2919 #[derive(Default)]
2920 struct TestRemoteLockProvider {
2921 owners: Mutex<std::collections::HashMap<String, String>>,
2922 }
2923
2924 #[async_trait::async_trait]
2925 impl RemoteLockProvider for TestRemoteLockProvider {
2926 async fn try_remote_lock(
2927 &self,
2928 key: &str,
2929 owner_token: &str,
2930 _timeout_millis: u64,
2931 _expire_millis: u64,
2932 ) -> bool {
2933 let mut owners = self.owners.lock().expect("remote lock state");
2934 if owners.contains_key(key) {
2935 return false;
2936 }
2937 owners.insert(key.to_owned(), owner_token.to_owned());
2938 true
2939 }
2940
2941 async fn unlock_remote(&self, key: &str, owner_token: &str) -> bool {
2942 let mut owners = self.owners.lock().expect("remote lock state");
2943 if owners.get(key).is_some_and(|owner| owner == owner_token) {
2944 owners.remove(key);
2945 return true;
2946 }
2947 false
2948 }
2949 }
2950
2951 #[tokio::test]
2952 async fn remote_lock_delegates_and_preserves_context_ownership() {
2953 let provider: Arc<dyn RemoteLockProvider> = Arc::new(TestRemoteLockProvider::default());
2954 let mut first = UserContext::new();
2955 first.insert_resource(provider.clone());
2956 let mut second = UserContext::new();
2957 second.insert_resource(provider);
2958 let key = format!("remote-lock-{:?}", std::time::SystemTime::now());
2959
2960 assert!(first.try_remote_lock(&key, 0, 1_000).await);
2961 assert!(!second.try_remote_lock(&key, 0, 1_000).await);
2962 assert!(!second.unlock_remote(&key).await);
2963 assert!(!second.try_remote_lock(&key, 0, 1_000).await);
2964 assert!(first.unlock_remote(&key).await);
2965 assert!(second.try_remote_lock(&key, 0, 1_000).await);
2966 assert!(second.unlock_remote(&key).await);
2967
2968 assert!(UserContext::new().try_remote_lock("optional", 0, 0).await);
2969 }
2970}
2971
2972pub use checker::{
2973 CHECK_OBJECT_STATUS_FIELD, CheckObjectStatus, CheckResult, CheckResults, CheckRule, Checker,
2974 CheckerRegistry, InMemoryCheckerRegistry, LocationSegment, ObjectLocation, TypedChecker,
2975 TypedEntityChecker, clear_entity_status, mark_entity_status,
2976};