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 fn data_service<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::<E>()?,
55            trace_context: Vec::new(),
56        })
57    }
58
59    /// Register a data-service executor and automatically set up the
60    /// type-erased [`DynGraphSaver`](crate::DynGraphSaver) 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::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 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 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 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 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 async fn insert(&self, command: &InsertCommand) -> Result<u64, DataServiceError<E::Error>> {
127        let affected = self.data_service().insert(command).await?;
128        self.invalidate_aggregation_cache_for(&command.entity);
129        Ok(affected)
130    }
131
132    pub async fn update(&self, command: &UpdateCommand) -> Result<u64, DataServiceError<E::Error>> {
133        let affected = self.data_service().update(command).await?;
134        self.invalidate_aggregation_cache_for(&command.entity);
135        Ok(affected)
136    }
137
138    pub async fn batch_insert(
139        &self,
140        command: &teaql_core::BatchInsertCommand,
141    ) -> Result<u64, DataServiceError<E::Error>> {
142        let affected = self.data_service().batch_insert(command).await?;
143        self.invalidate_aggregation_cache_for(&command.entity);
144        Ok(affected)
145    }
146
147    pub async fn batch_update(
148        &self,
149        command: &teaql_core::BatchUpdateCommand,
150    ) -> Result<u64, DataServiceError<E::Error>> {
151        let affected = self.data_service().batch_update(command).await?;
152        self.invalidate_aggregation_cache_for(&command.entity);
153        Ok(affected)
154    }
155
156    pub async fn delete(&self, command: &DeleteCommand) -> Result<u64, DataServiceError<E::Error>> {
157        let affected = self.data_service().delete(command).await?;
158        self.invalidate_aggregation_cache_for(&command.entity);
159        Ok(affected)
160    }
161
162    pub async fn recover(
163        &self,
164        command: &RecoverCommand,
165    ) -> Result<u64, DataServiceError<E::Error>> {
166        let affected = self.data_service().recover(command).await?;
167        self.invalidate_aggregation_cache_for(&command.entity);
168        Ok(affected)
169    }
170
171    pub(super) fn invalidate_aggregation_cache_for(&self, entity: &str) {
172        if let Some(cache) = self
173            .metadata
174            .context
175            .get_resource::<Arc<dyn AggregationCacheBackend>>()
176        {
177            invalidate_aggregation_cache_namespace(cache.as_ref(), entity);
178        }
179        if let Some(cache) = self
180            .metadata
181            .context
182            .get_resource::<InMemoryAggregationCache>()
183        {
184            invalidate_aggregation_cache_namespace(cache, entity);
185        }
186    }
187
188    pub(crate) fn resolve_final_comment(
189        &self,
190        trace_chain: &[teaql_core::TraceNode],
191        comment: Option<String>,
192    ) -> Option<String> {
193        let chain_str = if trace_chain.is_empty() {
194            None
195        } else {
196            let formatted = trace_chain
197                .iter()
198                .map(|n| {
199                    format!(
200                        "{}({}): {}",
201                        n.entity_type,
202                        n.entity_id
203                            .map(|id| id.to_string())
204                            .unwrap_or_else(|| "pending".to_owned()),
205                        n.comment
206                    )
207                })
208                .collect::<Vec<_>>()
209                .join(" -> ");
210            Some(formatted)
211        };
212
213        let business_comment = chain_str.or(comment);
214        let user_id = self
215            .metadata
216            .context
217            .user_identifier()
218            .map(|s| s.to_owned());
219
220        match (user_id, business_comment) {
221            (Some(user), Some(bus)) if !user.is_empty() && !bus.is_empty() => {
222                Some(format!("[{user}] {bus}"))
223            }
224            (Some(user), _) if !user.is_empty() => Some(format!("[{user}]")),
225            (_, Some(bus)) if !bus.is_empty() => Some(bus),
226            _ => None,
227        }
228    }
229}
teaql_runtime/data_service/context.rs

teaql_runtime/data_service/
context.rs
