teaql_runtime/data_service/
context.rs

1use std::sync::Arc;
2
3use teaql_core::{
4    DeleteCommand, Entity, InsertCommand, Record, RecoverCommand, SelectQuery, SmartList,
5    UpdateCommand,
6};
7
8use crate::{
9    ContextError, DataServiceError, GraphMutationPlan, GraphNode, RuntimeError, UserContext,
10};
11
12use super::{
13    AggregationCacheBackend, ContextDataService, EntityDataService, InMemoryAggregationCache,
14    RuntimeDataService, UserContextMetadata, helpers::invalidate_aggregation_cache_namespace,
15};
16
17impl UserContext {
18    pub(crate) fn data_service_internal<E>(&self) -> Result<ContextDataService<'_, E>, ContextError>
19    where
20        E: teaql_data_service::QueryExecutor
21            + teaql_data_service::MutationExecutor
22            + Send
23            + Sync
24            + 'static,
25    {
26        if self.metadata.is_none() {
27            return Err(ContextError::MissingResource("metadata".to_owned()));
28        }
29
30        let executor = self.require_resource::<E>()?;
31        Ok(ContextDataService {
32            metadata: UserContextMetadata { context: self },
33            executor,
34        })
35    }
36
37    pub fn entity_data_service<E>(
38        &self,
39        entity: impl Into<String>,
40    ) -> Result<EntityDataService<'_, E>, ContextError>
41    where
42        E: teaql_data_service::QueryExecutor
43            + teaql_data_service::MutationExecutor
44            + Send
45            + Sync
46            + 'static,
47    {
48        let entity = entity.into();
49        if !self.has_entity_data_service(&entity) {
50            return Err(ContextError::MissingEntityDataService(entity));
51        }
52        Ok(EntityDataService {
53            entity,
54            data_service: self.data_service_internal::<E>()?,
55            trace_context: Vec::new(),
56        })
57    }
58
59    /// Register a data-service executor and automatically set up the
60    /// type-erased graph saver so that
61    /// [`Audited::save`](crate::AuditedSaveExt::save) works.
62    pub fn register_executor<E>(&mut self, executor: E)
63    where
64        E: teaql_data_service::QueryExecutor
65            + teaql_data_service::MutationExecutor
66            + Send
67            + Sync
68            + 'static,
69    {
70        use std::sync::Arc;
71        self.insert_resource::<Arc<dyn crate::entity_save::DynGraphSaver>>(Arc::new(
72            crate::entity_save::GraphSaverFor::<E>::new(),
73        ));
74        self.insert_resource(executor);
75    }
76}
77
78impl<'a, E> ContextDataService<'a, E>
79where
80    E: teaql_data_service::QueryExecutor
81        + teaql_data_service::MutationExecutor
82        + Send
83        + Sync
84        + 'static,
85{
86    fn data_service(&self) -> RuntimeDataService<'_, UserContextMetadata<'_>, E> {
87        RuntimeDataService::new(&self.metadata, self.executor)
88    }
89
90    pub(crate) async fn fetch_all(
91        &self,
92        mut query: SelectQuery,
93    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
94        let final_comment = self.resolve_final_comment(&query.trace_chain, query.comment.clone());
95        query.comment = final_comment;
96        self.data_service().fetch_all(&query).await
97    }
98
99    pub(crate) async fn fetch_smart_list(
100        &self,
101        query: &SelectQuery,
102    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
103        self.data_service().fetch_smart_list(query).await
104    }
105
106    pub(crate) async fn fetch_entities<T>(
107        &self,
108        query: &SelectQuery,
109    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
110    where
111        T: Entity,
112    {
113        self.data_service().fetch_entities(query).await
114    }
115
116    pub(crate) async fn fetch_enhanced_entities<T>(
117        &self,
118        query: &SelectQuery,
119    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
120    where
121        T: Entity,
122    {
123        self.data_service().fetch_enhanced_entities(query).await
124    }
125
126    pub(crate) async fn insert(
127        &self,
128        command: &InsertCommand,
129    ) -> Result<u64, DataServiceError<E::Error>> {
130        let affected = self.data_service().insert(command).await?;
131        self.invalidate_aggregation_cache_for(&command.entity);
132        Ok(affected)
133    }
134
135    pub(crate) async fn update(
136        &self,
137        command: &UpdateCommand,
138    ) -> Result<u64, DataServiceError<E::Error>> {
139        let affected = self.data_service().update(command).await?;
140        self.invalidate_aggregation_cache_for(&command.entity);
141        Ok(affected)
142    }
143
144    pub(crate) async fn batch_insert(
145        &self,
146        command: &teaql_core::BatchInsertCommand,
147    ) -> Result<u64, DataServiceError<E::Error>> {
148        let affected = self.data_service().batch_insert(command).await?;
149        self.invalidate_aggregation_cache_for(&command.entity);
150        Ok(affected)
151    }
152
153    pub(crate) async fn batch_update(
154        &self,
155        command: &teaql_core::BatchUpdateCommand,
156    ) -> Result<u64, DataServiceError<E::Error>> {
157        let affected = self.data_service().batch_update(command).await?;
158        self.invalidate_aggregation_cache_for(&command.entity);
159        Ok(affected)
160    }
161
162    pub(crate) async fn delete(
163        &self,
164        command: &DeleteCommand,
165    ) -> Result<u64, DataServiceError<E::Error>> {
166        let affected = self.data_service().delete(command).await?;
167        self.invalidate_aggregation_cache_for(&command.entity);
168        Ok(affected)
169    }
170
171    pub(crate) async fn recover(
172        &self,
173        command: &RecoverCommand,
174    ) -> Result<u64, DataServiceError<E::Error>> {
175        let affected = self.data_service().recover(command).await?;
176        self.invalidate_aggregation_cache_for(&command.entity);
177        Ok(affected)
178    }
179
180    pub(super) fn invalidate_aggregation_cache_for(&self, entity: &str) {
181        if let Some(cache) = self
182            .metadata
183            .context
184            .get_resource::<Arc<dyn AggregationCacheBackend>>()
185        {
186            invalidate_aggregation_cache_namespace(cache.as_ref(), entity);
187        }
188        if let Some(cache) = self
189            .metadata
190            .context
191            .get_resource::<InMemoryAggregationCache>()
192        {
193            invalidate_aggregation_cache_namespace(cache, entity);
194        }
195    }
196
197    pub(crate) fn resolve_final_comment(
198        &self,
199        trace_chain: &[teaql_core::TraceNode],
200        comment: Option<String>,
201    ) -> Option<String> {
202        let chain_str = (!trace_chain.is_empty()).then(|| {
203            trace_chain
204                .iter()
205                .map(|n| {
206                    format!(
207                        "{}({}): {}",
208                        n.entity_type,
209                        n.entity_id
210                            .map(|id| id.to_string())
211                            .unwrap_or_else(|| "pending".to_owned()),
212                        n.comment
213                    )
214                })
215                .collect::<Vec<_>>()
216                .join(" -> ")
217        });
218
219        let business_comment = chain_str.or(comment);
220        let user_id = self
221            .metadata
222            .context
223            .user_identifier()
224            .map(|s| s.to_owned());
225
226        match (user_id, business_comment) {
227            (Some(user), Some(bus)) if !user.is_empty() && !bus.is_empty() => {
228                Some(format!("[{user}] {bus}"))
229            }
230            (Some(user), _) if !user.is_empty() => Some(format!("[{user}]")),
231            (_, Some(bus)) if !bus.is_empty() => Some(bus),
232            _ => None,
233        }
234    }
235}
teaql_runtime/data_service/context.rs

teaql_runtime/data_service/
context.rs
