teaql_runtime/data_service/
context.rs

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

teaql_runtime/data_service/
context.rs
