Skip to main content

teaql_runtime/
lib.rs

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