Skip to main content

teaql_runtime/
lib.rs

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