teaql_runtime/data_service/
context.rs

1use std::fmt::Write;
2use std::sync::Arc;
3use std::{collections::BTreeMap, future::Future};
4
5use teaql_core::{
6    DeleteCommand, Entity, InsertCommand, Record, RecoverCommand, SelectQuery, SmartList,
7    UpdateCommand,
8};
9
10use crate::{
11    ContextError, DataServiceError, GraphMutationPlan, GraphNode, RuntimeError, UserContext,
12};
13
14use super::{
15    AggregationCacheBackend, ContextDataService, EntityDataService, InMemoryAggregationCache,
16    RuntimeDataService, UserContextMetadata, helpers::invalidate_aggregation_cache_namespace,
17};
18
19impl UserContext {
20    pub(crate) fn data_service_internal<E>(&self) -> Result<ContextDataService<'_, E>, ContextError>
21    where
22        E: teaql_data_service::QueryExecutor
23            + teaql_data_service::MutationExecutor
24            + Send
25            + Sync
26            + 'static,
27    {
28        if self.metadata.is_none() {
29            return Err(ContextError::MissingResource("metadata".to_owned()));
30        }
31
32        let executor = self.require_resource::<E>()?;
33        Ok(ContextDataService {
34            metadata: UserContextMetadata { context: self },
35            executor,
36        })
37    }
38
39    pub fn entity_data_service<E>(
40        &self,
41        entity: impl Into<String>,
42    ) -> Result<EntityDataService<'_, E>, ContextError>
43    where
44        E: teaql_data_service::QueryExecutor
45            + teaql_data_service::MutationExecutor
46            + Send
47            + Sync
48            + 'static,
49    {
50        let entity = entity.into();
51        if !self.has_entity_data_service(&entity) {
52            return Err(ContextError::MissingEntityDataService(entity));
53        }
54        Ok(EntityDataService {
55            entity,
56            data_service: self.data_service_internal::<E>()?,
57            trace_context: Vec::new(),
58        })
59    }
60
61    /// Register a data-service executor and automatically set up the
62    /// type-erased graph saver so that
63    /// [`Audited::save`](crate::AuditedSaveExt::save) works.
64    pub fn register_executor<E>(&mut self, executor: E)
65    where
66        E: teaql_data_service::QueryExecutor
67            + teaql_data_service::MutationExecutor
68            + Send
69            + Sync
70            + 'static,
71    {
72        use std::sync::Arc;
73        self.insert_resource::<Arc<dyn crate::entity_save::DynGraphSaver>>(Arc::new(
74            crate::entity_save::GraphSaverFor::<E>::new(),
75        ));
76        self.insert_resource(executor);
77    }
78}
79
80impl<'a, E> ContextDataService<'a, E>
81where
82    E: teaql_data_service::QueryExecutor
83        + teaql_data_service::MutationExecutor
84        + Send
85        + Sync
86        + 'static,
87{
88    async fn observe<T, F>(
89        &self,
90        family: &str,
91        name: String,
92        entity: &str,
93        work: F,
94    ) -> Result<T, DataServiceError<E::Error>>
95    where
96        F: Future<Output = Result<T, DataServiceError<E::Error>>>,
97    {
98        let operation = crate::RuntimeOperation::new(family, name)
99            .attribute("teaql.entity.type", entity.to_owned());
100        let scope = self.metadata.context.start_runtime_operation(operation);
101        let provider_kind = std::any::type_name::<E>().to_owned();
102        let provider_operation = family.to_owned();
103        let result = scope
104            .run(async {
105                let provider_scope = self.metadata.context.start_runtime_operation(
106                    crate::RuntimeOperation::new(
107                        "provider",
108                        format!("{provider_kind}.{provider_operation}"),
109                    )
110                    .attribute("teaql.provider.kind", provider_kind)
111                    .attribute("teaql.provider.operation", provider_operation),
112                );
113                let result = provider_scope.run(work).await;
114                match &result {
115                    Ok(_) => provider_scope.success(BTreeMap::new()),
116                    Err(_) => provider_scope.failure("data_service_error"),
117                }
118                result
119            })
120            .await;
121        match result {
122            Ok(value) => {
123                scope.success(BTreeMap::new());
124                Ok(value)
125            }
126            Err(error) => {
127                scope.failure("data_service_error");
128                Err(error)
129            }
130        }
131    }
132
133    fn data_service(&self) -> RuntimeDataService<'_, UserContextMetadata<'_>, E> {
134        RuntimeDataService::new(&self.metadata, self.executor)
135    }
136
137    pub(crate) async fn fetch_all(
138        &self,
139        mut query: SelectQuery,
140    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
141        let final_comment = self.resolve_final_comment(&query.trace_chain, query.comment.clone());
142        query.comment = final_comment;
143        self.observe(
144            "query",
145            format!("{}.list", query.entity),
146            &query.entity,
147            self.data_service().fetch_all(&query),
148        )
149        .await
150    }
151
152    pub(crate) async fn fetch_smart_list(
153        &self,
154        query: &SelectQuery,
155    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
156        self.observe(
157            "query",
158            format!("{}.list", query.entity),
159            &query.entity,
160            self.data_service().fetch_smart_list(query),
161        )
162        .await
163    }
164
165    pub(crate) async fn fetch_entities<T>(
166        &self,
167        query: &SelectQuery,
168    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
169    where
170        T: Entity,
171    {
172        self.observe(
173            "query",
174            format!("{}.list", query.entity),
175            &query.entity,
176            self.data_service().fetch_entities(query),
177        )
178        .await
179    }
180
181    pub(crate) async fn fetch_enhanced_entities<T>(
182        &self,
183        query: &SelectQuery,
184    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
185    where
186        T: Entity,
187    {
188        self.observe(
189            "query",
190            format!("{}.list", query.entity),
191            &query.entity,
192            self.data_service().fetch_enhanced_entities(query),
193        )
194        .await
195    }
196
197    pub(crate) async fn insert(
198        &self,
199        command: &InsertCommand,
200    ) -> Result<u64, DataServiceError<E::Error>> {
201        let affected = self
202            .observe(
203                "mutation",
204                format!("{}.insert", command.entity),
205                &command.entity,
206                self.data_service().insert(command),
207            )
208            .await?;
209        self.invalidate_aggregation_cache_for(&command.entity);
210        Ok(affected)
211    }
212
213    pub(crate) async fn update(
214        &self,
215        command: &UpdateCommand,
216    ) -> Result<u64, DataServiceError<E::Error>> {
217        let affected = self
218            .observe(
219                "mutation",
220                format!("{}.update", command.entity),
221                &command.entity,
222                self.data_service().update(command),
223            )
224            .await?;
225        self.invalidate_aggregation_cache_for(&command.entity);
226        Ok(affected)
227    }
228
229    pub(crate) async fn batch_insert(
230        &self,
231        command: &teaql_core::BatchInsertCommand,
232    ) -> Result<u64, DataServiceError<E::Error>> {
233        let affected = self
234            .observe(
235                "mutation",
236                format!("{}.batch_insert", command.entity),
237                &command.entity,
238                self.data_service().batch_insert(command),
239            )
240            .await?;
241        self.invalidate_aggregation_cache_for(&command.entity);
242        Ok(affected)
243    }
244
245    pub(crate) async fn batch_update(
246        &self,
247        command: &teaql_core::BatchUpdateCommand,
248    ) -> Result<u64, DataServiceError<E::Error>> {
249        let affected = self
250            .observe(
251                "mutation",
252                format!("{}.batch_update", command.entity),
253                &command.entity,
254                self.data_service().batch_update(command),
255            )
256            .await?;
257        self.invalidate_aggregation_cache_for(&command.entity);
258        Ok(affected)
259    }
260
261    pub(crate) async fn delete(
262        &self,
263        command: &DeleteCommand,
264    ) -> Result<u64, DataServiceError<E::Error>> {
265        let affected = self
266            .observe(
267                "mutation",
268                format!("{}.delete", command.entity),
269                &command.entity,
270                self.data_service().delete(command),
271            )
272            .await?;
273        self.invalidate_aggregation_cache_for(&command.entity);
274        Ok(affected)
275    }
276
277    pub(crate) async fn recover(
278        &self,
279        command: &RecoverCommand,
280    ) -> Result<u64, DataServiceError<E::Error>> {
281        let affected = self
282            .observe(
283                "mutation",
284                format!("{}.recover", command.entity),
285                &command.entity,
286                self.data_service().recover(command),
287            )
288            .await?;
289        self.invalidate_aggregation_cache_for(&command.entity);
290        Ok(affected)
291    }
292
293    pub(super) fn invalidate_aggregation_cache_for(&self, entity: &str) {
294        if let Some(cache) = self
295            .metadata
296            .context
297            .get_resource::<Arc<dyn AggregationCacheBackend>>()
298        {
299            invalidate_aggregation_cache_namespace(cache.as_ref(), entity);
300        }
301        if let Some(cache) = self
302            .metadata
303            .context
304            .get_resource::<InMemoryAggregationCache>()
305        {
306            invalidate_aggregation_cache_namespace(cache, entity);
307        }
308    }
309
310    pub(crate) fn resolve_final_comment(
311        &self,
312        trace_chain: &[teaql_core::TraceNode],
313        comment: Option<String>,
314    ) -> Option<String> {
315        let chain_str = (!trace_chain.is_empty()).then(|| {
316            let mut chain = String::with_capacity(trace_chain.len().saturating_mul(64));
317            for (index, node) in trace_chain.iter().enumerate() {
318                if index > 0 {
319                    chain.push_str(" -> ");
320                }
321                match node.entity_id {
322                    Some(id) => {
323                        let _ = write!(chain, "{}({id}): {}", node.entity_type, node.comment);
324                    }
325                    None => {
326                        let _ = write!(
327                            chain,
328                            "{}(pending): {}",
329                            node.entity_type, node.comment
330                        );
331                    }
332                }
333            }
334            chain
335        });
336
337        let business_comment = chain_str.or(comment);
338        let user_id = self
339            .metadata
340            .context
341            .user_identifier()
342            .map(|s| s.to_owned());
343
344        match (user_id, business_comment) {
345            (Some(user), Some(bus)) if !user.is_empty() && !bus.is_empty() => {
346                Some(format!("[{user}] {bus}"))
347            }
348            (Some(user), _) if !user.is_empty() => Some(format!("[{user}]")),
349            (_, Some(bus)) if !bus.is_empty() => Some(bus),
350            _ => None,
351        }
352    }
353}
teaql_runtime/data_service/context.rs

teaql_runtime/data_service/
context.rs
