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