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        let scope = self.data_service.metadata.context.start_runtime_operation(
548            crate::RuntimeOperation::new("cache", format!("{}.aggregation.get", query.entity))
549                .attribute("teaql.cache.operation", "get"),
550        );
551        let result = scope
552            .run(async {
553                if let Some(rows) = cache.get(&key, options.cache_expired_millis) {
554                    return Ok((rows, "hit"));
555                }
556                let request = teaql_data_service::QueryRequest {
557                    query: query.clone(),
558                    trace_chain: query.trace_chain.clone(),
559                    comment: query.comment.clone(),
560                };
561                let provider_kind = std::any::type_name::<E>().to_owned();
562                let provider_scope = self.data_service.metadata.context.start_runtime_operation(
563                    crate::RuntimeOperation::new("provider", format!("{provider_kind}.query"))
564                        .attribute("teaql.provider.kind", provider_kind)
565                        .attribute("teaql.provider.operation", "query"),
566                );
567                let provider_result = provider_scope
568                    .run(self.data_service.executor.query(request))
569                    .await;
570                let res = match provider_result {
571                    Ok(value) => {
572                        provider_scope.success(std::collections::BTreeMap::new());
573                        value
574                    }
575                    Err(error) => {
576                        provider_scope.failure("data_service_error");
577                        return Err(DataServiceError::Executor(error));
578                    }
579                };
580                self.data_service
581                    .metadata
582                    .context
583                    .record_metadata_log(&res.metadata);
584                let rows = res.rows;
585                cache.put(key, rows.clone());
586                Ok((rows, "miss"))
587            })
588            .await;
589        match result {
590            Ok((rows, cache_result)) => {
591                scope.success(std::collections::BTreeMap::from([(
592                    "teaql.cache.result".to_owned(),
593                    crate::RuntimeAttributeValue::from(cache_result),
594                )]));
595                Ok(rows)
596            }
597            Err(error) => {
598                scope.failure("cache_load_error");
599                Err(error)
600            }
601        }
602    }
603
604    pub(crate) async fn fetch_all_with_relation_aggregates_internal(
605        &self,
606        query: &SelectQuery,
607        relation_aggregates: &[RelationAggregate],
608    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
609        let query = self
610            .prepare_select_query(query)
611            .map_err(DataServiceError::Runtime)?;
612
613        let mut rows = self.fetch_prepared_all(&query).await?;
614        self.enhance_relation_aggregates_internal(
615            &mut rows,
616            relation_aggregates,
617            query.aggregation_cache,
618            &query.trace_chain,
619        )
620        .await?;
621        Ok(rows)
622    }
623
624    pub(crate) async fn fetch_smart_list_internal(
625        &self,
626        query: &SelectQuery,
627    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
628        let query = self
629            .prepare_select_query(query)
630            .map_err(DataServiceError::Runtime)?;
631
632        self.data_service.fetch_smart_list(&query).await
633    }
634
635    pub(crate) async fn fetch_smart_list_with_relation_aggregates_internal(
636        &self,
637        query: &SelectQuery,
638        relation_aggregates: &[RelationAggregate],
639    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
640        self.fetch_all_with_relation_aggregates_internal(query, relation_aggregates)
641            .await
642            .map(SmartList::from)
643    }
644
645    pub(crate) async fn fetch_entities_internal<T>(
646        &self,
647        query: &SelectQuery,
648    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
649    where
650        T: Entity,
651    {
652        let query = self
653            .prepare_select_query(query)
654            .map_err(DataServiceError::Runtime)?;
655
656        self.data_service.fetch_entities(&query).await
657    }
658
659    pub(crate) async fn fetch_entities_with_relation_aggregates_internal<T>(
660        &self,
661        query: &SelectQuery,
662        relation_aggregates: &[RelationAggregate],
663    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
664    where
665        T: Entity,
666    {
667        self.fetch_all_with_relation_aggregates_internal(query, relation_aggregates)
668            .await?
669            .into_iter()
670            .map(|record| {
671                let mut entity = T::from_record(record)?;
672                let root = crate::EntityRoot::default();
673                entity.on_loaded(&root as &dyn std::any::Any);
674                Ok(entity)
675            })
676            .collect::<Result<Vec<_>, _>>()
677            .map(SmartList::from)
678            .map_err(DataServiceError::Entity)
679    }
680
681    pub(crate) async fn fetch_enhanced_entities_with_relation_aggregates_internal<T>(
682        &self,
683        query: &SelectQuery,
684        relation_aggregates: &[RelationAggregate],
685    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
686    where
687        T: Entity,
688    {
689        let query = self
690            .prepare_select_query(query)
691            .map_err(DataServiceError::Runtime)?;
692
693        let mut rows = self.fetch_prepared_all(&query).await?;
694        self.enhance_relation_aggregates_internal(
695            &mut rows,
696            relation_aggregates,
697            query.aggregation_cache,
698            &query.trace_chain,
699        )
700        .await?;
701        self.enhance_relations_internal(&mut rows).await?;
702        rows.into_iter()
703            .map(|record| {
704                let mut entity = T::from_record(record)?;
705                let root = crate::EntityRoot::default();
706                entity.on_loaded(&root as &dyn std::any::Any);
707                Ok(entity)
708            })
709            .collect::<Result<Vec<_>, _>>()
710            .map(SmartList::from)
711            .map_err(DataServiceError::Entity)
712    }
713
714    pub(crate) async fn fetch_enhanced_entities_internal<T>(
715        &self,
716        query: &SelectQuery,
717    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
718    where
719        T: Entity,
720    {
721        let query = self
722            .prepare_select_query(query)
723            .map_err(DataServiceError::Runtime)?;
724
725        let mut rows = self.fetch_prepared_all(&query).await?;
726        self.enhance_relations_internal(&mut rows).await?;
727        let root = self
728            .data_service
729            .metadata
730            .context
731            .get_resource::<crate::EntityRoot>()
732            .cloned();
733        rows.into_iter()
734            .map(|record| {
735                let mut entity = T::from_record(record)?;
736                if let Some(ref root) = root {
737                    entity.on_loaded(root as &dyn std::any::Any);
738                }
739                Ok(entity)
740            })
741            .collect::<Result<Vec<_>, _>>()
742            .map(SmartList::from)
743            .map_err(DataServiceError::Entity)
744    }
745
746    #[doc(hidden)]
747    pub async fn fetch_all(
748        &self,
749        query: &PurposedSelectQuery,
750    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
751        self.fetch_all_internal(query.as_query()).await
752    }
753
754    #[doc(hidden)]
755    pub async fn fetch_stream(
756        &self,
757        query: &PurposedSelectQuery,
758    ) -> Result<
759        std::pin::Pin<
760            Box<
761                dyn futures_core::Stream<
762                        Item = Result<teaql_data_service::StreamChunk, DataServiceError<E::Error>>,
763                    > + '_,
764            >,
765        >,
766        DataServiceError<E::Error>,
767    >
768    where
769        E: teaql_data_service::StreamQueryExecutor,
770    {
771        self.fetch_stream_internal(query.as_query()).await
772    }
773
774    #[doc(hidden)]
775    pub async fn fetch_smart_list(
776        &self,
777        query: &PurposedSelectQuery,
778    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
779        self.fetch_smart_list_internal(query.as_query()).await
780    }
781
782    #[doc(hidden)]
783    pub async fn fetch_smart_list_with_relation_aggregates(
784        &self,
785        query: &PurposedSelectQuery,
786        relation_aggregates: &[RelationAggregate],
787    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
788        self.fetch_smart_list_with_relation_aggregates_internal(
789            query.as_query(),
790            relation_aggregates,
791        )
792        .await
793    }
794
795    #[doc(hidden)]
796    pub async fn fetch_entities<T>(
797        &self,
798        query: &PurposedSelectQuery,
799    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
800    where
801        T: Entity,
802    {
803        self.fetch_entities_internal(query.as_query()).await
804    }
805
806    #[doc(hidden)]
807    pub async fn fetch_enhanced_entities<T>(
808        &self,
809        query: &PurposedSelectQuery,
810    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
811    where
812        T: Entity,
813    {
814        self.fetch_enhanced_entities_internal(query.as_query())
815            .await
816    }
817
818    #[doc(hidden)]
819    pub async fn fetch_enhanced_entities_with_relation_aggregates<T>(
820        &self,
821        query: &PurposedSelectQuery,
822        relation_aggregates: &[RelationAggregate],
823    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
824    where
825        T: Entity,
826    {
827        self.fetch_enhanced_entities_with_relation_aggregates_internal(
828            query.as_query(),
829            relation_aggregates,
830        )
831        .await
832    }
833
834    pub(crate) async fn insert_internal(
835        &self,
836        command: &InsertCommand,
837    ) -> Result<u64, DataServiceError<E::Error>> {
838        let command = self
839            .prepare_insert_command(command)
840            .map_err(DataServiceError::Runtime)?;
841        self.execute_prepared_insert_with_comment(command, self.trace_context.clone())
842            .await
843    }
844
845    pub(crate) async fn update_internal(
846        &self,
847        command: &UpdateCommand,
848    ) -> Result<u64, DataServiceError<E::Error>> {
849        let command = self
850            .prepare_update_command(command)
851            .map_err(DataServiceError::Runtime)?;
852        self.execute_prepared_update_with_comment(command, self.trace_context.clone())
853            .await
854    }
855
856    pub(crate) async fn delete_internal(
857        &self,
858        command: &DeleteCommand,
859    ) -> Result<u64, DataServiceError<E::Error>> {
860        self.delete_scoped_internal(command, self.trace_context.clone())
861            .await
862    }
863
864    pub(crate) async fn delete_scoped_internal(
865        &self,
866        command: &DeleteCommand,
867        trace_chain: Vec<teaql_core::TraceNode>,
868    ) -> Result<u64, DataServiceError<E::Error>> {
869        let mut command = command.clone();
870        command.trace_chain = trace_chain.clone();
871        if let Some(behavior) = self.behavior() {
872            behavior
873                .before_delete(self.data_service.metadata.context, &mut command)
874                .map_err(DataServiceError::Runtime)?;
875        }
876        self.enforce_delete_policy(&mut command)
877            .map_err(DataServiceError::Runtime)?;
878
879        let old_values =
880            self.fetch_current_event_row(&command.entity, &command.id, trace_chain.clone())?;
881        let affected = self.data_service.delete(&command).await?;
882
883        let mut event = RawAuditEvent::deleted_with_old_values(
884            command.entity,
885            command.id,
886            command.expected_version,
887            old_values,
888        );
889        event.trace_chain = trace_chain;
890        self.emit_event(event).map_err(DataServiceError::Runtime)?;
891        Ok(affected)
892    }
893
894    pub(crate) async fn recover_internal(
895        &self,
896        command: &RecoverCommand,
897    ) -> Result<u64, DataServiceError<E::Error>> {
898        let mut command = command.clone();
899        command.trace_chain = self.trace_context.clone();
900        if let Some(behavior) = self.behavior() {
901            behavior
902                .before_recover(self.data_service.metadata.context, &mut command)
903                .map_err(DataServiceError::Runtime)?;
904        }
905        self.enforce_recover_policy(&mut command)
906            .map_err(DataServiceError::Runtime)?;
907        let old_values = self.fetch_current_event_row(
908            &command.entity,
909            &command.id,
910            command.trace_chain.clone(),
911        )?;
912        let affected = self.data_service.recover(&command).await?;
913        let event = RawAuditEvent::recovered_with_old_values(
914            command.entity,
915            command.id,
916            command.expected_version,
917            old_values,
918        );
919        self.emit_event(event).map_err(DataServiceError::Runtime)?;
920        Ok(affected)
921    }
922
923    fn emit_event(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
924        self.data_service.metadata.context.send_event(event)
925    }
926
927    #[allow(dead_code)]
928    pub(super) async fn execute_prepared_insert(
929        &self,
930        command: InsertCommand,
931    ) -> Result<u64, DataServiceError<E::Error>> {
932        self.execute_prepared_insert_with_comment(command, Vec::new())
933            .await
934    }
935
936    pub(super) async fn execute_prepared_insert_with_comment(
937        &self,
938        mut command: InsertCommand,
939        trace_chain: Vec<teaql_core::TraceNode>,
940    ) -> Result<u64, DataServiceError<E::Error>> {
941        command.trace_chain = trace_chain.clone();
942        let affected = self.data_service.insert(&command).await?;
943        let mut event = RawAuditEvent::created(command.entity, command.values);
944        event.trace_chain = trace_chain;
945        self.emit_event(event).map_err(DataServiceError::Runtime)?;
946        Ok(affected)
947    }
948
949    pub(super) async fn execute_prepared_batch_insert(
950        &self,
951        command: teaql_core::BatchInsertCommand,
952    ) -> Result<u64, DataServiceError<E::Error>> {
953        if command.batch_values.is_empty() {
954            return Ok(0);
955        }
956        let affected = self.data_service.batch_insert(&command).await?;
957
958        let entity = command.entity.clone();
959        for (i, values) in command.batch_values.into_iter().enumerate() {
960            let mut event = RawAuditEvent::created(entity.clone(), values);
961            if i < command.trace_chains.len() {
962                event.trace_chain = command.trace_chains[i].clone();
963            }
964            self.emit_event(event).map_err(DataServiceError::Runtime)?;
965        }
966        Ok(affected)
967    }
968
969    #[allow(dead_code)]
970    pub(super) async fn execute_prepared_update(
971        &self,
972        command: UpdateCommand,
973    ) -> Result<u64, DataServiceError<E::Error>> {
974        self.execute_prepared_update_with_comment(command, Vec::new())
975            .await
976    }
977
978    pub(super) async fn execute_prepared_update_with_comment(
979        &self,
980        mut command: UpdateCommand,
981        trace_chain: Vec<teaql_core::TraceNode>,
982    ) -> Result<u64, DataServiceError<E::Error>> {
983        command.trace_chain = trace_chain.clone();
984
985        let mut old_values = command.old_values.clone();
986        let needs_fetch = match &old_values {
987            Some(snapshot) => !command.values.keys().all(|k| snapshot.contains_key(k)),
988            None => true,
989        };
990        if needs_fetch {
991            old_values =
992                self.fetch_current_event_row(&command.entity, &command.id, trace_chain.clone())?;
993        }
994
995        let affected = self.data_service.update(&command).await?;
996        let updated_fields = command.values.keys().cloned().collect();
997        let mut values = command.values.clone();
998        values.insert("id".to_owned(), command.id.clone());
999        if let Some(version) = command.expected_version {
1000            values.insert("version".to_owned(), Value::I64(version + 1));
1001        }
1002        let mut new_values = old_values.clone().unwrap_or_default();
1003        for (field, value) in &values {
1004            new_values.insert(field.clone(), value.clone());
1005        }
1006        let mut event = RawAuditEvent::updated_with_old_values(
1007            command.entity,
1008            values,
1009            old_values,
1010            new_values,
1011            updated_fields,
1012        );
1013        event.trace_chain = trace_chain;
1014        self.emit_event(event).map_err(DataServiceError::Runtime)?;
1015        Ok(affected)
1016    }
1017
1018    pub(super) async fn execute_prepared_batch_update(
1019        &self,
1020        command: teaql_core::BatchUpdateCommand,
1021    ) -> Result<u64, DataServiceError<E::Error>> {
1022        if command.batch_values.is_empty() {
1023            return Ok(0);
1024        }
1025        let affected = self.data_service.batch_update(&command).await?;
1026
1027        let entity = command.entity.clone();
1028        for (i, values) in command.batch_values.into_iter().enumerate() {
1029            let mut full_values = values.clone();
1030            full_values.insert("id".to_owned(), command.batch_ids[i].clone());
1031            if let Some(Some(version)) = command.batch_expected_versions.get(i) {
1032                full_values.insert("version".to_owned(), teaql_core::Value::I64(*version + 1));
1033            }
1034
1035            let old_values = command.batch_old_values.get(i).cloned().unwrap_or(None);
1036            let mut new_values = old_values.clone().unwrap_or_default();
1037            for (field, value) in &full_values {
1038                new_values.insert(field.clone(), value.clone());
1039            }
1040
1041            let mut event = RawAuditEvent::updated_with_old_values(
1042                entity.clone(),
1043                full_values,
1044                old_values,
1045                new_values,
1046                command.update_fields.clone(),
1047            );
1048            if i < command.trace_chains.len() {
1049                event.trace_chain = command.trace_chains[i].clone();
1050            }
1051            self.emit_event(event).map_err(DataServiceError::Runtime)?;
1052        }
1053        Ok(affected)
1054    }
1055
1056    fn fetch_current_event_row(
1057        &self,
1058        _entity: &str,
1059        _id: &Value,
1060        _trace_chain: Vec<teaql_core::TraceNode>,
1061    ) -> Result<Option<Record>, DataServiceError<E::Error>> {
1062        // PER THE USER: "我们不需要在审计的时候去抓旧的值"
1063        // Avoid DB queries during event emission. We rely on in-memory `original_values`.
1064        Ok(None)
1065    }
1066
1067    pub(crate) fn scoped_data_service_internal(&self, entity: String) -> EntityDataService<'a, E> {
1068        EntityDataService {
1069            entity,
1070            data_service: ContextDataService {
1071                metadata: UserContextMetadata {
1072                    context: self.data_service.metadata.context,
1073                },
1074                executor: self.data_service.executor,
1075            },
1076            trace_context: Vec::new(),
1077        }
1078    }
1079}
teaql_runtime/data_service/resolved.rs

teaql_runtime/data_service/
resolved.rs
