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