1#![allow(warnings)]
2mod checker;
3mod context;
4mod data_service;
5mod entity_runtime;
6mod entity_status;
7mod error;
8mod event;
9pub mod generated_support;
10mod graph;
11mod id;
12mod language;
13pub mod log_formatter;
14mod memory;
15mod registry;
16
17pub use context::{
18 DataStore, InMemoryDataStore, InfoLogEntry, LogPayload, SchemaProvider, SqlLogEntry,
19 SqlLogOperation, SqlLogOptions, UnifiedLogBuffer, UnifiedLogEntry, UserContext,
20};
21pub use data_service::{
22 AggregationCacheBackend, ContextDataService, EntityDataService, GraphTransactionBoundary,
23 InMemoryAggregationCache, RelationLoadPlan, RuntimeDataService,
24};
25pub use entity_runtime::{
26 ChangeSetStack, EntityChangeSet, EntityKey, EntityRoot, LedgerEntity, RootContext,
27};
28pub use entity_status::{EntityAction, EntityStatus};
29pub use error::{ContextError, DataServiceError, RuntimeError};
30pub use event::{
31 EntityPropertyChange, InMemoryRawAuditEventSink, RawAuditEvent, RawAuditEventKind,
32 RawAuditEventSink, SafeAuditEvent, SafeAuditEventSink, SafeAuditField,
33};
34pub use generated_support::*;
35pub use graph::{
36 GraphMutationBatch, GraphMutationKind, GraphMutationPlan, GraphMutationPlanItem, GraphNode,
37 GraphOperation, ScopedCommentNode, TraceScopeToken, sorted_update_fields,
38};
39pub(crate) use id::local_id_generator;
40pub use id::{InternalIdGenerator, SnowflakeIdGenerator};
41pub use language::{
42 BuiltinTranslator, Language, MessageTranslator, translate_check_result, translate_location,
43};
44pub use memory::{MemoryDataService, MemoryDataServiceError};
45pub use registry::{
46 EntityDataServiceBehavior, EntityDataServiceBehaviorRegistry, EntityRegistry,
47 InMemoryEntityDataServiceBehaviorRegistry, InMemoryEntityRegistry, InMemoryMetadataStore,
48 MetadataStore, RequestPolicy, RuntimeModule,
49};
50
51#[cfg(test)]
52mod tests {
53 use std::collections::{BTreeMap, VecDeque};
54 use std::sync::{Arc, Mutex};
55
56 use super::{
57 AggregationCacheBackend, CHECK_OBJECT_STATUS_FIELD, CheckObjectStatus, CheckResult,
58 CheckResults, CheckRule, Checker, DataServiceError, EntityDataServiceBehavior,
59 GraphMutationKind, GraphNode, InMemoryAggregationCache, InMemoryCheckerRegistry,
60 InMemoryEntityDataServiceBehaviorRegistry, InMemoryEntityRegistry, InMemoryMetadataStore,
61 InternalIdGenerator, Language, MemoryDataService, MetadataStore, ObjectLocation,
62 RawAuditEvent, RawAuditEventKind, RawAuditEventSink, RequestPolicy, RuntimeDataService,
63 RuntimeError, RuntimeModule, SqlLogOperation, SqlLogOptions, TypedChecker,
64 TypedEntityChecker, UserContext, translate_check_result,
65 };
66 use teaql_core::{
67 Aggregate, AggregateFunction, BinaryOp, DataType, Decimal, DeleteCommand, Entity,
68 EntityDescriptor, EntityError, Expr, InsertCommand, OrderBy, PropertyDescriptor, Record,
69 RecoverCommand, RelationAggregate, SelectQuery, TeaqlEntity, UpdateCommand, Value,
70 };
71 use teaql_data_service::{
72 DataServiceCapabilities, DataServiceExecutor, DataServiceOperation, ExecutionMetadata,
73 MutationExecutor, MutationRequest, MutationResult, QueryExecutor, QueryRequest,
74 QueryResult,
75 };
76 use teaql_macros::TeaqlEntity as DeriveTeaqlEntity;
77 use teaql_sql::{
78 CompiledQuery, DatabaseKind, SqlCompileError, SqlDialect, quote_identifier_if_needed,
79 };
80
81 const ORDER_DEFAULT_PROJECTION: &str = "id, version, name";
82
83 #[derive(Debug, Default, Clone, Copy)]
84 struct PostgresDialect;
85
86 impl SqlDialect for PostgresDialect {
87 fn kind(&self) -> DatabaseKind {
88 DatabaseKind::PostgreSql
89 }
90
91 fn quote_ident(&self, ident: &str) -> String {
92 quote_identifier_if_needed(ident, '"')
93 }
94
95 fn placeholder(&self, index: usize) -> String {
96 format!("${index}")
97 }
98
99 fn schema_type_sql(
100 &self,
101 data_type: DataType,
102 _property: &PropertyDescriptor,
103 ) -> Result<&'static str, SqlCompileError> {
104 match data_type {
105 DataType::Bool => Ok("BOOLEAN"),
106 DataType::I64 | DataType::U64 => Ok("BIGINT"),
107 DataType::F64 => Ok("DOUBLE PRECISION"),
108 DataType::Decimal => Ok("NUMERIC"),
109 DataType::Text => Ok("TEXT"),
110 DataType::Json => Ok("JSONB"),
111 DataType::Date => Ok("DATE"),
112 DataType::Timestamp => Ok("TIMESTAMPTZ"),
113 }
114 }
115 }
116
117 fn entity() -> EntityDescriptor {
118 EntityDescriptor::new("Order")
119 .table_name("orders")
120 .property(
121 PropertyDescriptor::new("id", DataType::U64)
122 .column_name("id")
123 .id()
124 .not_null(),
125 )
126 .property(
127 PropertyDescriptor::new("version", DataType::I64)
128 .column_name("version")
129 .version()
130 .not_null(),
131 )
132 .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
133 .relation(
134 teaql_core::RelationDescriptor::new("lines", "OrderLine")
135 .local_key("id")
136 .foreign_key("order_id")
137 .many(),
138 )
139 }
140
141 fn line_entity() -> EntityDescriptor {
142 EntityDescriptor::new("OrderLine")
143 .table_name("orderline")
144 .property(
145 PropertyDescriptor::new("id", DataType::U64)
146 .column_name("id")
147 .id()
148 .not_null(),
149 )
150 .property(
151 PropertyDescriptor::new("version", DataType::I64)
152 .column_name("version")
153 .version(),
154 )
155 .property(
156 PropertyDescriptor::new("order_id", DataType::U64)
157 .column_name("order_id")
158 .not_null(),
159 )
160 .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
161 .property(
162 PropertyDescriptor::new("product_id", DataType::U64)
163 .column_name("product_id")
164 .not_null(),
165 )
166 .relation(
167 teaql_core::RelationDescriptor::new("product", "Product")
168 .local_key("product_id")
169 .foreign_key("id"),
170 )
171 }
172
173 fn product_entity() -> EntityDescriptor {
174 EntityDescriptor::new("Product")
175 .table_name("product")
176 .property(
177 PropertyDescriptor::new("id", DataType::U64)
178 .column_name("id")
179 .id()
180 .not_null(),
181 )
182 .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
183 }
184
185 #[derive(Debug, Default)]
186 struct StubExecutor {
187 affected: u64,
188 rows: Vec<Record>,
189 }
190
191 #[derive(Debug, Default)]
192 struct QueueExecutor {
193 affected: u64,
194 rows: Mutex<VecDeque<Vec<Record>>>,
195 queries: Mutex<Vec<String>>,
196 }
197
198 struct OrderBehavior;
199
200 #[allow(dead_code)]
201 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
202 #[teaql(entity = "CatalogProduct", table = "catalog_product")]
203 struct CatalogProductRow {
204 #[teaql(id)]
205 id: u64,
206 name: String,
207 }
208
209 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
210 #[teaql(entity = "OrderAggregate", table = "orders")]
211 struct OrderAggregateDynamic {
212 #[teaql(id)]
213 id: u64,
214 #[teaql(dynamic)]
215 dynamic: BTreeMap<String, Value>,
216 }
217
218 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
219 #[teaql(entity = "Product", table = "product")]
220 struct ProductEntityRow {
221 #[teaql(id)]
222 id: u64,
223 name: String,
224 }
225
226 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
227 #[teaql(entity = "OrderLine", table = "orderline")]
228 struct OrderLineEntityRow {
229 #[teaql(id)]
230 id: u64,
231 #[teaql(column = "order_id")]
232 order_id: u64,
233 name: String,
234 #[teaql(column = "product_id")]
235 product_id: u64,
236 #[teaql(relation(target = "Product", local_key = "product_id", foreign_key = "id"))]
237 product: Option<ProductEntityRow>,
238 }
239
240 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
241 #[teaql(entity = "OrderLine", table = "orderline")]
242 struct ProductLineEntityRow {
243 #[teaql(id)]
244 id: u64,
245 #[teaql(column = "order_id")]
246 order_id: u64,
247 name: String,
248 #[teaql(column = "product_id")]
249 product_id: u64,
250 }
251
252 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
253 #[teaql(entity = "Product", table = "product")]
254 struct ProductWithLinesEntityRow {
255 #[teaql(id)]
256 id: u64,
257 name: String,
258 #[teaql(relation(
259 target = "OrderLine",
260 local_key = "id",
261 foreign_key = "product_id",
262 many
263 ))]
264 lines: teaql_core::SmartList<ProductLineEntityRow>,
265 }
266
267 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
268 #[teaql(entity = "OrderLine", table = "orderline")]
269 struct OrderLineWithProductEntityRow {
270 #[teaql(id)]
271 id: u64,
272 #[teaql(column = "order_id")]
273 order_id: u64,
274 name: String,
275 #[teaql(column = "product_id")]
276 product_id: u64,
277 #[teaql(relation(target = "Product", local_key = "product_id", foreign_key = "id"))]
278 product: Option<ProductWithLinesEntityRow>,
279 }
280
281 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
282 #[teaql(entity = "Order", table = "orders")]
283 struct OrderAggregateRow {
284 #[teaql(id)]
285 id: u64,
286 #[teaql(version)]
287 version: i64,
288 name: String,
289 #[teaql(relation(target = "OrderLine", local_key = "id", foreign_key = "order_id", many))]
290 lines: teaql_core::SmartList<OrderLineEntityRow>,
291 }
292
293 #[derive(Debug, Clone, PartialEq, DeriveTeaqlEntity)]
294 #[teaql(entity = "Order", table = "orders")]
295 struct Order {
296 #[teaql(id)]
297 id: u64,
298 #[teaql(version)]
299 version: i64,
300 name: String,
301 }
302
303 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
304 #[teaql(entity = "Product", table = "product")]
305 struct TypedGraphProduct {
306 #[teaql(id)]
307 id: u64,
308 name: String,
309 }
310
311 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
312 #[teaql(entity = "OrderLine", table = "orderline")]
313 struct TypedGraphLine {
314 #[teaql(id)]
315 id: u64,
316 #[teaql(column = "order_id")]
317 order_id: Option<u64>,
318 name: String,
319 #[teaql(column = "product_id")]
320 product_id: Option<u64>,
321 #[teaql(relation(target = "Product", local_key = "product_id", foreign_key = "id"))]
322 product: Option<TypedGraphProduct>,
323 }
324
325 #[derive(Debug, PartialEq, DeriveTeaqlEntity)]
326 #[teaql(entity = "Order", table = "orders")]
327 struct TypedGraphOrder {
328 #[teaql(id)]
329 id: u64,
330 #[teaql(version)]
331 version: i64,
332 name: String,
333 #[teaql(relation(target = "OrderLine", local_key = "id", foreign_key = "order_id", many))]
334 lines: teaql_core::SmartList<TypedGraphLine>,
335 }
336
337 #[derive(Debug, PartialEq, Eq)]
338 struct OrderEntity {
339 id: u64,
340 version: i64,
341 name: String,
342 }
343
344 impl teaql_core::TeaqlEntity for OrderEntity {
345 fn entity_descriptor() -> EntityDescriptor {
346 entity()
347 }
348 }
349
350 impl Entity for OrderEntity {
351 fn from_record(record: Record) -> Result<Self, EntityError> {
352 let id = match record.get("id") {
353 Some(Value::U64(v)) => *v,
354 Some(Value::I64(v)) if *v >= 0 => *v as u64,
355 other => {
356 return Err(EntityError::new(
357 "Order",
358 format!("invalid id field: {other:?}"),
359 ));
360 }
361 };
362 let version = match record.get("version") {
363 Some(Value::I64(v)) => *v,
364 other => {
365 return Err(EntityError::new(
366 "Order",
367 format!("invalid version field: {other:?}"),
368 ));
369 }
370 };
371 let name = match record.get("name") {
372 Some(Value::Text(v)) => v.clone(),
373 other => {
374 return Err(EntityError::new(
375 "Order",
376 format!("invalid name field: {other:?}"),
377 ));
378 }
379 };
380 Ok(Self { id, version, name })
381 }
382
383 fn into_record(self) -> Record {
384 Record::from([
385 (String::from("id"), Value::U64(self.id)),
386 (String::from("version"), Value::I64(self.version)),
387 (String::from("name"), Value::Text(self.name)),
388 ])
389 }
390 }
391
392 #[derive(Debug)]
393 struct StubError;
394
395 impl std::fmt::Display for StubError {
396 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
397 write!(f, "stub error")
398 }
399 }
400
401 impl std::error::Error for StubError {}
402
403 impl DataServiceExecutor for StubExecutor {
404 type Error = StubError;
405
406 fn capabilities(&self) -> DataServiceCapabilities {
407 DataServiceCapabilities::default()
408 }
409 }
410
411 impl QueryExecutor for StubExecutor {
412 async fn query(&self, _request: QueryRequest) -> Result<QueryResult, Self::Error> {
413 Ok(QueryResult {
414 rows: self.rows.clone(),
415 metadata: ExecutionMetadata {
416 debug_query: None,
417 backend: "stub".to_owned(),
418 operation: DataServiceOperation::Query,
419 started_at: std::time::SystemTime::now(),
420 ended_at: std::time::SystemTime::now(),
421 affected_rows: None,
422 result_count: Some(self.rows.len()),
423 trace_chain: Vec::new(),
424 comment: None,
425 backend_request_id: None,
426 },
427 })
428 }
429 }
430
431 impl MutationExecutor for StubExecutor {
432 async fn mutate(&self, _request: MutationRequest) -> Result<MutationResult, Self::Error> {
433 Ok(MutationResult {
434 affected_rows: self.affected,
435 generated_values: Record::new(),
436 metadata: ExecutionMetadata {
437 debug_query: None,
438 backend: "stub".to_owned(),
439 operation: DataServiceOperation::Update,
440 started_at: std::time::SystemTime::now(),
441 ended_at: std::time::SystemTime::now(),
442 affected_rows: Some(self.affected),
443 result_count: None,
444 trace_chain: Vec::new(),
445 comment: None,
446 backend_request_id: None,
447 },
448 })
449 }
450 }
451
452 impl DataServiceExecutor for QueueExecutor {
453 type Error = StubError;
454
455 fn capabilities(&self) -> DataServiceCapabilities {
456 DataServiceCapabilities::default()
457 }
458 }
459
460 impl QueryExecutor for QueueExecutor {
461 async fn query(&self, request: QueryRequest) -> Result<QueryResult, Self::Error> {
462 let sql_approx = format!("SELECT ... FROM {} ...", request.query.entity);
463 self.queries.lock().unwrap().push(sql_approx);
464 Ok(QueryResult {
465 rows: self.rows.lock().unwrap().pop_front().unwrap_or_default(),
466 metadata: ExecutionMetadata {
467 debug_query: None,
468 backend: "queue".to_owned(),
469 operation: DataServiceOperation::Query,
470 started_at: std::time::SystemTime::now(),
471 ended_at: std::time::SystemTime::now(),
472 affected_rows: None,
473 result_count: Some(0),
474 trace_chain: Vec::new(),
475 comment: None,
476 backend_request_id: None,
477 },
478 })
479 }
480 }
481
482 impl MutationExecutor for QueueExecutor {
483 async fn mutate(&self, _request: MutationRequest) -> Result<MutationResult, Self::Error> {
484 Ok(MutationResult {
485 affected_rows: self.affected,
486 generated_values: Record::new(),
487 metadata: ExecutionMetadata {
488 debug_query: None,
489 backend: "queue".to_owned(),
490 operation: DataServiceOperation::Update,
491 started_at: std::time::SystemTime::now(),
492 ended_at: std::time::SystemTime::now(),
493 affected_rows: Some(self.affected),
494 result_count: None,
495 trace_chain: Vec::new(),
496 comment: None,
497 backend_request_id: None,
498 },
499 })
500 }
501 }
502
503 impl EntityDataServiceBehavior for OrderBehavior {
504 fn before_select(
505 &self,
506 _ctx: &UserContext,
507 query: &mut teaql_core::SelectQuery,
508 ) -> Result<(), RuntimeError> {
509 query.filter = Some(Expr::eq("version", 1_i64));
510 Ok(())
511 }
512
513 fn before_insert(
514 &self,
515 _ctx: &UserContext,
516 command: &mut InsertCommand,
517 ) -> Result<(), RuntimeError> {
518 command
519 .values
520 .entry("version".to_owned())
521 .or_insert(Value::I64(1));
522 Ok(())
523 }
524
525 fn relation_loads(&self, _ctx: &UserContext) -> Vec<String> {
526 vec!["lines".to_owned()]
527 }
528 }
529
530 struct ContextAwareOrderBehavior;
531 struct TenantRequestPolicy;
532 struct OrderChecker;
533 struct TypedOrderChecker;
534 #[derive(Clone)]
535 struct RecordingEventSink {
536 events: Arc<Mutex<Vec<RawAuditEvent>>>,
537 }
538
539 impl EntityDataServiceBehavior for ContextAwareOrderBehavior {
540 fn before_insert(
541 &self,
542 ctx: &UserContext,
543 command: &mut InsertCommand,
544 ) -> Result<(), RuntimeError> {
545 let tenant = ctx
546 .get_named_resource::<String>("tenant")
547 .cloned()
548 .ok_or_else(|| RuntimeError::Behavior("missing tenant resource".to_owned()))?;
549 let version = *ctx
550 .get_named_resource::<i64>("initial_version")
551 .ok_or_else(|| {
552 RuntimeError::Behavior("missing initial_version resource".to_owned())
553 })?;
554 let trace_id = match ctx.local("trace_id") {
555 Some(Value::Text(value)) => value.clone(),
556 other => {
557 return Err(RuntimeError::Behavior(format!(
558 "missing trace_id local, got {other:?}"
559 )));
560 }
561 };
562
563 command
564 .values
565 .entry("name".to_owned())
566 .or_insert(Value::Text(format!("{tenant}:{trace_id}")));
567 command
568 .values
569 .entry("version".to_owned())
570 .or_insert(Value::I64(version));
571 Ok(())
572 }
573 }
574
575 impl RequestPolicy for TenantRequestPolicy {
576 fn enforce_select(
577 &self,
578 ctx: &UserContext,
579 query: &mut SelectQuery,
580 ) -> Result<(), RuntimeError> {
581 if query.entity == "Order" {
582 let tenant_id = ctx
583 .get_named_resource::<u64>("tenant_id")
584 .copied()
585 .ok_or_else(|| RuntimeError::Policy("missing tenant_id".to_owned()))?;
586 query.filter = Some(match query.filter.take() {
587 Some(filter) => filter.and_expr(Expr::eq("id", tenant_id)),
588 None => Expr::eq("id", tenant_id),
589 });
590 }
591 Ok(())
592 }
593
594 fn enforce_insert(
595 &self,
596 ctx: &UserContext,
597 command: &mut InsertCommand,
598 ) -> Result<(), RuntimeError> {
599 if command.entity == "Order" {
600 let tenant_id = ctx
601 .get_named_resource::<u64>("tenant_id")
602 .copied()
603 .ok_or_else(|| RuntimeError::Policy("missing tenant_id".to_owned()))?;
604 command
605 .values
606 .insert("version".to_owned(), Value::I64(tenant_id as i64));
607 }
608 Ok(())
609 }
610 }
611
612 impl Checker for OrderChecker {
613 fn entity(&self) -> &str {
614 "Order"
615 }
616
617 fn check_and_fix(
618 &self,
619 _ctx: &UserContext,
620 record: &mut Record,
621 location: &ObjectLocation,
622 results: &mut CheckResults,
623 ) {
624 let status = CheckObjectStatus::from_record(record);
625 if status.is_create() {
626 self.required(record, "name", location, results);
627 record.entry("version".to_owned()).or_insert(Value::I64(1));
628 }
629 if status.is_update()
630 && record.get("name") == Some(&Value::Text("graph-update".to_owned()))
631 {
632 record.insert(
633 "name".to_owned(),
634 Value::Text("graph-update-checked".to_owned()),
635 );
636 }
637 self.min_string_length(record, "name", 3, location, results);
638 }
639 }
640
641 impl TypedChecker<Order> for TypedOrderChecker {
642 fn check_and_fix_typed(
643 &self,
644 _ctx: &UserContext,
645 entity: &mut Order,
646 status: CheckObjectStatus,
647 location: &ObjectLocation,
648 results: &mut CheckResults,
649 ) {
650 if status.is_create() {
651 if entity.name.is_empty() {
652 results.push(CheckResult::required(location.clone().member("name")));
653 }
654 }
655 if entity.name.chars().count() < 3 {
656 results.push(CheckResult::min_str(
657 location.clone().member("name"),
658 3,
659 entity.name.clone(),
660 ));
661 }
662 if entity.name == "fix" {
663 entity.name = "fixed".to_owned();
664 }
665 }
666 }
667
668 impl RawAuditEventSink for RecordingEventSink {
669 fn on_event(&self, _ctx: &UserContext, event: &RawAuditEvent) -> Result<(), RuntimeError> {
670 self.events.lock().unwrap().push(event.clone());
671 Ok(())
672 }
673 }
674
675 struct FixedIdGenerator(u64);
676
677 impl InternalIdGenerator for FixedIdGenerator {
678 fn generate_id(&self, _entity: &str) -> Result<u64, RuntimeError> {
679 Ok(self.0)
680 }
681 }
682
683 struct SequentialIdGenerator {
684 next: Mutex<u64>,
685 }
686
687 impl SequentialIdGenerator {
688 fn new(next: u64) -> Self {
689 Self {
690 next: Mutex::new(next),
691 }
692 }
693 }
694
695 impl InternalIdGenerator for SequentialIdGenerator {
696 fn generate_id(&self, _entity: &str) -> Result<u64, RuntimeError> {
697 let mut next = self
698 .next
699 .lock()
700 .map_err(|err| RuntimeError::IdGeneration(err.to_string()))?;
701 let id = *next;
702 *next += 1;
703 Ok(id)
704 }
705 }
706
707 #[tokio::test]
708 async fn metadata_store_registers_entities() {
709 let store = InMemoryMetadataStore::new().with_entity(entity());
710 assert!(store.entity("Order").is_some());
711 }
712
713 #[tokio::test]
714 async fn runtime_module_registers_descriptor_into_context() {
715 let ctx = UserContext::new().with_module(RuntimeModule::new().descriptor(entity()));
716 assert!(ctx.entity("Order").is_some());
717 assert!(ctx.has_entity_data_service("Order"));
718 }
719
720 #[tokio::test]
721 async fn runtime_module_registers_derived_entity_and_behavior() {
722 let ctx = UserContext::new().with_module(
723 RuntimeModule::new().entity_with_behavior::<CatalogProductRow, _>(OrderBehavior),
724 );
725 assert!(ctx.entity("CatalogProduct").is_some());
726 assert!(ctx.has_entity_data_service("CatalogProduct"));
727 assert!(ctx.entity_data_service_behavior("CatalogProduct").is_some());
728 }
729
730 #[tokio::test]
731 async fn module_macro_registers_multiple_entities() {
732 let ctx = UserContext::new().with_module(crate::module!(CatalogProductRow));
733 assert!(ctx.entity("CatalogProduct").is_some());
734 assert!(ctx.has_entity_data_service("CatalogProduct"));
735 }
736
737 #[tokio::test]
738 async fn module_macro_registers_entity_behavior_pairs() {
739 let ctx =
740 UserContext::new().with_module(crate::module!(CatalogProductRow => OrderBehavior));
741 assert!(ctx.entity("CatalogProduct").is_some());
742 assert!(ctx.entity_data_service_behavior("CatalogProduct").is_some());
743 }
744
745 #[tokio::test]
746 async fn data_service_returns_optimistic_lock_conflict() {
747 let store = InMemoryMetadataStore::new().with_entity(entity());
748 let executor = StubExecutor {
749 affected: 0,
750 rows: Vec::new(),
751 };
752 let repo = RuntimeDataService::new(&store, &executor);
753
754 let err = repo
755 .update(
756 &UpdateCommand::new("Order", 1_u64)
757 .expected_version(3)
758 .value("name", "next"),
759 )
760 .await
761 .unwrap_err();
762
763 match err {
764 DataServiceError::Runtime(RuntimeError::OptimisticLockConflict { .. }) => {}
765 other => panic!("unexpected error: {other}"),
766 }
767 }
768
769 #[tokio::test]
770 async fn user_context_indexes_resources_and_locals() {
771 let mut ctx =
772 UserContext::new().with_metadata(InMemoryMetadataStore::new().with_entity(entity()));
773 ctx.insert_resource::<u64>(42);
774 ctx.insert_named_resource("tenant", String::from("acme"));
775 ctx.put_local("trace_id", "req-1");
776
777 assert!(ctx.entity("Order").is_some());
778 assert_eq!(ctx.get_resource::<u64>(), Some(&42));
779 assert_eq!(
780 ctx.get_named_resource::<String>("tenant"),
781 Some(&String::from("acme"))
782 );
783 assert_eq!(
784 ctx.local("trace_id"),
785 Some(&Value::Text("req-1".to_owned()))
786 );
787 }
788
789 #[tokio::test]
790 async fn user_context_builds_context_data_service() {
791 let mut ctx =
792 UserContext::new().with_metadata(InMemoryMetadataStore::new().with_entity(entity()));
793 ctx.insert_resource(PostgresDialect);
794 ctx.insert_resource(StubExecutor {
795 affected: 1,
796 rows: Vec::new(),
797 });
798
799 let repo = ctx.data_service::<StubExecutor>().unwrap();
800 let affected = repo
801 .update(
802 &UpdateCommand::new("Order", 1_u64)
803 .expected_version(3)
804 .value("name", "next"),
805 )
806 .await
807 .unwrap();
808
809 assert_eq!(affected, 1);
810 }
811
812 #[tokio::test]
813 async fn user_context_resolves_entity_data_service_by_entity_type() {
814 let mut ctx = UserContext::new()
815 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
816 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
817 ctx.insert_resource(PostgresDialect);
818 ctx.insert_resource(StubExecutor {
819 affected: 1,
820 rows: Vec::new(),
821 });
822
823 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
824 assert_eq!(repo.entity(), "Order");
825 assert_eq!(repo.select().entity, "Order");
826
827 let affected = repo
828 .insert(
829 &repo
830 .insert_command()
831 .value("id", 1_u64)
832 .value("version", 1_i64)
833 .value("name", "n"),
834 )
835 .await
836 .unwrap();
837 assert_eq!(affected, 1);
838 }
839
840 #[tokio::test]
841 async fn entity_data_service_applies_behavior_hooks() {
842 let mut ctx = UserContext::new()
843 .with_metadata(
844 InMemoryMetadataStore::new()
845 .with_entity(entity())
846 .with_entity(line_entity())
847 .with_entity(product_entity()),
848 )
849 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
850 .with_entity_data_service_behavior_registry(
851 InMemoryEntityDataServiceBehaviorRegistry::new()
852 .with_behavior("Order", OrderBehavior),
853 );
854 ctx.insert_resource(PostgresDialect);
855 ctx.insert_resource(StubExecutor {
856 affected: 1,
857 rows: Vec::new(),
858 });
859
860 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
861
862 let insert = repo.insert_command().value("id", 1_u64).value("name", "n");
866 let affected = repo.insert(&insert).await.unwrap();
867 assert_eq!(affected, 1);
868 assert_eq!(repo.relation_loads(), vec!["lines".to_owned()]);
869 }
870
871 #[tokio::test]
872 async fn entity_data_service_applies_request_policy_after_behavior_hooks() {
873 let mut ctx = UserContext::new()
874 .with_metadata(
875 InMemoryMetadataStore::new()
876 .with_entity(entity())
877 .with_entity(line_entity())
878 .with_entity(product_entity()),
879 )
880 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
881 .with_entity_data_service_behavior_registry(
882 InMemoryEntityDataServiceBehaviorRegistry::new()
883 .with_behavior("Order", OrderBehavior),
884 )
885 .with_request_policy(TenantRequestPolicy);
886 ctx.insert_named_resource("tenant_id", 9_u64);
887 ctx.insert_resource(PostgresDialect);
888 ctx.insert_resource(StubExecutor {
889 affected: 1,
890 rows: Vec::new(),
891 });
892
893 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
894
895 let insert = repo.insert_command().value("id", 1_u64).value("name", "n");
900 let command = repo.prepare_insert_command(&insert).unwrap();
901 assert_eq!(command.values.get("version"), Some(&Value::I64(9)));
902 }
903
904 #[tokio::test]
905 async fn entity_data_service_prepares_insert_command_with_generated_id() {
906 let mut ctx = UserContext::new()
907 .with_metadata(
908 InMemoryMetadataStore::new()
909 .with_entity(entity())
910 .with_entity(line_entity())
911 .with_entity(product_entity()),
912 )
913 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
914 .with_entity_data_service_behavior_registry(
915 InMemoryEntityDataServiceBehaviorRegistry::new()
916 .with_behavior("Order", OrderBehavior),
917 )
918 .with_internal_id_generator(FixedIdGenerator(42));
919 ctx.insert_resource(PostgresDialect);
920 ctx.insert_resource(StubExecutor {
921 affected: 1,
922 rows: Vec::new(),
923 });
924
925 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
926
927 let prepared = repo
928 .prepare_insert_command(&repo.insert_command().value("id", 0_u64).value("name", "n"))
929 .unwrap();
930
931 assert_eq!(prepared.values.get("id"), Some(&Value::U64(42)));
932 assert_eq!(prepared.values.get("version"), Some(&Value::I64(1)));
933 assert_eq!(
934 prepared.values.get("name"),
935 Some(&Value::Text("n".to_owned()))
936 );
937
938 let prepared_zero_version = repo
939 .prepare_insert_command(
940 &repo
941 .insert_command()
942 .value("id", 0_u64)
943 .value("version", 0_i64)
944 .value("name", "zero-version"),
945 )
946 .unwrap();
947 assert_eq!(
948 prepared_zero_version.values.get("version"),
949 Some(&Value::I64(1))
950 );
951 }
952
953 #[tokio::test]
954 async fn custom_user_context_can_drive_insert_preparation() {
955 let mut ctx = UserContext::new()
956 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
957 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
958 .with_entity_data_service_behavior_registry(
959 InMemoryEntityDataServiceBehaviorRegistry::new()
960 .with_behavior("Order", ContextAwareOrderBehavior),
961 )
962 .with_internal_id_generator(FixedIdGenerator(99));
963 ctx.insert_named_resource("tenant", String::from("acme"));
964 ctx.insert_named_resource("initial_version", 7_i64);
965 ctx.put_local("trace_id", "req-9");
966 ctx.insert_resource(PostgresDialect);
967 ctx.insert_resource(StubExecutor {
968 affected: 1,
969 rows: Vec::new(),
970 });
971
972 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
973 let prepared = repo.prepare_insert_command(&repo.insert_command()).unwrap();
974
975 assert_eq!(prepared.values.get("id"), Some(&Value::U64(99)));
976 assert_eq!(prepared.values.get("version"), Some(&Value::I64(7)));
977 assert_eq!(
978 prepared.values.get("name"),
979 Some(&Value::Text("acme:req-9".to_owned()))
980 );
981 }
982
983 #[tokio::test]
984 async fn checker_registry_validates_and_fixes_insert_commands() {
985 let mut ctx = UserContext::new()
986 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
987 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
988 .with_checker_registry(InMemoryCheckerRegistry::new().with_checker(OrderChecker))
989 .with_internal_id_generator(FixedIdGenerator(77));
990 ctx.insert_resource(PostgresDialect);
991 ctx.insert_resource(StubExecutor {
992 affected: 1,
993 rows: Vec::new(),
994 });
995
996 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
997 let prepared = repo
998 .prepare_insert_command(&repo.insert_command().value("name", "valid"))
999 .unwrap();
1000
1001 assert_eq!(prepared.values.get("id"), Some(&Value::U64(77)));
1002 assert_eq!(prepared.values.get("version"), Some(&Value::I64(1)));
1003 assert!(!prepared.values.contains_key(CHECK_OBJECT_STATUS_FIELD));
1004
1005 let error = repo
1006 .prepare_insert_command(&repo.insert_command().value("name", "no"))
1007 .unwrap_err();
1008 match error {
1009 RuntimeError::Check(results) => {
1010 assert_eq!(results.len(), 1);
1011 assert_eq!(results[0].location.to_string(), "name");
1012 }
1013 other => panic!("unexpected checker error: {other:?}"),
1014 }
1015 }
1016
1017 #[tokio::test]
1018 async fn typed_checker_validates_and_fixes_derived_entities_without_record_access() {
1019 let mut ctx = UserContext::new()
1020 .with_metadata(InMemoryMetadataStore::new().with_entity(Order::entity_descriptor()))
1021 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1022 .with_checker_registry(
1023 InMemoryCheckerRegistry::new()
1024 .with_checker(TypedEntityChecker::<Order, _>::new(TypedOrderChecker)),
1025 )
1026 .with_internal_id_generator(FixedIdGenerator(79));
1027 ctx.insert_resource(PostgresDialect);
1028 ctx.insert_resource(StubExecutor {
1029 affected: 1,
1030 rows: Vec::new(),
1031 });
1032
1033 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
1034 let prepared = repo
1035 .prepare_insert_command(
1036 &repo
1037 .insert_command()
1038 .value("name", "fix")
1039 .value("version", 1_i64),
1040 )
1041 .unwrap();
1042 assert_eq!(
1043 prepared.values.get("name"),
1044 Some(&Value::Text("fixed".to_owned()))
1045 );
1046 assert_eq!(prepared.values.get("id"), Some(&Value::U64(79)));
1047 assert!(!prepared.values.contains_key(CHECK_OBJECT_STATUS_FIELD));
1048
1049 let error = repo
1050 .prepare_insert_command(&repo.insert_command().value("version", 1_i64))
1051 .unwrap_err();
1052 match error {
1053 RuntimeError::Check(results) => {
1054 assert!(
1055 results
1056 .iter()
1057 .any(|result| result.rule == CheckRule::Required
1058 && result.location.to_string() == "name")
1059 );
1060 }
1061 other => panic!("unexpected typed checker error: {other:?}"),
1062 }
1063 }
1064
1065 #[tokio::test]
1066 async fn checker_registry_reports_nested_create_locations_and_fixes_records() {
1067 let ctx = UserContext::new()
1068 .with_checker_registry(InMemoryCheckerRegistry::new().with_checker(OrderChecker));
1069
1070 let mut child = Record::from([
1071 (String::from("id"), Value::U64(10)),
1072 (
1073 String::from(CHECK_OBJECT_STATUS_FIELD),
1074 Value::from(CheckObjectStatus::Create),
1075 ),
1076 ]);
1077 let error = ctx
1078 .check_and_fix_record_at(
1079 "Order",
1080 &mut child,
1081 &ObjectLocation::hash_root("lines").element(0),
1082 )
1083 .unwrap_err();
1084
1085 assert_eq!(child.get("version"), Some(&Value::I64(1)));
1086 match error {
1087 RuntimeError::Check(results) => {
1088 assert_eq!(results.len(), 1);
1089 assert_eq!(results[0].rule, CheckRule::Required);
1090 assert_eq!(results[0].location.to_string(), "lines[0].name");
1091 }
1092 other => panic!("unexpected checker error: {other:?}"),
1093 }
1094
1095 child.insert("name".to_owned(), Value::Text("valid child".to_owned()));
1096 ctx.check_and_fix_record_at(
1097 "Order",
1098 &mut child,
1099 &ObjectLocation::hash_root("lines").element(0),
1100 )
1101 .unwrap();
1102 }
1103
1104 #[tokio::test]
1105 async fn built_in_language_translators_cover_fifteen_languages() {
1106 assert_eq!(Language::ALL.len(), 15);
1107 let result = super::CheckResult::required(ObjectLocation::hash_root("name"));
1108 let messages = Language::ALL
1109 .iter()
1110 .map(|language| translate_check_result(*language, &result))
1111 .collect::<Vec<_>>();
1112
1113 assert!(messages.iter().all(|message| !message.is_empty()));
1114 assert!(messages.iter().any(|message| message.contains("required")));
1115 assert!(messages.iter().any(|message| message.contains("å¿…å¡«")));
1116 assert!(
1117 messages
1118 .iter()
1119 .any(|message| message.contains("obligatoire"))
1120 );
1121 assert_eq!(Language::from_code("zh-CN"), Some(Language::Chinese));
1122 assert_eq!(
1123 Language::from_code("zh-TW"),
1124 Some(Language::TraditionalChinese)
1125 );
1126 }
1127
1128 #[tokio::test]
1129 async fn user_context_language_switch_translates_checker_errors() {
1130 let mut ctx = UserContext::new()
1131 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1132 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1133 .with_checker_registry(InMemoryCheckerRegistry::new().with_checker(OrderChecker))
1134 .with_internal_id_generator(FixedIdGenerator(77))
1135 .with_language(Language::Chinese);
1136 ctx.insert_resource(PostgresDialect);
1137 ctx.insert_resource(StubExecutor {
1138 affected: 1,
1139 rows: Vec::new(),
1140 });
1141
1142 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
1143 let error = repo
1144 .prepare_insert_command(&repo.insert_command())
1145 .unwrap_err();
1146 match error {
1147 RuntimeError::Check(results) => {
1148 assert_eq!(results.len(), 1);
1149 assert!(
1150 results[0]
1151 .message
1152 .as_ref()
1153 .is_some_and(|message| message.contains("å¿…å¡«"))
1154 );
1155 }
1156 other => panic!("unexpected checker error: {other:?}"),
1157 }
1158
1159 let mut ctx = UserContext::new().with_language(Language::English);
1160 ctx.set_language_code("es").unwrap();
1161 assert_eq!(ctx.language(), Language::Spanish);
1162 }
1163
1164 #[tokio::test]
1165 async fn user_context_event_sink_receives_data_service_mutation_events() {
1166 let events = Arc::new(Mutex::new(Vec::new()));
1167 let mut ctx = UserContext::new()
1168 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1169 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1170 .with_internal_id_generator(FixedIdGenerator(88))
1171 .with_event_sink(RecordingEventSink {
1172 events: events.clone(),
1173 });
1174 ctx.insert_resource(PostgresDialect);
1175 ctx.insert_resource(StubExecutor {
1176 affected: 1,
1177 rows: vec![Record::from([
1178 ("id".to_owned(), Value::U64(88)),
1179 ("version".to_owned(), Value::I64(1)),
1180 ("name".to_owned(), Value::Text("old".to_owned())),
1181 ])],
1182 });
1183
1184 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
1185 repo.insert(&repo.insert_command().value("name", "created"))
1186 .await
1187 .unwrap();
1188 repo.update(
1189 &repo
1190 .update_command(88_u64)
1191 .expected_version(1)
1192 .value("name", "updated"),
1193 )
1194 .await
1195 .unwrap();
1196 repo.delete(&repo.delete_command(88_u64).expected_version(2))
1197 .await
1198 .unwrap();
1199 repo.recover(&repo.recover_command(88_u64, -3))
1200 .await
1201 .unwrap();
1202
1203 let events = events.lock().unwrap();
1204 assert_eq!(events.len(), 4);
1205 assert_eq!(events[0].kind, RawAuditEventKind::Created);
1206 assert_eq!(events[0].entity, "Order");
1207 assert_eq!(events[0].values.get("id"), Some(&Value::U64(88)));
1208 assert_eq!(events[1].kind, RawAuditEventKind::Updated);
1209 assert_eq!(events[1].values.get("id"), Some(&Value::U64(88)));
1210 assert_eq!(events[1].values.get("version"), Some(&Value::I64(2)));
1211 assert_eq!(events[1].updated_fields, vec!["name".to_owned()]);
1212 assert_eq!(
1213 events[1]
1214 .old_values
1215 .as_ref()
1216 .and_then(|values| values.get("name")),
1217 None );
1219 assert_eq!(
1220 events[1]
1221 .new_values
1222 .as_ref()
1223 .and_then(|values| values.get("name")),
1224 Some(&Value::Text("updated".to_owned()))
1225 );
1226 assert_eq!(events[1].changes.len(), 1);
1227 assert_eq!(events[1].changes[0].field, "name");
1228 assert_eq!(
1229 events[1].changes[0].old_value,
1230 None );
1232 assert_eq!(
1233 events[1].changes[0].new_value,
1234 Some(Value::Text("updated".to_owned()))
1235 );
1236 assert_eq!(events[2].kind, RawAuditEventKind::Deleted);
1237 assert!(events[2].old_values.is_none()); assert!(events[2].new_values.is_none());
1239 assert_eq!(events[3].kind, RawAuditEventKind::Recovered);
1240 assert_eq!(
1241 events[3]
1242 .old_values
1243 .as_ref()
1244 .and_then(|values| values.get("version")),
1245 None );
1247 assert_eq!(
1248 events[3]
1249 .new_values
1250 .as_ref()
1251 .and_then(|values| values.get("version")),
1252 Some(&Value::I64(4))
1253 );
1254 assert_eq!(events[3].changes[0].field, "version");
1255 }
1256
1257 #[tokio::test]
1258 async fn entity_data_service_builds_relation_plans() {
1259 let mut ctx = UserContext::new()
1260 .with_metadata(
1261 InMemoryMetadataStore::new()
1262 .with_entity(entity())
1263 .with_entity(line_entity())
1264 .with_entity(product_entity()),
1265 )
1266 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1267 .with_entity_data_service_behavior_registry(
1268 InMemoryEntityDataServiceBehaviorRegistry::new()
1269 .with_behavior("Order", OrderBehavior),
1270 );
1271 ctx.insert_resource(PostgresDialect);
1272 ctx.insert_resource(StubExecutor {
1273 affected: 1,
1274 rows: Vec::new(),
1275 });
1276
1277 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
1278 let plans = repo.relation_plans().unwrap();
1279
1280 assert_eq!(plans.len(), 1);
1281 assert_eq!(plans[0].relation_name, "lines");
1282 assert_eq!(plans[0].target_entity, "OrderLine");
1283 assert_eq!(plans[0].local_key, "id");
1284 assert_eq!(plans[0].foreign_key, "order_id");
1285 assert!(plans[0].many);
1286 }
1287
1288 #[tokio::test]
1289 async fn entity_data_service_builds_relation_query_from_parent_rows() {
1290 let mut ctx = UserContext::new()
1291 .with_metadata(
1292 InMemoryMetadataStore::new()
1293 .with_entity(entity())
1294 .with_entity(line_entity())
1295 .with_entity(product_entity()),
1296 )
1297 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1298 .with_entity_data_service_behavior_registry(
1299 InMemoryEntityDataServiceBehaviorRegistry::new()
1300 .with_behavior("Order", OrderBehavior),
1301 );
1302 ctx.insert_resource(PostgresDialect);
1303 ctx.insert_resource(StubExecutor {
1304 affected: 1,
1305 rows: Vec::new(),
1306 });
1307
1308 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
1309 let parent_rows = vec![
1310 Record::from([(String::from("id"), Value::U64(11))]),
1311 Record::from([(String::from("id"), Value::U64(12))]),
1312 ];
1313
1314 let query = repo.relation_query("lines", &parent_rows).unwrap();
1315 }
1320
1321 #[tokio::test]
1322 async fn entity_data_service_enhances_parent_rows_with_relations() {
1323 let mut ctx = UserContext::new()
1324 .with_metadata(
1325 InMemoryMetadataStore::new()
1326 .with_entity(entity())
1327 .with_entity(line_entity())
1328 .with_entity(product_entity()),
1329 )
1330 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"))
1331 .with_entity_data_service_behavior_registry(
1332 InMemoryEntityDataServiceBehaviorRegistry::new()
1333 .with_behavior("Order", OrderBehavior),
1334 );
1335 ctx.insert_resource(PostgresDialect);
1336 ctx.insert_resource(StubExecutor {
1337 affected: 1,
1338 rows: vec![
1339 Record::from([
1340 (String::from("id"), Value::U64(101)),
1341 (String::from("order_id"), Value::U64(11)),
1342 (String::from("name"), Value::Text(String::from("l1"))),
1343 ]),
1344 Record::from([
1345 (String::from("id"), Value::U64(102)),
1346 (String::from("order_id"), Value::U64(11)),
1347 (String::from("name"), Value::Text(String::from("l2"))),
1348 ]),
1349 Record::from([
1350 (String::from("id"), Value::U64(201)),
1351 (String::from("order_id"), Value::U64(12)),
1352 (String::from("name"), Value::Text(String::from("l3"))),
1353 ]),
1354 ],
1355 });
1356
1357 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
1358 let mut parents = vec![
1359 Record::from([(String::from("id"), Value::U64(11))]),
1360 Record::from([(String::from("id"), Value::U64(12))]),
1361 ];
1362
1363 repo.enhance_relations(&mut parents).await.unwrap();
1364
1365 match parents[0].get("lines") {
1366 Some(Value::List(lines)) => assert_eq!(lines.len(), 2),
1367 other => panic!("unexpected lines payload: {other:?}"),
1368 }
1369 match parents[1].get("lines") {
1370 Some(Value::List(lines)) => assert_eq!(lines.len(), 1),
1371 other => panic!("unexpected lines payload: {other:?}"),
1372 }
1373 }
1374
1375 #[tokio::test]
1376 async fn relation_enhancement_wraps_inverse_many_relation_as_list() {
1377 let mut ctx = UserContext::new()
1378 .with_metadata(
1379 InMemoryMetadataStore::new()
1380 .with_entity(OrderLineWithProductEntityRow::entity_descriptor())
1381 .with_entity(ProductWithLinesEntityRow::entity_descriptor()),
1382 )
1383 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("OrderLine"));
1384 ctx.insert_resource(PostgresDialect);
1385 ctx.insert_resource(QueueExecutor {
1386 affected: 1,
1387 rows: Mutex::new(VecDeque::from([
1388 vec![Record::from([
1389 (String::from("id"), Value::U64(11)),
1390 (String::from("order_id"), Value::U64(7)),
1391 (String::from("name"), Value::Text(String::from("line"))),
1392 (String::from("product_id"), Value::U64(101)),
1393 ])],
1394 vec![Record::from([
1395 (String::from("id"), Value::U64(101)),
1396 (String::from("name"), Value::Text(String::from("sku"))),
1397 ])],
1398 ])),
1399 queries: Mutex::new(Vec::new()),
1400 });
1401
1402 let repo = ctx
1403 .entity_data_service::<QueueExecutor>("OrderLine")
1404 .unwrap();
1405 let rows = repo
1406 .fetch_enhanced_entities::<OrderLineWithProductEntityRow>(
1407 &SelectQuery::new("OrderLine").relation("product"),
1408 )
1409 .await
1410 .unwrap();
1411
1412 let product = rows.data[0].product.as_ref().unwrap();
1413 assert_eq!(product.lines.data.len(), 1);
1414 assert_eq!(product.lines.data[0].id, 11);
1415 }
1416
1417 #[tokio::test]
1418 async fn entity_data_service_fetches_smart_list_of_entities() {
1419 let mut ctx = UserContext::new()
1420 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1421 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1422 ctx.insert_resource(PostgresDialect);
1423 ctx.insert_resource(StubExecutor {
1424 affected: 1,
1425 rows: vec![Record::from([
1426 (String::from("id"), Value::U64(7)),
1427 (String::from("version"), Value::I64(2)),
1428 (String::from("name"), Value::Text(String::from("typed"))),
1429 ])],
1430 });
1431
1432 let repo = ctx.entity_data_service::<StubExecutor>("Order").unwrap();
1433 let rows = repo
1434 .fetch_entities::<OrderEntity>(&repo.select())
1435 .await
1436 .unwrap();
1437
1438 assert_eq!(rows.len(), 1);
1439 assert_eq!(
1440 rows.first(),
1441 Some(&OrderEntity {
1442 id: 7,
1443 version: 2,
1444 name: String::from("typed"),
1445 })
1446 );
1447 }
1448
1449 #[tokio::test]
1450 async fn entity_data_service_fetches_smart_list_of_derived_entities() {
1451 let mut ctx = UserContext::new()
1452 .with_metadata(
1453 InMemoryMetadataStore::new().with_entity(CatalogProductRow::entity_descriptor()),
1454 )
1455 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("CatalogProduct"));
1456 ctx.insert_resource(PostgresDialect);
1457 ctx.insert_resource(StubExecutor {
1458 affected: 1,
1459 rows: vec![Record::from([
1460 (String::from("id"), Value::U64(9)),
1461 (String::from("name"), Value::Text(String::from("derived"))),
1462 ])],
1463 });
1464
1465 let repo = ctx
1466 .entity_data_service::<StubExecutor>("CatalogProduct")
1467 .unwrap();
1468 let rows = repo
1469 .fetch_entities::<CatalogProductRow>(&repo.select())
1470 .await
1471 .unwrap();
1472
1473 assert_eq!(rows.len(), 1);
1474 assert_eq!(
1475 rows.first(),
1476 Some(&CatalogProductRow {
1477 id: 9,
1478 name: String::from("derived"),
1479 })
1480 );
1481 }
1482
1483 #[tokio::test]
1484 async fn entity_data_service_collects_dynamic_properties_for_aggregate_output() {
1485 let mut ctx = UserContext::new()
1486 .with_metadata(
1487 InMemoryMetadataStore::new()
1488 .with_entity(OrderAggregateDynamic::entity_descriptor()),
1489 )
1490 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("OrderAggregate"));
1491 ctx.insert_resource(PostgresDialect);
1492 ctx.insert_resource(StubExecutor {
1493 affected: 1,
1494 rows: vec![Record::from([
1495 (String::from("id"), Value::U64(1)),
1496 (String::from("lineCount"), Value::I64(3)),
1497 (String::from("amount"), Value::F64(18.5)),
1498 ])],
1499 });
1500
1501 let repo = ctx
1502 .entity_data_service::<StubExecutor>("OrderAggregate")
1503 .unwrap();
1504 let rows = repo
1505 .fetch_entities::<OrderAggregateDynamic>(&repo.select())
1506 .await
1507 .unwrap();
1508
1509 assert_eq!(rows.len(), 1);
1510 assert_eq!(rows.data[0].id, 1);
1511 assert_eq!(rows.data[0].dynamic.get("lineCount"), Some(&Value::I64(3)));
1512 assert_eq!(rows.data[0].dynamic.get("amount"), Some(&Value::F64(18.5)));
1513 assert_eq!(
1514 rows.into_vec().into_iter().next().unwrap().into_json(),
1515 serde_json::json!({
1516 "id": 1,
1517 "lineCount": 3,
1518 "amount": 18.5
1519 })
1520 );
1521 }
1522
1523 #[tokio::test]
1524 async fn entity_data_service_executes_relation_aggregates_into_dynamic_properties() {
1525 let executor = QueueExecutor {
1526 affected: 1,
1527 rows: Mutex::new(VecDeque::from([
1528 vec![
1529 Record::from([
1530 (String::from("id"), Value::U64(1)),
1531 (String::from("version"), Value::I64(1)),
1532 (String::from("name"), Value::Text(String::from("first"))),
1533 ]),
1534 Record::from([
1535 (String::from("id"), Value::U64(2)),
1536 (String::from("version"), Value::I64(1)),
1537 (String::from("name"), Value::Text(String::from("second"))),
1538 ]),
1539 ],
1540 vec![Record::from([
1541 (String::from("order_id"), Value::U64(1)),
1542 (String::from("lineCount"), Value::I64(3)),
1543 ])],
1544 ])),
1545 queries: Mutex::new(Vec::new()),
1546 };
1547 let mut ctx = UserContext::new()
1548 .with_metadata(
1549 InMemoryMetadataStore::new()
1550 .with_entity(entity())
1551 .with_entity(line_entity()),
1552 )
1553 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1554 ctx.insert_resource(PostgresDialect);
1555 ctx.insert_resource(executor);
1556
1557 let repo = ctx.entity_data_service::<QueueExecutor>("Order").unwrap();
1558 let rows = repo
1559 .fetch_all_with_relation_aggregates(
1560 &repo
1561 .select()
1562 .project("id")
1563 .project("version")
1564 .project("name"),
1565 &[RelationAggregate::new(
1566 "lines",
1567 "lineCount",
1568 SelectQuery::new("OrderLine"),
1569 true,
1570 )],
1571 )
1572 .await
1573 .unwrap();
1574
1575 assert_eq!(rows[0].get("lineCount"), Some(&Value::I64(3)));
1576 assert_eq!(rows[1].get("lineCount"), Some(&Value::U64(0)));
1577
1578 let executor = ctx.get_resource::<QueueExecutor>().unwrap();
1579 let queries = executor.queries.lock().unwrap();
1580 assert_eq!(queries.len(), 2);
1581 assert_eq!(queries[1], "SELECT ... FROM OrderLine ...");
1582 }
1583
1584 #[tokio::test]
1585 async fn entity_data_service_maps_relation_aggregate_storage_key_to_property_key() {
1586 let mut line = line_entity();
1587 line.properties
1588 .iter_mut()
1589 .find(|property| property.name == "order_id")
1590 .unwrap()
1591 .column_name = "order_ref".to_owned();
1592 let executor = QueueExecutor {
1593 affected: 1,
1594 rows: Mutex::new(VecDeque::from([
1595 vec![Record::from([
1596 (String::from("id"), Value::U64(1)),
1597 (String::from("version"), Value::I64(1)),
1598 (String::from("name"), Value::Text(String::from("first"))),
1599 ])],
1600 vec![Record::from([
1601 (String::from("order_ref"), Value::I64(1)),
1602 (String::from("lineCount"), Value::I64(3)),
1603 ])],
1604 ])),
1605 queries: Mutex::new(Vec::new()),
1606 };
1607 let mut ctx = UserContext::new()
1608 .with_metadata(
1609 InMemoryMetadataStore::new()
1610 .with_entity(entity())
1611 .with_entity(line),
1612 )
1613 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1614 ctx.insert_resource(PostgresDialect);
1615 ctx.insert_resource(executor);
1616
1617 let repo = ctx.entity_data_service::<QueueExecutor>("Order").unwrap();
1618 let rows = repo
1619 .fetch_all_with_relation_aggregates(
1620 &repo
1621 .select()
1622 .project("id")
1623 .project("version")
1624 .project("name"),
1625 &[RelationAggregate::new(
1626 "lines",
1627 "lineCount",
1628 SelectQuery::new("OrderLine"),
1629 true,
1630 )],
1631 )
1632 .await
1633 .unwrap();
1634
1635 assert_eq!(rows[0].get("lineCount"), Some(&Value::I64(3)));
1636 let executor = ctx.get_resource::<QueueExecutor>().unwrap();
1637 assert_eq!(
1638 executor.queries.lock().unwrap()[1],
1639 "SELECT ... FROM OrderLine ..."
1640 );
1641 }
1642
1643 #[tokio::test]
1644 async fn entity_data_service_uses_aggregation_cache_when_resource_is_registered() {
1645 let executor = QueueExecutor {
1646 affected: 1,
1647 rows: Mutex::new(VecDeque::from([vec![Record::from([(
1648 String::from("count"),
1649 Value::I64(2),
1650 )])]])),
1651 queries: Mutex::new(Vec::new()),
1652 };
1653 let mut ctx = UserContext::new()
1654 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1655 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1656 ctx.insert_resource(PostgresDialect);
1657 ctx.insert_resource(executor);
1658 ctx.insert_resource(InMemoryAggregationCache::default());
1659
1660 let repo = ctx.entity_data_service::<QueueExecutor>("Order").unwrap();
1661 let query = repo
1662 .select()
1663 .count("count")
1664 .enable_aggregation_cache_for(60_000);
1665
1666 let first = repo.fetch_all(&query).await.unwrap();
1667 let second = repo.fetch_all(&query).await.unwrap();
1668
1669 assert_eq!(first, second);
1670 let executor = ctx.get_resource::<QueueExecutor>().unwrap();
1671 assert_eq!(executor.queries.lock().unwrap().len(), 1);
1672 }
1673
1674 #[tokio::test]
1675 async fn aggregation_cache_is_namespaced_and_invalidated_after_write() {
1676 let executor = QueueExecutor {
1677 affected: 1,
1678 rows: Mutex::new(VecDeque::from([
1679 vec![Record::from([(String::from("count"), Value::I64(2))])],
1680 vec![Record::from([(String::from("count"), Value::I64(3))])],
1681 ])),
1682 queries: Mutex::new(Vec::new()),
1683 };
1684 let mut ctx = UserContext::new()
1685 .with_metadata(InMemoryMetadataStore::new().with_entity(entity()))
1686 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1687 ctx.insert_resource(PostgresDialect);
1688 ctx.insert_resource(executor);
1689 ctx.insert_resource(
1690 Arc::new(InMemoryAggregationCache::with_namespace("tenant-a"))
1691 as Arc<dyn AggregationCacheBackend>,
1692 );
1693
1694 let repo = ctx.entity_data_service::<QueueExecutor>("Order").unwrap();
1695 let query = repo
1696 .select()
1697 .count("count")
1698 .enable_aggregation_cache_for(60_000);
1699
1700 let first = repo.fetch_all(&query).await.unwrap();
1701 let cached = repo.fetch_all(&query).await.unwrap();
1702 repo.insert(
1703 &InsertCommand::new("Order")
1704 .value("id", 9_u64)
1705 .value("version", 1_i64)
1706 .value("name", "new"),
1707 )
1708 .await
1709 .unwrap();
1710 let refreshed = repo.fetch_all(&query).await.unwrap();
1711
1712 assert_eq!(first, cached);
1713 assert_ne!(cached, refreshed);
1714 let executor = ctx.get_resource::<QueueExecutor>().unwrap();
1715 assert_eq!(executor.queries.lock().unwrap().len(), 2);
1716 }
1717
1718 #[tokio::test]
1719 async fn aggregation_cache_propagates_to_relation_aggregates() {
1720 let parent_rows = vec![
1721 Record::from([
1722 (String::from("id"), Value::U64(1)),
1723 (String::from("version"), Value::I64(1)),
1724 (String::from("name"), Value::Text(String::from("first"))),
1725 ]),
1726 Record::from([
1727 (String::from("id"), Value::U64(2)),
1728 (String::from("version"), Value::I64(1)),
1729 (String::from("name"), Value::Text(String::from("second"))),
1730 ]),
1731 ];
1732 let aggregate_rows = vec![Record::from([
1733 (String::from("order_id"), Value::U64(1)),
1734 (String::from("lineCount"), Value::I64(3)),
1735 ])];
1736 let executor = QueueExecutor {
1737 affected: 1,
1738 rows: Mutex::new(VecDeque::from([parent_rows, aggregate_rows])),
1739 queries: Mutex::new(Vec::new()),
1740 };
1741 let mut ctx = UserContext::new()
1742 .with_metadata(
1743 InMemoryMetadataStore::new()
1744 .with_entity(entity())
1745 .with_entity(line_entity()),
1746 )
1747 .with_entity_registry(InMemoryEntityRegistry::new().with_entity("Order"));
1748 ctx.insert_resource(PostgresDialect);
1749 ctx.insert_resource(executor);
1750 ctx.insert_resource(InMemoryAggregationCache::default());
1751
1752 let repo = ctx.entity_data_service::<QueueExecutor>("Order").unwrap();
1753 let query = repo
1754 .select()
1755 .project("id")
1756 .project("version")
1757 .project("name")
1758 .enable_aggregation_cache_for(60_000)
1759 .propagate_aggregation_cache(60_000);
1760 let aggregate =
1761 RelationAggregate::new("lines", "lineCount", SelectQuery::new("OrderLine"), true);
1762
1763 let first = repo
1764 .fetch_all_with_relation_aggregates(&query, &[aggregate.clone()])
1765 .await
1766 .unwrap();
1767 let second = repo
1768 .fetch_all_with_relation_aggregates(&query, &[aggregate])
1769 .await
1770 .unwrap();
1771
1772 assert_eq!(first, second);
1773 let executor = ctx.get_resource::<QueueExecutor>().unwrap();
1774 assert_eq!(executor.queries.lock().unwrap().len(), 2);
1775 }
1776
1777 #[tokio::test]
1778 async fn memory_data_service_fetches_smart_list_entities_with_query_features() {
1779 let metadata = InMemoryMetadataStore::new().with_entity(entity());
1780 let data_service = MemoryDataService::new(metadata).with_rows(
1781 "Order",
1782 vec![
1783 Record::from([
1784 (String::from("id"), Value::U64(1)),
1785 (String::from("version"), Value::I64(1)),
1786 (String::from("name"), Value::Text(String::from("alpha"))),
1787 ]),
1788 Record::from([
1789 (String::from("id"), Value::U64(2)),
1790 (String::from("version"), Value::I64(1)),
1791 (String::from("name"), Value::Text(String::from("beta"))),
1792 ]),
1793 Record::from([
1794 (String::from("id"), Value::U64(3)),
1795 (String::from("version"), Value::I64(1)),
1796 (String::from("name"), Value::Text(String::from("gamma"))),
1797 ]),
1798 ],
1799 );
1800
1801 let query = teaql_core::SelectQuery::new("Order")
1802 .filter(Expr::Binary {
1803 left: Box::new(Expr::column("id")),
1804 op: teaql_core::BinaryOp::Gte,
1805 right: Box::new(Expr::value(2_u64)),
1806 })
1807 .order_by(OrderBy::desc("id"))
1808 .limit(1);
1809
1810 let orders = data_service.fetch_entities::<Order>(&query).unwrap();
1811
1812 assert_eq!(orders.ids(), vec![Value::U64(3)]);
1813 assert_eq!(orders.versions(), vec![1]);
1814 assert_eq!(orders.first().unwrap().name, "gamma");
1815 }
1816
1817 #[tokio::test]
1818 async fn memory_data_service_runs_relation_aggregates() {
1819 let metadata = InMemoryMetadataStore::new()
1820 .with_entity(entity())
1821 .with_entity(line_entity());
1822
1823 let data_service = MemoryDataService::new(metadata)
1824 .with_rows(
1825 "Order",
1826 vec![
1827 Record::from([
1828 (String::from("id"), Value::U64(1)),
1829 (String::from("version"), Value::I64(1)),
1830 (String::from("name"), Value::Text(String::from("first"))),
1831 ]),
1832 Record::from([
1833 (String::from("id"), Value::U64(2)),
1834 (String::from("version"), Value::I64(1)),
1835 (String::from("name"), Value::Text(String::from("second"))),
1836 ]),
1837 ],
1838 )
1839 .with_rows(
1840 "OrderLine",
1841 vec![
1842 Record::from([
1843 (String::from("id"), Value::U64(10)),
1844 (String::from("version"), Value::I64(1)),
1845 (String::from("order_id"), Value::U64(1)),
1846 (String::from("name"), Value::Text(String::from("line1"))),
1847 ]),
1848 Record::from([
1849 (String::from("id"), Value::U64(11)),
1850 (String::from("version"), Value::I64(1)),
1851 (String::from("order_id"), Value::U64(1)),
1852 (String::from("name"), Value::Text(String::from("line2"))),
1853 ]),
1854 Record::from([
1855 (String::from("id"), Value::U64(12)),
1856 (String::from("version"), Value::I64(1)),
1857 (String::from("order_id"), Value::U64(2)),
1858 (String::from("name"), Value::Text(String::from("line3"))),
1859 ]),
1860 ],
1861 );
1862
1863 let query = SelectQuery::new("Order").project("id").project("name");
1864 let aggregate =
1865 RelationAggregate::new("lines", "lineCount", SelectQuery::new("OrderLine"), true);
1866
1867 let rows = data_service
1868 .fetch_all_with_relation_aggregates(&query, &[aggregate])
1869 .unwrap();
1870
1871 assert_eq!(rows.len(), 2);
1872
1873 let first_order = rows
1874 .iter()
1875 .find(|r| r.get("id") == Some(&Value::U64(1)))
1876 .unwrap();
1877 assert_eq!(first_order.get("lineCount"), Some(&Value::U64(2)));
1878
1879 let second_order = rows
1880 .iter()
1881 .find(|r| r.get("id") == Some(&Value::U64(2)))
1882 .unwrap();
1883 assert_eq!(second_order.get("lineCount"), Some(&Value::U64(1)));
1884 }
1885
1886 #[tokio::test]
1887 async fn memory_data_service_runs_aggregates() {
1888 let metadata = InMemoryMetadataStore::new().with_entity(entity());
1889 let data_service = MemoryDataService::new(metadata).with_rows(
1890 "Order",
1891 vec![
1892 Record::from([
1893 (String::from("id"), Value::U64(1)),
1894 (String::from("version"), Value::I64(1)),
1895 (String::from("name"), Value::Text(String::from("alpha"))),
1896 ]),
1897 Record::from([
1898 (String::from("id"), Value::U64(2)),
1899 (String::from("version"), Value::I64(2)),
1900 (String::from("name"), Value::Text(String::from("beta"))),
1901 ]),
1902 ],
1903 );
1904
1905 let query = teaql_core::SelectQuery {
1906 entity: String::from("Order"),
1907 projection: Vec::new(),
1908 expr_projection: Vec::new(),
1909 filter: None,
1910 having: None,
1911 order_by: Vec::new(),
1912 slice: None,
1913 trace_chain: Vec::new(),
1914 aggregates: vec![
1915 Aggregate {
1916 function: AggregateFunction::Count,
1917 field: String::from("id"),
1918 alias: String::from("count"),
1919 },
1920 Aggregate {
1921 function: AggregateFunction::Sum,
1922 field: String::from("version"),
1923 alias: String::from("versionSum"),
1924 },
1925 ],
1926 group_by: Vec::new(),
1927 relations: Vec::new(),
1928 aggregation_cache: None,
1929 comment: None,
1930 raw_sql: None,
1931 raw_sql_search_criteria: Vec::new(),
1932 dynamic_properties: Vec::new(),
1933 raw_projections: Vec::new(),
1934 object_group_bys: Vec::new(),
1935 search_with_text: None,
1936 child_enhancements: Vec::new(),
1937 stream_config: None,
1938 };
1939
1940 let rows = data_service.fetch_all(&query).unwrap();
1941
1942 assert_eq!(rows.len(), 1);
1943 assert_eq!(rows[0].get("count"), Some(&Value::U64(2)));
1944 assert_eq!(rows[0].get("versionSum"), Some(&Value::U64(3)));
1945 }
1946
1947 #[tokio::test]
1948 async fn memory_data_service_runs_grouped_aggregates_and_extended_filters() {
1949 let metadata = InMemoryMetadataStore::new().with_entity(entity());
1950 let data_service = MemoryDataService::new(metadata).with_rows(
1951 "Order",
1952 vec![
1953 Record::from([
1954 (String::from("id"), Value::U64(1)),
1955 (String::from("version"), Value::I64(1)),
1956 (String::from("name"), Value::Text(String::from("alpha"))),
1957 ]),
1958 Record::from([
1959 (String::from("id"), Value::U64(2)),
1960 (String::from("version"), Value::I64(2)),
1961 (String::from("name"), Value::Text(String::from("alpha"))),
1962 ]),
1963 Record::from([
1964 (String::from("id"), Value::U64(3)),
1965 (String::from("version"), Value::I64(3)),
1966 (String::from("name"), Value::Text(String::from("tmp-beta"))),
1967 ]),
1968 ],
1969 );
1970
1971 let rows = data_service
1972 .fetch_all(
1973 &teaql_core::SelectQuery::new("Order")
1974 .filter(
1975 Expr::between("version", 1_i64, 3_i64)
1976 .and_expr(Expr::not_like("name", "tmp%"))
1977 .and_expr(Expr::not_in_list("name", vec![Value::from("deleted")])),
1978 )
1979 .group_by("name")
1980 .count("total")
1981 .sum("version", "versionSum"),
1982 )
1983 .unwrap();
1984
1985 assert_eq!(rows.len(), 1);
1986 assert_eq!(
1987 rows[0].get("name"),
1988 Some(&Value::Text(String::from("alpha")))
1989 );
1990 assert_eq!(rows[0].get("total"), Some(&Value::U64(2)));
1991 assert_eq!(rows[0].get("versionSum"), Some(&Value::U64(3)));
1992 }
1993
1994 #[tokio::test]
1995 async fn memory_data_service_runs_extended_aggregates_and_having() {
1996 let metadata = InMemoryMetadataStore::new().with_entity(entity());
1997 let data_service = MemoryDataService::new(metadata).with_rows(
1998 "Order",
1999 vec![
2000 Record::from([
2001 (String::from("id"), Value::U64(1)),
2002 (String::from("version"), Value::I64(1)),
2003 (String::from("name"), Value::Text(String::from("alpha"))),
2004 ]),
2005 Record::from([
2006 (String::from("id"), Value::U64(2)),
2007 (String::from("version"), Value::I64(3)),
2008 (String::from("name"), Value::Text(String::from("alpha"))),
2009 ]),
2010 Record::from([
2011 (String::from("id"), Value::U64(3)),
2012 (String::from("version"), Value::I64(7)),
2013 (String::from("name"), Value::Text(String::from("beta"))),
2014 ]),
2015 ],
2016 );
2017
2018 let rows = data_service
2019 .fetch_all(
2020 &teaql_core::SelectQuery::new("Order")
2021 .group_by("name")
2022 .count("total")
2023 .stddev("version", "stddevVersion")
2024 .var_pop("version", "varPopVersion")
2025 .bit_or("version", "bitOrVersion")
2026 .having(Expr::gt("total", 1_i64)),
2027 )
2028 .unwrap();
2029
2030 assert_eq!(rows.len(), 1);
2031 assert_eq!(
2032 rows[0].get("name"),
2033 Some(&Value::Text(String::from("alpha")))
2034 );
2035 assert_eq!(rows[0].get("total"), Some(&Value::U64(2)));
2036 assert_eq!(
2037 rows[0].get("stddevVersion").map(Value::to_json_value),
2038 Some(serde_json::Value::String(
2039 "1.4142135623730951454746218583".to_owned()
2040 ))
2041 );
2042 assert_eq!(
2043 rows[0].get("varPopVersion"),
2044 Some(&Value::Decimal(Decimal::ONE))
2045 );
2046 assert_eq!(rows[0].get("bitOrVersion"), Some(&Value::I64(3)));
2047 }
2048
2049 #[tokio::test]
2050 async fn memory_data_service_runs_sound_like_filter() {
2051 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2052 let data_service = MemoryDataService::new(metadata).with_rows(
2053 "Order",
2054 vec![
2055 Record::from([
2056 (String::from("id"), Value::U64(1)),
2057 (String::from("version"), Value::I64(1)),
2058 (String::from("name"), Value::Text(String::from("Robert"))),
2059 ]),
2060 Record::from([
2061 (String::from("id"), Value::U64(2)),
2062 (String::from("version"), Value::I64(1)),
2063 (String::from("name"), Value::Text(String::from("Rupert"))),
2064 ]),
2065 Record::from([
2066 (String::from("id"), Value::U64(3)),
2067 (String::from("version"), Value::I64(1)),
2068 (String::from("name"), Value::Text(String::from("Ashcraft"))),
2069 ]),
2070 ],
2071 );
2072
2073 let rows = data_service
2074 .fetch_all(
2075 &teaql_core::SelectQuery::new("Order")
2076 .filter(Expr::sound_like("name", "Robert"))
2077 .order_asc("id"),
2078 )
2079 .unwrap();
2080
2081 assert_eq!(rows.len(), 2);
2082 assert_eq!(rows[0].get("name"), Some(&Value::Text("Robert".to_owned())));
2083 assert_eq!(rows[1].get("name"), Some(&Value::Text("Rupert".to_owned())));
2084 }
2085
2086 #[tokio::test]
2087 async fn memory_data_service_runs_java_style_string_match_filters() {
2088 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2089 let data_service = MemoryDataService::new(metadata).with_rows(
2090 "Order",
2091 vec![
2092 Record::from([
2093 (String::from("id"), Value::U64(1)),
2094 (String::from("version"), Value::I64(1)),
2095 (String::from("name"), Value::Text(String::from("tea-order"))),
2096 ]),
2097 Record::from([
2098 (String::from("id"), Value::U64(2)),
2099 (String::from("version"), Value::I64(1)),
2100 (
2101 String::from("name"),
2102 Value::Text(String::from("coffee-order")),
2103 ),
2104 ]),
2105 Record::from([
2106 (String::from("id"), Value::U64(3)),
2107 (String::from("version"), Value::I64(1)),
2108 (
2109 String::from("name"),
2110 Value::Text(String::from("tea-archived")),
2111 ),
2112 ]),
2113 ],
2114 );
2115
2116 let rows = data_service
2117 .fetch_all(
2118 &teaql_core::SelectQuery::new("Order")
2119 .filter(
2120 Expr::contain("name", "tea")
2121 .and_expr(Expr::begin_with("name", "tea"))
2122 .and_expr(Expr::end_with("name", "order"))
2123 .and_expr(Expr::not_contain("name", "coffee"))
2124 .and_expr(Expr::not_begin_with("name", "archived"))
2125 .and_expr(Expr::not_end_with("name", "draft")),
2126 )
2127 .order_asc("id"),
2128 )
2129 .unwrap();
2130
2131 assert_eq!(rows.len(), 1);
2132 assert_eq!(
2133 rows[0].get("name"),
2134 Some(&Value::Text("tea-order".to_owned()))
2135 );
2136 }
2137
2138 #[tokio::test]
2139 async fn memory_data_service_runs_property_to_property_filters() {
2140 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2141 let data_service = MemoryDataService::new(metadata).with_rows(
2142 "Order",
2143 vec![
2144 Record::from([
2145 (String::from("id"), Value::U64(1)),
2146 (String::from("version"), Value::I64(2)),
2147 (String::from("name"), Value::Text(String::from("keep"))),
2148 ]),
2149 Record::from([
2150 (String::from("id"), Value::U64(2)),
2151 (String::from("version"), Value::I64(1)),
2152 (String::from("name"), Value::Text(String::from("skip"))),
2153 ]),
2154 ],
2155 );
2156
2157 let rows = data_service
2158 .fetch_all(
2159 &teaql_core::SelectQuery::new("Order")
2160 .filter(Expr::compare_columns("version", BinaryOp::Gte, "id"))
2161 .order_asc("id"),
2162 )
2163 .unwrap();
2164
2165 assert_eq!(rows.len(), 1);
2166 assert_eq!(rows[0].get("name"), Some(&Value::Text("keep".to_owned())));
2167 }
2168
2169 #[tokio::test]
2170 async fn memory_data_service_supports_mutations_and_optimistic_locking() {
2171 let metadata = InMemoryMetadataStore::new().with_entity(entity());
2172 let data_service = MemoryDataService::new(metadata);
2173
2174 data_service
2175 .insert(
2176 &InsertCommand::new("Order")
2177 .value("id", 10_u64)
2178 .value("version", 1_i64)
2179 .value("name", "draft"),
2180 )
2181 .unwrap();
2182 data_service
2183 .update(
2184 &UpdateCommand::new("Order", 10_u64)
2185 .expected_version(1)
2186 .value("name", "submitted"),
2187 )
2188 .unwrap();
2189
2190 let row = data_service
2191 .fetch_all(&teaql_core::SelectQuery::new("Order").filter(Expr::eq("id", 10_u64)))
2192 .unwrap()
2193 .pop()
2194 .unwrap();
2195 assert_eq!(
2196 row.get("name"),
2197 Some(&Value::Text(String::from("submitted")))
2198 );
2199 assert_eq!(row.get("version"), Some(&Value::I64(2)));
2200
2201 let conflict = data_service
2202 .update(
2203 &UpdateCommand::new("Order", 10_u64)
2204 .expected_version(1)
2205 .value("name", "stale"),
2206 )
2207 .unwrap_err();
2208 assert!(matches!(
2209 conflict,
2210 DataServiceError::Runtime(RuntimeError::OptimisticLockConflict { .. })
2211 ));
2212
2213 data_service
2214 .delete(&DeleteCommand::new("Order", 10_u64).expected_version(2))
2215 .unwrap();
2216 let row = data_service
2217 .fetch_all(&teaql_core::SelectQuery::new("Order").filter(Expr::eq("id", 10_u64)))
2218 .unwrap()
2219 .pop()
2220 .unwrap();
2221 assert_eq!(row.get("version"), Some(&Value::I64(-3)));
2222
2223 data_service
2224 .recover(&RecoverCommand::new("Order", 10_u64, -3))
2225 .unwrap();
2226 let row = data_service
2227 .fetch_all(&teaql_core::SelectQuery::new("Order").filter(Expr::eq("id", 10_u64)))
2228 .unwrap()
2229 .pop()
2230 .unwrap();
2231 assert_eq!(row.get("version"), Some(&Value::I64(4)));
2232 }
2233
2234 #[tokio::test]
2235 async fn user_context_reports_missing_schema_provider() {
2236 let err = UserContext::new().ensure_schema().await.unwrap_err();
2237 assert!(
2238 matches!(err, RuntimeError::Schema(message) if message == "missing schema provider")
2239 );
2240 }
2241
2242 #[tokio::test]
2243 async fn user_context_stores_and_exposes_user_identifier() {
2244 let mut ctx = UserContext::new();
2245 let pid = std::process::id();
2246 let thread_id_str = format!("{:?}", std::thread::current().id());
2247 let numeric_thread_id = thread_id_str
2248 .strip_prefix("ThreadId(")
2249 .and_then(|s| s.strip_suffix(")"))
2250 .unwrap_or(&thread_id_str);
2251 let os_user = std::env::var("USER")
2252 .or_else(|_| std::env::var("USERNAME"))
2253 .unwrap_or_else(|_| "main".to_owned());
2254 let expected_default = format!("{os_user}@pid-{pid}.tid-{numeric_thread_id}");
2255 assert_eq!(ctx.user_identifier(), Some(expected_default.as_str()));
2256
2257 ctx.set_user_identifier("user-123");
2258 assert_eq!(ctx.user_identifier(), Some("user-123"));
2259
2260 let ctx2 = UserContext::new().with_user_identifier("user-456");
2261 assert_eq!(ctx2.user_identifier(), Some("user-456"));
2262
2263 let mut ctx3 = UserContext::new();
2264 ctx3.set_user_identifier_option(Some("user-789".to_owned()));
2265 assert_eq!(ctx3.user_identifier(), Some("user-789"));
2266 ctx3.set_user_identifier_option(None);
2267 assert_eq!(ctx3.user_identifier(), None);
2268
2269 let ctx4 = UserContext::new().with_user_identifier_option(Some("user-abc".to_owned()));
2270 assert_eq!(ctx4.user_identifier(), Some("user-abc"));
2271 }
2272}
2273
2274pub use checker::{
2275 CHECK_OBJECT_STATUS_FIELD, CheckObjectStatus, CheckResult, CheckResults, CheckRule, Checker,
2276 CheckerRegistry, InMemoryCheckerRegistry, LocationSegment, ObjectLocation, TypedChecker,
2277 TypedEntityChecker, clear_record_status, mark_record_status,
2278};