Skip to main content

teaql_runtime/
lib.rs

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