teaql_runtime/data_service/
resolved.rs

1use std::sync::Arc;
2
3use teaql_core::{
4    AggregationCacheOptions, DeleteCommand, Entity, InsertCommand, Record, RecoverCommand,
5    RelationAggregate, SelectQuery, SmartList, UpdateCommand, Value,
6};
7
8use crate::{
9    CheckObjectStatus, DataServiceError, EntityDataServiceBehavior, RawAuditEvent, RuntimeError,
10    clear_record_status, mark_record_status,
11};
12
13use super::{
14    AggregationCacheBackend, ContextDataService, EntityDataService, InMemoryAggregationCache,
15    UserContextMetadata, helpers::*,
16};
17
18impl<'a, E> EntityDataService<'a, E>
19where
20    E: teaql_data_service::QueryExecutor
21        + teaql_data_service::MutationExecutor
22        + Send
23        + Sync
24        + 'static,
25{
26    pub(super) fn query_behavior(
27        &self,
28        entity: &str,
29    ) -> Option<Arc<dyn EntityDataServiceBehavior>> {
30        self.data_service
31            .metadata
32            .context
33            .entity_data_service_behavior(entity)
34    }
35
36    pub(super) fn behavior(&self) -> Option<Arc<dyn EntityDataServiceBehavior>> {
37        self.data_service
38            .metadata
39            .context
40            .entity_data_service_behavior(&self.entity)
41    }
42
43    pub fn entity(&self) -> &str {
44        &self.entity
45    }
46
47    pub fn select(&self) -> SelectQuery {
48        SelectQuery::new(self.entity.clone())
49    }
50
51    pub fn insert_command(&self) -> InsertCommand {
52        InsertCommand::new(self.entity.clone())
53    }
54
55    fn enforce_insert_policy(&self, command: &mut InsertCommand) -> Result<(), RuntimeError> {
56        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
57            policy.enforce_insert(self.data_service.metadata.context, command)?;
58        }
59        Ok(())
60    }
61
62    fn enforce_update_policy(&self, command: &mut UpdateCommand) -> Result<(), RuntimeError> {
63        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
64            policy.enforce_update(self.data_service.metadata.context, command)?;
65        }
66        Ok(())
67    }
68
69    fn enforce_delete_policy(&self, command: &mut DeleteCommand) -> Result<(), RuntimeError> {
70        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
71            policy.enforce_delete(self.data_service.metadata.context, command)?;
72        }
73        Ok(())
74    }
75
76    fn enforce_recover_policy(&self, command: &mut RecoverCommand) -> Result<(), RuntimeError> {
77        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
78            policy.enforce_recover(self.data_service.metadata.context, command)?;
79        }
80        Ok(())
81    }
82
83    fn prepare_select_query(&self, query: &SelectQuery) -> Result<SelectQuery, RuntimeError> {
84        let mut query = query.clone();
85
86        let mut full_trace = self.trace_context.clone();
87        full_trace.extend(query.trace_chain);
88        query.trace_chain = full_trace;
89
90        if let Some(behavior) = self.query_behavior(&query.entity) {
91            behavior.before_select(self.data_service.metadata.context, &mut query)?;
92        }
93        if let Some(policy) = self.data_service.metadata.context.request_policy.as_ref() {
94            policy.enforce_select(self.data_service.metadata.context, &mut query)?;
95        }
96        // Ensure local_key fields for relation loads are projected so that
97        // enhance_query_relations can match parent rows to child records.
98        if !query.relations.is_empty() {
99            if let Some(descriptor) = self.data_service.metadata.context.entity(&query.entity) {
100                for load in &query.relations {
101                    if let Some(relation) = descriptor.relation_by_name(&load.name) {
102                        if !query.projection.contains(&relation.local_key) {
103                            query.projection.push(relation.local_key.clone());
104                        }
105                    }
106                }
107            }
108        }
109        Ok(query)
110    }
111
112    pub fn prepare_insert_command(
113        &self,
114        command: &InsertCommand,
115    ) -> Result<InsertCommand, RuntimeError> {
116        let mut command = command.clone();
117        if let Some(behavior) = self.behavior() {
118            behavior.before_insert(self.data_service.metadata.context, &mut command)?;
119        }
120        self.enforce_insert_policy(&mut command)?;
121
122        let entity = self
123            .data_service
124            .metadata
125            .context
126            .require_entity(&command.entity)?;
127        if let Some(id_property) = entity.id_property() {
128            let needs_id = !command.values.contains_key(&id_property.name)
129                || is_unassigned_id(command.values.get(&id_property.name));
130            if needs_id {
131                let id = self
132                    .data_service
133                    .metadata
134                    .context
135                    .next_id(&command.entity)?;
136                command
137                    .values
138                    .insert(id_property.name.clone(), Value::U64(id));
139            }
140        }
141        ensure_initial_version(&mut command.values, entity);
142        mark_record_status(&mut command.values, CheckObjectStatus::Create);
143        let check_result = self
144            .data_service
145            .metadata
146            .context
147            .check_and_fix_record(&command.entity, &mut command.values);
148        clear_record_status(&mut command.values);
149        check_result?;
150
151        Ok(command)
152    }
153
154    pub fn update_command(&self, id: impl Into<Value>) -> UpdateCommand {
155        UpdateCommand::new(self.entity.clone(), id)
156    }
157
158    pub fn prepare_update_command(
159        &self,
160        command: &UpdateCommand,
161    ) -> Result<UpdateCommand, RuntimeError> {
162        let mut command = command.clone();
163        if let Some(behavior) = self.behavior() {
164            behavior.before_update(self.data_service.metadata.context, &mut command)?;
165        }
166        self.enforce_update_policy(&mut command)?;
167
168        Ok(command)
169    }
170
171    pub fn delete_command(&self, id: impl Into<Value>) -> DeleteCommand {
172        DeleteCommand::new(self.entity.clone(), id)
173    }
174
175    pub fn recover_command(&self, id: impl Into<Value>, expected_version: i64) -> RecoverCommand {
176        RecoverCommand::new(self.entity.clone(), id, expected_version)
177    }
178
179    pub async fn fetch_all(
180        &self,
181        query: &SelectQuery,
182    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
183        let query = self
184            .prepare_select_query(query)
185            .map_err(DataServiceError::Runtime)?;
186        self.fetch_prepared_all(&query).await
187    }
188
189    /// Fetch records in streaming mode (chunked).
190    /// Returns a Vec of chunks, each chunk containing up to `chunk_size` rows.
191    /// Each chunk is enhanced (relations, children) before returning.
192    /// Requires E to implement StreamQueryExecutor.
193    pub async fn fetch_stream(
194        &self,
195        query: &SelectQuery,
196    ) -> Result<Vec<teaql_data_service::StreamChunk>, DataServiceError<E::Error>>
197    where
198        E: teaql_data_service::StreamQueryExecutor,
199    {
200        let query = self
201            .prepare_select_query(query)
202            .map_err(DataServiceError::Runtime)?;
203
204        let chunk_size = query
205            .stream_config
206            .as_ref()
207            .map(|c| c.chunk_size)
208            .unwrap_or(1000);
209
210        let final_comment = self
211            .data_service
212            .resolve_final_comment(&query.trace_chain, query.comment.clone());
213        let mut query = query.clone();
214        query.comment = final_comment;
215
216        let request = teaql_data_service::QueryRequest {
217            query: query.clone(),
218            trace_chain: query.trace_chain.clone(),
219            comment: query.comment.clone(),
220        };
221
222        let chunks = self
223            .data_service
224            .executor
225            .query_stream(request, chunk_size)
226            .await
227            .map_err(DataServiceError::Executor)?;
228
229        // Enhance each chunk
230        let mut enhanced_chunks = Vec::with_capacity(chunks.len());
231        for mut chunk in chunks {
232            self.enhance_object_group_bys(
233                &mut chunk.rows,
234                &query.object_group_bys,
235                &query.trace_chain,
236            )
237            .await?;
238            self.enhance_child_queries(
239                &mut chunk.rows,
240                &query.child_enhancements,
241                &query.trace_chain,
242            )
243            .await?;
244            self.enhance_query_relations(&mut chunk.rows, &query)
245                .await?;
246            enhanced_chunks.push(chunk);
247        }
248
249        Ok(enhanced_chunks)
250    }
251
252    async fn fetch_prepared_all(
253        &self,
254        query: &SelectQuery,
255    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
256        let mut rows = self.fetch_prepared_query(query).await?;
257        self.enhance_object_group_bys(&mut rows, &query.object_group_bys, &query.trace_chain)
258            .await?;
259        self.enhance_child_queries(&mut rows, &query.child_enhancements, &query.trace_chain)
260            .await?;
261        self.enhance_query_relations(&mut rows, query).await?;
262        Ok(rows)
263    }
264
265    async fn fetch_prepared_query(
266        &self,
267        query: &SelectQuery,
268    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
269        let final_comment = self
270            .data_service
271            .resolve_final_comment(&query.trace_chain, query.comment.clone());
272        let mut query = query.clone();
273        query.comment = final_comment;
274        if let Some(options) = query.aggregation_cache.filter(|options| options.enabled) {
275            if let Some(cache) = self
276                .data_service
277                .metadata
278                .context
279                .get_resource::<Arc<dyn AggregationCacheBackend>>()
280            {
281                return self
282                    .fetch_prepared_query_with_cache(&query, options, cache.as_ref())
283                    .await;
284            }
285            if let Some(cache) = self
286                .data_service
287                .metadata
288                .context
289                .get_resource::<InMemoryAggregationCache>()
290            {
291                return self
292                    .fetch_prepared_query_with_cache(&query, options, cache)
293                    .await;
294            }
295        }
296        let request = teaql_data_service::QueryRequest {
297            query: query.clone(),
298            trace_chain: query.trace_chain.clone(),
299            comment: query.comment.clone(),
300        };
301        let res = self
302            .data_service
303            .executor
304            .query(request)
305            .await
306            .map_err(DataServiceError::Executor)?;
307        self.data_service
308            .metadata
309            .context
310            .record_metadata_log(&res.metadata);
311        Ok(res.rows)
312    }
313
314    async fn fetch_prepared_query_with_cache(
315        &self,
316        query: &SelectQuery,
317        options: AggregationCacheOptions,
318        cache: &dyn AggregationCacheBackend,
319    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
320        let key = aggregation_cache_key(
321            cache.namespace(),
322            &aggregation_cache_namespace(&query.entity),
323            query,
324        );
325        if let Some(rows) = cache.get(&key, options.cache_expired_millis) {
326            return Ok(rows);
327        }
328        let request = teaql_data_service::QueryRequest {
329            query: query.clone(),
330            trace_chain: query.trace_chain.clone(),
331            comment: query.comment.clone(),
332        };
333        let res = self
334            .data_service
335            .executor
336            .query(request)
337            .await
338            .map_err(DataServiceError::Executor)?;
339        self.data_service
340            .metadata
341            .context
342            .record_metadata_log(&res.metadata);
343        let rows = res.rows;
344        cache.put(key, rows.clone());
345        Ok(rows)
346    }
347
348    pub async fn fetch_all_with_relation_aggregates(
349        &self,
350        query: &SelectQuery,
351        relation_aggregates: &[RelationAggregate],
352    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
353        let query = self
354            .prepare_select_query(query)
355            .map_err(DataServiceError::Runtime)?;
356
357        let mut rows = self.fetch_prepared_all(&query).await?;
358        self.enhance_relation_aggregates(
359            &mut rows,
360            relation_aggregates,
361            query.aggregation_cache,
362            &query.trace_chain,
363        )
364        .await?;
365        Ok(rows)
366    }
367
368    pub async fn fetch_smart_list(
369        &self,
370        query: &SelectQuery,
371    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
372        let query = self
373            .prepare_select_query(query)
374            .map_err(DataServiceError::Runtime)?;
375
376        self.data_service.fetch_smart_list(&query).await
377    }
378
379    pub async fn fetch_smart_list_with_relation_aggregates(
380        &self,
381        query: &SelectQuery,
382        relation_aggregates: &[RelationAggregate],
383    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
384        self.fetch_all_with_relation_aggregates(query, relation_aggregates)
385            .await
386            .map(SmartList::from)
387    }
388
389    pub async fn fetch_entities<T>(
390        &self,
391        query: &SelectQuery,
392    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
393    where
394        T: Entity,
395    {
396        let query = self
397            .prepare_select_query(query)
398            .map_err(DataServiceError::Runtime)?;
399
400        self.data_service.fetch_entities(&query).await
401    }
402
403    pub async fn fetch_entities_with_relation_aggregates<T>(
404        &self,
405        query: &SelectQuery,
406        relation_aggregates: &[RelationAggregate],
407    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
408    where
409        T: Entity,
410    {
411        self.fetch_all_with_relation_aggregates(query, relation_aggregates)
412            .await?
413            .into_iter()
414            .map(|record| {
415                let mut entity = T::from_record(record)?;
416                let root = crate::EntityRoot::default();
417                entity.on_loaded(&root as &dyn std::any::Any);
418                Ok(entity)
419            })
420            .collect::<Result<Vec<_>, _>>()
421            .map(SmartList::from)
422            .map_err(DataServiceError::Entity)
423    }
424
425    pub async fn fetch_enhanced_entities_with_relation_aggregates<T>(
426        &self,
427        query: &SelectQuery,
428        relation_aggregates: &[RelationAggregate],
429    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
430    where
431        T: Entity,
432    {
433        let query = self
434            .prepare_select_query(query)
435            .map_err(DataServiceError::Runtime)?;
436
437        let mut rows = self.fetch_prepared_all(&query).await?;
438        self.enhance_relation_aggregates(
439            &mut rows,
440            relation_aggregates,
441            query.aggregation_cache,
442            &query.trace_chain,
443        )
444        .await?;
445        self.enhance_relations(&mut rows).await?;
446        rows.into_iter()
447            .map(|record| {
448                let mut entity = T::from_record(record)?;
449                let root = crate::EntityRoot::default();
450                entity.on_loaded(&root as &dyn std::any::Any);
451                Ok(entity)
452            })
453            .collect::<Result<Vec<_>, _>>()
454            .map(SmartList::from)
455            .map_err(DataServiceError::Entity)
456    }
457
458    pub async fn fetch_enhanced_entities<T>(
459        &self,
460        query: &SelectQuery,
461    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
462    where
463        T: Entity,
464    {
465        let query = self
466            .prepare_select_query(query)
467            .map_err(DataServiceError::Runtime)?;
468
469        let mut rows = self.fetch_prepared_all(&query).await?;
470        self.enhance_relations(&mut rows).await?;
471        let root = self
472            .data_service
473            .metadata
474            .context
475            .get_resource::<crate::EntityRoot>()
476            .cloned();
477        rows.into_iter()
478            .map(|record| {
479                let mut entity = T::from_record(record)?;
480                if let Some(ref root) = root {
481                    entity.on_loaded(root as &dyn std::any::Any);
482                }
483                Ok(entity)
484            })
485            .collect::<Result<Vec<_>, _>>()
486            .map(SmartList::from)
487            .map_err(DataServiceError::Entity)
488    }
489
490    pub async fn insert(&self, command: &InsertCommand) -> Result<u64, DataServiceError<E::Error>> {
491        let command = self
492            .prepare_insert_command(command)
493            .map_err(DataServiceError::Runtime)?;
494        self.execute_prepared_insert_with_comment(command, self.trace_context.clone())
495            .await
496    }
497
498    pub async fn update(&self, command: &UpdateCommand) -> Result<u64, DataServiceError<E::Error>> {
499        let command = self
500            .prepare_update_command(command)
501            .map_err(DataServiceError::Runtime)?;
502        self.execute_prepared_update_with_comment(command, self.trace_context.clone())
503            .await
504    }
505
506    pub async fn delete(&self, command: &DeleteCommand) -> Result<u64, DataServiceError<E::Error>> {
507        self.delete_scoped(command, self.trace_context.clone())
508            .await
509    }
510
511    pub async fn delete_scoped(
512        &self,
513        command: &DeleteCommand,
514        trace_chain: Vec<teaql_core::TraceNode>,
515    ) -> Result<u64, DataServiceError<E::Error>> {
516        let mut command = command.clone();
517        command.trace_chain = trace_chain.clone();
518        if let Some(behavior) = self.behavior() {
519            behavior
520                .before_delete(self.data_service.metadata.context, &mut command)
521                .map_err(DataServiceError::Runtime)?;
522        }
523        self.enforce_delete_policy(&mut command)
524            .map_err(DataServiceError::Runtime)?;
525
526        let old_values =
527            self.fetch_current_event_row(&command.entity, &command.id, trace_chain.clone())?;
528        let affected = self.data_service.delete(&command).await?;
529
530        let mut event = RawAuditEvent::deleted_with_old_values(
531            command.entity,
532            command.id,
533            command.expected_version,
534            old_values,
535        );
536        event.trace_chain = trace_chain;
537        self.emit_event(event).map_err(DataServiceError::Runtime)?;
538        Ok(affected)
539    }
540
541    pub async fn recover(
542        &self,
543        command: &RecoverCommand,
544    ) -> Result<u64, DataServiceError<E::Error>> {
545        let mut command = command.clone();
546        command.trace_chain = self.trace_context.clone();
547        if let Some(behavior) = self.behavior() {
548            behavior
549                .before_recover(self.data_service.metadata.context, &mut command)
550                .map_err(DataServiceError::Runtime)?;
551        }
552        self.enforce_recover_policy(&mut command)
553            .map_err(DataServiceError::Runtime)?;
554        let old_values = self.fetch_current_event_row(
555            &command.entity,
556            &command.id,
557            command.trace_chain.clone(),
558        )?;
559        let affected = self.data_service.recover(&command).await?;
560        let event = RawAuditEvent::recovered_with_old_values(
561            command.entity,
562            command.id,
563            command.expected_version,
564            old_values,
565        );
566        self.emit_event(event).map_err(DataServiceError::Runtime)?;
567        Ok(affected)
568    }
569
570    fn emit_event(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
571        self.data_service.metadata.context.send_event(event)
572    }
573
574    #[allow(dead_code)]
575    pub(super) async fn execute_prepared_insert(
576        &self,
577        command: InsertCommand,
578    ) -> Result<u64, DataServiceError<E::Error>> {
579        self.execute_prepared_insert_with_comment(command, Vec::new())
580            .await
581    }
582
583    pub(super) async fn execute_prepared_insert_with_comment(
584        &self,
585        mut command: InsertCommand,
586        trace_chain: Vec<teaql_core::TraceNode>,
587    ) -> Result<u64, DataServiceError<E::Error>> {
588        command.trace_chain = trace_chain.clone();
589        let affected = self.data_service.insert(&command).await?;
590        let mut event = RawAuditEvent::created(command.entity, command.values);
591        event.trace_chain = trace_chain;
592        self.emit_event(event).map_err(DataServiceError::Runtime)?;
593        Ok(affected)
594    }
595
596    pub(super) async fn execute_prepared_batch_insert(
597        &self,
598        command: teaql_core::BatchInsertCommand,
599    ) -> Result<u64, DataServiceError<E::Error>> {
600        if command.batch_values.is_empty() {
601            return Ok(0);
602        }
603        let affected = self.data_service.batch_insert(&command).await?;
604
605        let entity = command.entity.clone();
606        for (i, values) in command.batch_values.into_iter().enumerate() {
607            let mut event = RawAuditEvent::created(entity.clone(), values);
608            if i < command.trace_chains.len() {
609                event.trace_chain = command.trace_chains[i].clone();
610            }
611            self.emit_event(event).map_err(DataServiceError::Runtime)?;
612        }
613        Ok(affected)
614    }
615
616    #[allow(dead_code)]
617    pub(super) async fn execute_prepared_update(
618        &self,
619        command: UpdateCommand,
620    ) -> Result<u64, DataServiceError<E::Error>> {
621        self.execute_prepared_update_with_comment(command, Vec::new())
622            .await
623    }
624
625    pub(super) async fn execute_prepared_update_with_comment(
626        &self,
627        mut command: UpdateCommand,
628        trace_chain: Vec<teaql_core::TraceNode>,
629    ) -> Result<u64, DataServiceError<E::Error>> {
630        command.trace_chain = trace_chain.clone();
631
632        let mut old_values = command.old_values.clone();
633        let needs_fetch = match &old_values {
634            Some(snapshot) => !command.values.keys().all(|k| snapshot.contains_key(k)),
635            None => true,
636        };
637        if needs_fetch {
638            old_values =
639                self.fetch_current_event_row(&command.entity, &command.id, trace_chain.clone())?;
640        }
641
642        let affected = self.data_service.update(&command).await?;
643        let updated_fields = command.values.keys().cloned().collect();
644        let mut values = command.values.clone();
645        values.insert("id".to_owned(), command.id.clone());
646        if let Some(version) = command.expected_version {
647            values.insert("version".to_owned(), Value::I64(version + 1));
648        }
649        let mut new_values = old_values.clone().unwrap_or_default();
650        for (field, value) in &values {
651            new_values.insert(field.clone(), value.clone());
652        }
653        let mut event = RawAuditEvent::updated_with_old_values(
654            command.entity,
655            values,
656            old_values,
657            new_values,
658            updated_fields,
659        );
660        event.trace_chain = trace_chain;
661        self.emit_event(event).map_err(DataServiceError::Runtime)?;
662        Ok(affected)
663    }
664
665    pub(super) async fn execute_prepared_batch_update(
666        &self,
667        command: teaql_core::BatchUpdateCommand,
668    ) -> Result<u64, DataServiceError<E::Error>> {
669        if command.batch_values.is_empty() {
670            return Ok(0);
671        }
672        let affected = self.data_service.batch_update(&command).await?;
673
674        let entity = command.entity.clone();
675        for (i, values) in command.batch_values.into_iter().enumerate() {
676            let mut full_values = values.clone();
677            full_values.insert("id".to_owned(), command.batch_ids[i].clone());
678            if let Some(Some(version)) = command.batch_expected_versions.get(i) {
679                full_values.insert("version".to_owned(), teaql_core::Value::I64(*version + 1));
680            }
681
682            let old_values = command.batch_old_values.get(i).cloned().unwrap_or(None);
683            let mut new_values = old_values.clone().unwrap_or_default();
684            for (field, value) in &full_values {
685                new_values.insert(field.clone(), value.clone());
686            }
687
688            let mut event = RawAuditEvent::updated_with_old_values(
689                entity.clone(),
690                full_values,
691                old_values,
692                new_values,
693                command.update_fields.clone(),
694            );
695            if i < command.trace_chains.len() {
696                event.trace_chain = command.trace_chains[i].clone();
697            }
698            self.emit_event(event).map_err(DataServiceError::Runtime)?;
699        }
700        Ok(affected)
701    }
702
703    fn fetch_current_event_row(
704        &self,
705        _entity: &str,
706        _id: &Value,
707        _trace_chain: Vec<teaql_core::TraceNode>,
708    ) -> Result<Option<Record>, DataServiceError<E::Error>> {
709        // PER THE USER: "我们不需要在审计的时候去抓旧的值"
710        // Avoid DB queries during event emission. We rely on in-memory `original_values`.
711        Ok(None)
712    }
713
714    pub fn scoped_data_service(&self, entity: String) -> EntityDataService<'a, E> {
715        EntityDataService {
716            entity,
717            data_service: ContextDataService {
718                metadata: UserContextMetadata {
719                    context: self.data_service.metadata.context,
720                },
721                executor: self.data_service.executor,
722            },
723            trace_context: Vec::new(),
724        }
725    }
726}
teaql_runtime/data_service/resolved.rs

teaql_runtime/data_service/
resolved.rs
