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