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