Skip to main content

teaql_runtime/
lib.rs

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