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