teaql_runtime/
lib.rs

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

teaql_runtime/
lib.rs
