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