teaql_runtime/data_service/
resolved.rs

1use std::hash::{Hash, Hasher};
2use std::sync::Arc;
3use std::time::{Duration, SystemTime};
4
5use teaql_core::{
6    AggregationCacheOptions, DeleteCommand, Entity, Expr, InsertCommand, Record, RecoverCommand,
7    RelationAggregate, SelectQuery, SmartList, SortDirection, UpdateCommand, Value,
8};
9
10use crate::{
11    CheckObjectStatus, ContinuousPageCursor, DataServiceError, EntityDataServiceBehavior,
12    PurposedSelectQuery, RawAuditEvent, RuntimeError, clear_record_status, mark_record_status,
13};
14
15use super::{
16    AggregationCacheBackend, ContextDataService, EntityDataService, InMemoryAggregationCache,
17    UserContextMetadata, helpers::*,
18};
19
20#[derive(Debug, Clone)]
21struct ContinuousPageExecution {
22    query_key: String,
23    direction: SortDirection,
24    page_size: u64,
25    original_offset: u64,
26    ttl_seconds: u64,
27    optimized: bool,
28    seek_cursor_id: Option<String>,
29}
30
31impl<'a, E> EntityDataService<'a, E>
32where
33    E: teaql_data_service::QueryExecutor
34        + teaql_data_service::MutationExecutor
35        + Send
36        + Sync
37        + 'static,
38{
39    pub(super) fn query_behavior(
40        &self,
41        entity: &str,
42    ) -> Option<Arc<dyn EntityDataServiceBehavior>> {
43        self.data_service
44            .metadata
45            .context
46            .entity_data_service_behavior(entity)
47    }
48
49    pub(super) fn behavior(&self) -> Option<Arc<dyn EntityDataServiceBehavior>> {
50        self.data_service
51            .metadata
52            .context
53            .entity_data_service_behavior(&self.entity)
54    }
55
56    pub fn entity(&self) -> &str {
57        &self.entity
58    }
59
60    pub fn select(&self) -> SelectQuery {
61        SelectQuery::new(self.entity.clone())
62    }
63
64    pub fn insert_command(&self) -> InsertCommand {
65        InsertCommand::new(self.entity.clone())
66    }
67
68    fn enforce_insert_policy(&self, command: &mut InsertCommand) -> Result<(), RuntimeError> {
69        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
70            policy.enforce_insert(self.data_service.metadata.context, command)?;
71        }
72        Ok(())
73    }
74
75    fn enforce_update_policy(&self, command: &mut UpdateCommand) -> Result<(), RuntimeError> {
76        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
77            policy.enforce_update(self.data_service.metadata.context, command)?;
78        }
79        Ok(())
80    }
81
82    fn enforce_delete_policy(&self, command: &mut DeleteCommand) -> Result<(), RuntimeError> {
83        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
84            policy.enforce_delete(self.data_service.metadata.context, command)?;
85        }
86        Ok(())
87    }
88
89    fn enforce_recover_policy(&self, command: &mut RecoverCommand) -> Result<(), RuntimeError> {
90        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
91            policy.enforce_recover(self.data_service.metadata.context, command)?;
92        }
93        Ok(())
94    }
95
96    fn prepare_select_query(&self, query: &SelectQuery) -> Result<SelectQuery, RuntimeError> {
97        let mut query = query.clone();
98
99        let mut full_trace = self.trace_context.clone();
100        full_trace.extend(query.trace_chain);
101        query.trace_chain = full_trace;
102
103        if let Some(behavior) = self.query_behavior(&query.entity) {
104            behavior.before_select(self.data_service.metadata.context, &mut query)?;
105        }
106        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
107            policy.enforce_select(self.data_service.metadata.context, &mut query)?;
108        }
109        // Ensure local_key fields for relation loads are projected so that
110        // enhance_query_relations can match parent rows to child records.
111        if !query.relations.is_empty() {
112            if let Some(descriptor) = self.data_service.metadata.context.entity(&query.entity) {
113                for load in &query.relations {
114                    if let Some(relation) = descriptor.relation_by_name(&load.name) {
115                        if !query.projection.contains(&relation.local_key) {
116                            query.projection.push(relation.local_key.clone());
117                        }
118                    }
119                }
120            }
121        }
122        Ok(query)
123    }
124
125    pub fn prepare_insert_command(
126        &self,
127        command: &InsertCommand,
128    ) -> Result<InsertCommand, RuntimeError> {
129        let mut command = command.clone();
130        if let Some(behavior) = self.behavior() {
131            behavior.before_insert(self.data_service.metadata.context, &mut command)?;
132        }
133        self.enforce_insert_policy(&mut command)?;
134
135        let entity = self
136            .data_service
137            .metadata
138            .context
139            .require_entity(&command.entity)?;
140        if let Some(id_property) = entity.id_property() {
141            let needs_id = !command.values.contains_key(&id_property.name)
142                || is_unassigned_id(command.values.get(&id_property.name));
143            if needs_id {
144                let id = self
145                    .data_service
146                    .metadata
147                    .context
148                    .next_id(&command.entity)?;
149                command
150                    .values
151                    .insert(id_property.name.clone(), Value::U64(id));
152            }
153        }
154        ensure_initial_version(&mut command.values, entity);
155        mark_record_status(&mut command.values, CheckObjectStatus::Create);
156        let check_result = self
157            .data_service
158            .metadata
159            .context
160            .check_and_fix_record(&command.entity, &mut command.values);
161        clear_record_status(&mut command.values);
162        check_result?;
163
164        Ok(command)
165    }
166
167    pub fn update_command(&self, id: impl Into<Value>) -> UpdateCommand {
168        UpdateCommand::new(self.entity.clone(), id)
169    }
170
171    pub fn prepare_update_command(
172        &self,
173        command: &UpdateCommand,
174    ) -> Result<UpdateCommand, RuntimeError> {
175        let mut command = command.clone();
176        if let Some(behavior) = self.behavior() {
177            behavior.before_update(self.data_service.metadata.context, &mut command)?;
178        }
179        self.enforce_update_policy(&mut command)?;
180
181        Ok(command)
182    }
183
184    pub fn delete_command(&self, id: impl Into<Value>) -> DeleteCommand {
185        DeleteCommand::new(self.entity.clone(), id)
186    }
187
188    pub fn recover_command(&self, id: impl Into<Value>, expected_version: i64) -> RecoverCommand {
189        RecoverCommand::new(self.entity.clone(), id, expected_version)
190    }
191
192    pub(crate) async fn fetch_all_internal(
193        &self,
194        query: &SelectQuery,
195    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
196        let query = self
197            .prepare_select_query(query)
198            .map_err(DataServiceError::Runtime)?;
199        let query = query
200            .prepare_for_list()
201            .map_err(|message| DataServiceError::Runtime(RuntimeError::Graph(message)))?;
202        self.fetch_prepared_all(&query).await
203    }
204
205    async fn prepare_continuous_page(
206        &self,
207        query: SelectQuery,
208    ) -> (SelectQuery, Option<ContinuousPageExecution>) {
209        let Some(options) = query.continuous_page_fetch.as_ref() else {
210            self.data_service
211                .metadata
212                .context
213                .observe_continuous_page("DISABLED", None);
214            return (query, None);
215        };
216        let Some(slice) = query.slice.as_ref() else {
217            self.data_service
218                .metadata
219                .context
220                .observe_continuous_page("OFFSET_FALLBACK:INVALID_SLICE", None);
221            return (query, None);
222        };
223        let Some(page_size) = slice.limit else {
224            self.data_service
225                .metadata
226                .context
227                .observe_continuous_page("OFFSET_FALLBACK:INVALID_SLICE", None);
228            return (query, None);
229        };
230        if query.partition_by.is_some()
231            || !query.aggregates.is_empty()
232            || !query.group_by.is_empty()
233        {
234            self.data_service
235                .metadata
236                .context
237                .observe_continuous_page("OFFSET_FALLBACK:UNSUPPORTED_QUERY_SHAPE", None);
238            return (query, None);
239        }
240        if query.order_by.len() != 1
241            || query.order_by[0].field != "id"
242            || query.order_by[0].expr.is_some()
243        {
244            self.data_service
245                .metadata
246                .context
247                .observe_continuous_page("OFFSET_FALLBACK:ORDER_NOT_SEEKABLE_ID", None);
248            return (query, None);
249        }
250        let direction = query.order_by[0].direction;
251        let query_key = self.continuous_page_query_key(&query, &options.namespace);
252        let execution = ContinuousPageExecution {
253            query_key: query_key.clone(),
254            direction,
255            page_size,
256            original_offset: slice.offset,
257            ttl_seconds: options.ttl_seconds,
258            optimized: false,
259            seek_cursor_id: None,
260        };
261        if slice.offset == 0 {
262            self.data_service
263                .metadata
264                .context
265                .observe_continuous_page("OFFSET_FALLBACK:FIRST_PAGE", None);
266            return (query, Some(execution));
267        }
268        let cursor = match self
269            .data_service
270            .metadata
271            .context
272            .continuous_page_cursor_store()
273            .get(&query_key, slice.offset)
274            .await
275        {
276            Ok(Some(cursor)) => cursor,
277            Ok(None) => {
278                self.data_service
279                    .metadata
280                    .context
281                    .observe_continuous_page("OFFSET_FALLBACK:CACHE_MISS", None);
282                return (query, Some(execution));
283            }
284            Err(_) => {
285                self.data_service
286                    .metadata
287                    .context
288                    .observe_continuous_page("OFFSET_FALLBACK:STORE_UNAVAILABLE", None);
289                return (query, Some(execution));
290            }
291        };
292        if cursor.entity != query.entity
293            || cursor.direction != direction
294            || cursor.page_size != page_size
295            || cursor.next_offset != slice.offset
296            || cursor.expires_at <= SystemTime::now()
297        {
298            self.data_service
299                .metadata
300                .context
301                .observe_continuous_page("OFFSET_FALLBACK:CURSOR_INVALID", None);
302            return (query, Some(execution));
303        }
304        let mut optimized = query;
305        optimized.slice.as_mut().expect("validated slice").offset = 0;
306        optimized = optimized.and_filter(match direction {
307            SortDirection::Asc => Expr::gt("id", cursor.boundary.clone()),
308            SortDirection::Desc => Expr::lt("id", cursor.boundary.clone()),
309        });
310        let seek_cursor_id = cursor.cursor_id;
311        self.data_service
312            .metadata
313            .context
314            .observe_continuous_page("CURSOR_SEEK", Some(seek_cursor_id.clone()));
315        (
316            optimized,
317            Some(ContinuousPageExecution {
318                optimized: true,
319                seek_cursor_id: Some(seek_cursor_id),
320                ..execution
321            }),
322        )
323    }
324
325    async fn register_continuous_page(
326        &self,
327        execution: &Option<ContinuousPageExecution>,
328        rows: &[Record],
329    ) {
330        let Some(execution) = execution else { return };
331        if rows.len() as u64 != execution.page_size {
332            return;
333        }
334        let Some(boundary) = rows.last().and_then(|row| row.get("id")).cloned() else {
335            return;
336        };
337        let cursor_id = format!(
338            "cpg_{:x}",
339            SystemTime::now()
340                .duration_since(SystemTime::UNIX_EPOCH)
341                .unwrap_or_default()
342                .as_nanos()
343        );
344        let cursor = ContinuousPageCursor {
345            cursor_id,
346            query_key: execution.query_key.clone(),
347            entity: self.entity.clone(),
348            direction: execution.direction,
349            boundary,
350            page_size: execution.page_size,
351            next_offset: execution.original_offset + rows.len() as u64,
352            expires_at: SystemTime::now() + Duration::from_secs(execution.ttl_seconds),
353        };
354        if self
355            .data_service
356            .metadata
357            .context
358            .continuous_page_cursor_store()
359            .put(cursor)
360            .await
361            .is_err()
362        {
363            self.data_service
364                .metadata
365                .context
366                .observe_continuous_page("OFFSET_FALLBACK:STORE_UNAVAILABLE", None);
367        } else if execution.optimized {
368            self.data_service
369                .metadata
370                .context
371                .observe_continuous_page("CURSOR_SEEK", execution.seek_cursor_id.clone());
372        } else {
373            self.data_service
374                .metadata
375                .context
376                .observe_continuous_page("OFFSET_FALLBACK:FIRST_PAGE", None);
377        }
378    }
379
380    fn continuous_page_query_key(&self, query: &SelectQuery, namespace: &str) -> String {
381        let mut normalized = query.clone();
382        if let Some(slice) = normalized.slice.as_mut() {
383            slice.offset = 0;
384        }
385        normalized.comment = None;
386        normalized.trace_chain.clear();
387        let mut hasher = std::collections::hash_map::DefaultHasher::new();
388        namespace.hash(&mut hasher);
389        format!("{normalized:?}").hash(&mut hasher);
390        self.data_service
391            .metadata
392            .context
393            .user_identifier()
394            .hash(&mut hasher);
395        format!("teaql:continuous-page:v1:{:016x}", hasher.finish())
396    }
397
398    /// Fetch root records from the provider cursor without materializing them.
399    /// Relation and aggregate enhancement needs a separate batched protocol and
400    /// is rejected here instead of silently returning incomplete entities.
401    pub(crate) async fn fetch_stream_internal(
402        &self,
403        query: &SelectQuery,
404    ) -> Result<
405        std::pin::Pin<
406            Box<
407                dyn futures_core::Stream<
408                        Item = Result<teaql_data_service::StreamChunk, DataServiceError<E::Error>>,
409                    > + '_,
410            >,
411        >,
412        DataServiceError<E::Error>,
413    >
414    where
415        E: teaql_data_service::StreamQueryExecutor,
416    {
417        let query = self
418            .prepare_select_query(query)
419            .map_err(DataServiceError::Runtime)?;
420        let query = query
421            .prepare_for_list()
422            .map_err(|message| DataServiceError::Runtime(RuntimeError::Graph(message)))?;
423
424        if !query.relations.is_empty()
425            || !query.child_enhancements.is_empty()
426            || !query.object_group_bys.is_empty()
427        {
428            return Err(DataServiceError::Runtime(RuntimeError::Graph(
429                "streaming relation or aggregate enhancement is not supported; stream a root query or use execute_for_list"
430                    .to_owned(),
431            )));
432        }
433
434        let chunk_size = query
435            .stream_config
436            .as_ref()
437            .map(|c| c.chunk_size)
438            .unwrap_or(1000);
439
440        let final_comment = self
441            .data_service
442            .resolve_final_comment(&query.trace_chain, query.comment.clone());
443        let mut query = query.clone();
444        query.comment = final_comment;
445
446        let request = teaql_data_service::QueryRequest {
447            query: query.clone(),
448            trace_chain: query.trace_chain.clone(),
449            comment: query.comment.clone(),
450        };
451
452        let chunks = self.data_service.executor.query_stream(request, chunk_size);
453        use futures_util::StreamExt;
454        Ok(Box::pin(
455            chunks.map(|item| item.map_err(DataServiceError::Executor)),
456        ))
457    }
458
459    async fn fetch_prepared_all(
460        &self,
461        query: &SelectQuery,
462    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
463        let query = query
464            .clone()
465            .prepare_for_list()
466            .map_err(|message| DataServiceError::Runtime(RuntimeError::Graph(message)))?;
467        let (execution_query, continuous) = self.prepare_continuous_page(query).await;
468        let mut rows = self.fetch_prepared_query(&execution_query).await?;
469        self.enhance_object_group_bys_internal(
470            &mut rows,
471            &execution_query.object_group_bys,
472            &execution_query.trace_chain,
473        )
474        .await?;
475        self.enhance_child_queries_internal(
476            &mut rows,
477            &execution_query.child_enhancements,
478            &execution_query.trace_chain,
479        )
480        .await?;
481        self.enhance_query_relations_internal(&mut rows, &execution_query)
482            .await?;
483        self.register_continuous_page(&continuous, &rows).await;
484        Ok(rows)
485    }
486
487    async fn fetch_prepared_query(
488        &self,
489        query: &SelectQuery,
490    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
491        let final_comment = self
492            .data_service
493            .resolve_final_comment(&query.trace_chain, query.comment.clone());
494        let mut query = query.clone();
495        query.comment = final_comment;
496        if let Some(options) = query.aggregation_cache.filter(|options| options.enabled) {
497            if let Some(cache) = self
498                .data_service
499                .metadata
500                .context
501                .get_resource::<Arc<dyn AggregationCacheBackend>>()
502            {
503                return self
504                    .fetch_prepared_query_with_cache(&query, options, cache.as_ref())
505                    .await;
506            }
507            if let Some(cache) = self
508                .data_service
509                .metadata
510                .context
511                .get_resource::<InMemoryAggregationCache>()
512            {
513                return self
514                    .fetch_prepared_query_with_cache(&query, options, cache)
515                    .await;
516            }
517        }
518        let request = teaql_data_service::QueryRequest {
519            query: query.clone(),
520            trace_chain: query.trace_chain.clone(),
521            comment: query.comment.clone(),
522        };
523        let res = self
524            .data_service
525            .executor
526            .query(request)
527            .await
528            .map_err(DataServiceError::Executor)?;
529        self.data_service
530            .metadata
531            .context
532            .record_metadata_log(&res.metadata);
533        Ok(res.rows)
534    }
535
536    async fn fetch_prepared_query_with_cache(
537        &self,
538        query: &SelectQuery,
539        options: AggregationCacheOptions,
540        cache: &dyn AggregationCacheBackend,
541    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
542        let key = aggregation_cache_key(
543            cache.namespace(),
544            &aggregation_cache_namespace(&query.entity),
545            query,
546        );
547        if let Some(rows) = cache.get(&key, options.cache_expired_millis) {
548            return Ok(rows);
549        }
550        let request = teaql_data_service::QueryRequest {
551            query: query.clone(),
552            trace_chain: query.trace_chain.clone(),
553            comment: query.comment.clone(),
554        };
555        let res = self
556            .data_service
557            .executor
558            .query(request)
559            .await
560            .map_err(DataServiceError::Executor)?;
561        self.data_service
562            .metadata
563            .context
564            .record_metadata_log(&res.metadata);
565        let rows = res.rows;
566        cache.put(key, rows.clone());
567        Ok(rows)
568    }
569
570    pub(crate) async fn fetch_all_with_relation_aggregates_internal(
571        &self,
572        query: &SelectQuery,
573        relation_aggregates: &[RelationAggregate],
574    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
575        let query = self
576            .prepare_select_query(query)
577            .map_err(DataServiceError::Runtime)?;
578
579        let mut rows = self.fetch_prepared_all(&query).await?;
580        self.enhance_relation_aggregates_internal(
581            &mut rows,
582            relation_aggregates,
583            query.aggregation_cache,
584            &query.trace_chain,
585        )
586        .await?;
587        Ok(rows)
588    }
589
590    pub(crate) async fn fetch_smart_list_internal(
591        &self,
592        query: &SelectQuery,
593    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
594        let query = self
595            .prepare_select_query(query)
596            .map_err(DataServiceError::Runtime)?;
597
598        self.data_service.fetch_smart_list(&query).await
599    }
600
601    pub(crate) async fn fetch_smart_list_with_relation_aggregates_internal(
602        &self,
603        query: &SelectQuery,
604        relation_aggregates: &[RelationAggregate],
605    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
606        self.fetch_all_with_relation_aggregates_internal(query, relation_aggregates)
607            .await
608            .map(SmartList::from)
609    }
610
611    pub(crate) async fn fetch_entities_internal<T>(
612        &self,
613        query: &SelectQuery,
614    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
615    where
616        T: Entity,
617    {
618        let query = self
619            .prepare_select_query(query)
620            .map_err(DataServiceError::Runtime)?;
621
622        self.data_service.fetch_entities(&query).await
623    }
624
625    pub(crate) async fn fetch_entities_with_relation_aggregates_internal<T>(
626        &self,
627        query: &SelectQuery,
628        relation_aggregates: &[RelationAggregate],
629    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
630    where
631        T: Entity,
632    {
633        self.fetch_all_with_relation_aggregates_internal(query, relation_aggregates)
634            .await?
635            .into_iter()
636            .map(|record| {
637                let mut entity = T::from_record(record)?;
638                let root = crate::EntityRoot::default();
639                entity.on_loaded(&root as &dyn std::any::Any);
640                Ok(entity)
641            })
642            .collect::<Result<Vec<_>, _>>()
643            .map(SmartList::from)
644            .map_err(DataServiceError::Entity)
645    }
646
647    pub(crate) async fn fetch_enhanced_entities_with_relation_aggregates_internal<T>(
648        &self,
649        query: &SelectQuery,
650        relation_aggregates: &[RelationAggregate],
651    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
652    where
653        T: Entity,
654    {
655        let query = self
656            .prepare_select_query(query)
657            .map_err(DataServiceError::Runtime)?;
658
659        let mut rows = self.fetch_prepared_all(&query).await?;
660        self.enhance_relation_aggregates_internal(
661            &mut rows,
662            relation_aggregates,
663            query.aggregation_cache,
664            &query.trace_chain,
665        )
666        .await?;
667        self.enhance_relations_internal(&mut rows).await?;
668        rows.into_iter()
669            .map(|record| {
670                let mut entity = T::from_record(record)?;
671                let root = crate::EntityRoot::default();
672                entity.on_loaded(&root as &dyn std::any::Any);
673                Ok(entity)
674            })
675            .collect::<Result<Vec<_>, _>>()
676            .map(SmartList::from)
677            .map_err(DataServiceError::Entity)
678    }
679
680    pub(crate) async fn fetch_enhanced_entities_internal<T>(
681        &self,
682        query: &SelectQuery,
683    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
684    where
685        T: Entity,
686    {
687        let query = self
688            .prepare_select_query(query)
689            .map_err(DataServiceError::Runtime)?;
690
691        let mut rows = self.fetch_prepared_all(&query).await?;
692        self.enhance_relations_internal(&mut rows).await?;
693        let root = self
694            .data_service
695            .metadata
696            .context
697            .get_resource::<crate::EntityRoot>()
698            .cloned();
699        rows.into_iter()
700            .map(|record| {
701                let mut entity = T::from_record(record)?;
702                if let Some(ref root) = root {
703                    entity.on_loaded(root as &dyn std::any::Any);
704                }
705                Ok(entity)
706            })
707            .collect::<Result<Vec<_>, _>>()
708            .map(SmartList::from)
709            .map_err(DataServiceError::Entity)
710    }
711
712    #[doc(hidden)]
713    pub async fn fetch_all(
714        &self,
715        query: &PurposedSelectQuery,
716    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
717        self.fetch_all_internal(query.as_query()).await
718    }
719
720    #[doc(hidden)]
721    pub async fn fetch_stream(
722        &self,
723        query: &PurposedSelectQuery,
724    ) -> Result<
725        std::pin::Pin<
726            Box<
727                dyn futures_core::Stream<
728                        Item = Result<teaql_data_service::StreamChunk, DataServiceError<E::Error>>,
729                    > + '_,
730            >,
731        >,
732        DataServiceError<E::Error>,
733    >
734    where
735        E: teaql_data_service::StreamQueryExecutor,
736    {
737        self.fetch_stream_internal(query.as_query()).await
738    }
739
740    #[doc(hidden)]
741    pub async fn fetch_smart_list(
742        &self,
743        query: &PurposedSelectQuery,
744    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
745        self.fetch_smart_list_internal(query.as_query()).await
746    }
747
748    #[doc(hidden)]
749    pub async fn fetch_smart_list_with_relation_aggregates(
750        &self,
751        query: &PurposedSelectQuery,
752        relation_aggregates: &[RelationAggregate],
753    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
754        self.fetch_smart_list_with_relation_aggregates_internal(
755            query.as_query(),
756            relation_aggregates,
757        )
758        .await
759    }
760
761    #[doc(hidden)]
762    pub async fn fetch_entities<T>(
763        &self,
764        query: &PurposedSelectQuery,
765    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
766    where
767        T: Entity,
768    {
769        self.fetch_entities_internal(query.as_query()).await
770    }
771
772    #[doc(hidden)]
773    pub async fn fetch_enhanced_entities<T>(
774        &self,
775        query: &PurposedSelectQuery,
776    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
777    where
778        T: Entity,
779    {
780        self.fetch_enhanced_entities_internal(query.as_query())
781            .await
782    }
783
784    #[doc(hidden)]
785    pub async fn fetch_enhanced_entities_with_relation_aggregates<T>(
786        &self,
787        query: &PurposedSelectQuery,
788        relation_aggregates: &[RelationAggregate],
789    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
790    where
791        T: Entity,
792    {
793        self.fetch_enhanced_entities_with_relation_aggregates_internal(
794            query.as_query(),
795            relation_aggregates,
796        )
797        .await
798    }
799
800    pub(crate) async fn insert_internal(
801        &self,
802        command: &InsertCommand,
803    ) -> Result<u64, DataServiceError<E::Error>> {
804        let command = self
805            .prepare_insert_command(command)
806            .map_err(DataServiceError::Runtime)?;
807        self.execute_prepared_insert_with_comment(command, self.trace_context.clone())
808            .await
809    }
810
811    pub(crate) async fn update_internal(
812        &self,
813        command: &UpdateCommand,
814    ) -> Result<u64, DataServiceError<E::Error>> {
815        let command = self
816            .prepare_update_command(command)
817            .map_err(DataServiceError::Runtime)?;
818        self.execute_prepared_update_with_comment(command, self.trace_context.clone())
819            .await
820    }
821
822    pub(crate) async fn delete_internal(
823        &self,
824        command: &DeleteCommand,
825    ) -> Result<u64, DataServiceError<E::Error>> {
826        self.delete_scoped_internal(command, self.trace_context.clone())
827            .await
828    }
829
830    pub(crate) async fn delete_scoped_internal(
831        &self,
832        command: &DeleteCommand,
833        trace_chain: Vec<teaql_core::TraceNode>,
834    ) -> Result<u64, DataServiceError<E::Error>> {
835        let mut command = command.clone();
836        command.trace_chain = trace_chain.clone();
837        if let Some(behavior) = self.behavior() {
838            behavior
839                .before_delete(self.data_service.metadata.context, &mut command)
840                .map_err(DataServiceError::Runtime)?;
841        }
842        self.enforce_delete_policy(&mut command)
843            .map_err(DataServiceError::Runtime)?;
844
845        let old_values =
846            self.fetch_current_event_row(&command.entity, &command.id, trace_chain.clone())?;
847        let affected = self.data_service.delete(&command).await?;
848
849        let mut event = RawAuditEvent::deleted_with_old_values(
850            command.entity,
851            command.id,
852            command.expected_version,
853            old_values,
854        );
855        event.trace_chain = trace_chain;
856        self.emit_event(event).map_err(DataServiceError::Runtime)?;
857        Ok(affected)
858    }
859
860    pub(crate) async fn recover_internal(
861        &self,
862        command: &RecoverCommand,
863    ) -> Result<u64, DataServiceError<E::Error>> {
864        let mut command = command.clone();
865        command.trace_chain = self.trace_context.clone();
866        if let Some(behavior) = self.behavior() {
867            behavior
868                .before_recover(self.data_service.metadata.context, &mut command)
869                .map_err(DataServiceError::Runtime)?;
870        }
871        self.enforce_recover_policy(&mut command)
872            .map_err(DataServiceError::Runtime)?;
873        let old_values = self.fetch_current_event_row(
874            &command.entity,
875            &command.id,
876            command.trace_chain.clone(),
877        )?;
878        let affected = self.data_service.recover(&command).await?;
879        let event = RawAuditEvent::recovered_with_old_values(
880            command.entity,
881            command.id,
882            command.expected_version,
883            old_values,
884        );
885        self.emit_event(event).map_err(DataServiceError::Runtime)?;
886        Ok(affected)
887    }
888
889    fn emit_event(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
890        self.data_service.metadata.context.send_event(event)
891    }
892
893    #[allow(dead_code)]
894    pub(super) async fn execute_prepared_insert(
895        &self,
896        command: InsertCommand,
897    ) -> Result<u64, DataServiceError<E::Error>> {
898        self.execute_prepared_insert_with_comment(command, Vec::new())
899            .await
900    }
901
902    pub(super) async fn execute_prepared_insert_with_comment(
903        &self,
904        mut command: InsertCommand,
905        trace_chain: Vec<teaql_core::TraceNode>,
906    ) -> Result<u64, DataServiceError<E::Error>> {
907        command.trace_chain = trace_chain.clone();
908        let affected = self.data_service.insert(&command).await?;
909        let mut event = RawAuditEvent::created(command.entity, command.values);
910        event.trace_chain = trace_chain;
911        self.emit_event(event).map_err(DataServiceError::Runtime)?;
912        Ok(affected)
913    }
914
915    pub(super) async fn execute_prepared_batch_insert(
916        &self,
917        command: teaql_core::BatchInsertCommand,
918    ) -> Result<u64, DataServiceError<E::Error>> {
919        if command.batch_values.is_empty() {
920            return Ok(0);
921        }
922        let affected = self.data_service.batch_insert(&command).await?;
923
924        let entity = command.entity.clone();
925        for (i, values) in command.batch_values.into_iter().enumerate() {
926            let mut event = RawAuditEvent::created(entity.clone(), values);
927            if i < command.trace_chains.len() {
928                event.trace_chain = command.trace_chains[i].clone();
929            }
930            self.emit_event(event).map_err(DataServiceError::Runtime)?;
931        }
932        Ok(affected)
933    }
934
935    #[allow(dead_code)]
936    pub(super) async fn execute_prepared_update(
937        &self,
938        command: UpdateCommand,
939    ) -> Result<u64, DataServiceError<E::Error>> {
940        self.execute_prepared_update_with_comment(command, Vec::new())
941            .await
942    }
943
944    pub(super) async fn execute_prepared_update_with_comment(
945        &self,
946        mut command: UpdateCommand,
947        trace_chain: Vec<teaql_core::TraceNode>,
948    ) -> Result<u64, DataServiceError<E::Error>> {
949        command.trace_chain = trace_chain.clone();
950
951        let mut old_values = command.old_values.clone();
952        let needs_fetch = match &old_values {
953            Some(snapshot) => !command.values.keys().all(|k| snapshot.contains_key(k)),
954            None => true,
955        };
956        if needs_fetch {
957            old_values =
958                self.fetch_current_event_row(&command.entity, &command.id, trace_chain.clone())?;
959        }
960
961        let affected = self.data_service.update(&command).await?;
962        let updated_fields = command.values.keys().cloned().collect();
963        let mut values = command.values.clone();
964        values.insert("id".to_owned(), command.id.clone());
965        if let Some(version) = command.expected_version {
966            values.insert("version".to_owned(), Value::I64(version + 1));
967        }
968        let mut new_values = old_values.clone().unwrap_or_default();
969        for (field, value) in &values {
970            new_values.insert(field.clone(), value.clone());
971        }
972        let mut event = RawAuditEvent::updated_with_old_values(
973            command.entity,
974            values,
975            old_values,
976            new_values,
977            updated_fields,
978        );
979        event.trace_chain = trace_chain;
980        self.emit_event(event).map_err(DataServiceError::Runtime)?;
981        Ok(affected)
982    }
983
984    pub(super) async fn execute_prepared_batch_update(
985        &self,
986        command: teaql_core::BatchUpdateCommand,
987    ) -> Result<u64, DataServiceError<E::Error>> {
988        if command.batch_values.is_empty() {
989            return Ok(0);
990        }
991        let affected = self.data_service.batch_update(&command).await?;
992
993        let entity = command.entity.clone();
994        for (i, values) in command.batch_values.into_iter().enumerate() {
995            let mut full_values = values.clone();
996            full_values.insert("id".to_owned(), command.batch_ids[i].clone());
997            if let Some(Some(version)) = command.batch_expected_versions.get(i) {
998                full_values.insert("version".to_owned(), teaql_core::Value::I64(*version + 1));
999            }
1000
1001            let old_values = command.batch_old_values.get(i).cloned().unwrap_or(None);
1002            let mut new_values = old_values.clone().unwrap_or_default();
1003            for (field, value) in &full_values {
1004                new_values.insert(field.clone(), value.clone());
1005            }
1006
1007            let mut event = RawAuditEvent::updated_with_old_values(
1008                entity.clone(),
1009                full_values,
1010                old_values,
1011                new_values,
1012                command.update_fields.clone(),
1013            );
1014            if i < command.trace_chains.len() {
1015                event.trace_chain = command.trace_chains[i].clone();
1016            }
1017            self.emit_event(event).map_err(DataServiceError::Runtime)?;
1018        }
1019        Ok(affected)
1020    }
1021
1022    fn fetch_current_event_row(
1023        &self,
1024        _entity: &str,
1025        _id: &Value,
1026        _trace_chain: Vec<teaql_core::TraceNode>,
1027    ) -> Result<Option<Record>, DataServiceError<E::Error>> {
1028        // PER THE USER: "我们不需要在审计的时候去抓旧的值"
1029        // Avoid DB queries during event emission. We rely on in-memory `original_values`.
1030        Ok(None)
1031    }
1032
1033    pub(crate) fn scoped_data_service_internal(&self, entity: String) -> EntityDataService<'a, E> {
1034        EntityDataService {
1035            entity,
1036            data_service: ContextDataService {
1037                metadata: UserContextMetadata {
1038                    context: self.data_service.metadata.context,
1039                },
1040                executor: self.data_service.executor,
1041            },
1042            trace_context: Vec::new(),
1043        }
1044    }
1045}
teaql_runtime/data_service/resolved.rs

teaql_runtime/data_service/
resolved.rs
