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