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
60impl<'a, E> ContextDataService<'a, E>
61where
62    E: teaql_data_service::QueryExecutor
63        + teaql_data_service::MutationExecutor
64        + Send
65        + Sync
66        + 'static,
67{
68    fn data_service(&self) -> RuntimeDataService<'_, UserContextMetadata<'_>, E> {
69        RuntimeDataService::new(&self.metadata, self.executor)
70    }
71
72    pub async fn fetch_all(
73        &self,
74        mut query: SelectQuery,
75    ) -> Result<Vec<Record>, DataServiceError<E::Error>> {
76        let final_comment = self.resolve_final_comment(&query.trace_chain, query.comment.clone());
77        query.comment = final_comment;
78        self.data_service().fetch_all(&query).await
79    }
80
81    pub async fn fetch_smart_list(
82        &self,
83        query: &SelectQuery,
84    ) -> Result<SmartList<Record>, DataServiceError<E::Error>> {
85        self.data_service().fetch_smart_list(query).await
86    }
87
88    pub async fn fetch_entities<T>(
89        &self,
90        query: &SelectQuery,
91    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
92    where
93        T: Entity,
94    {
95        self.data_service().fetch_entities(query).await
96    }
97
98    pub async fn fetch_enhanced_entities<T>(
99        &self,
100        query: &SelectQuery,
101    ) -> Result<SmartList<T>, DataServiceError<E::Error>>
102    where
103        T: Entity,
104    {
105        self.data_service().fetch_enhanced_entities(query).await
106    }
107
108    pub async fn insert(&self, command: &InsertCommand) -> Result<u64, DataServiceError<E::Error>> {
109        let affected = self.data_service().insert(command).await?;
110        self.invalidate_aggregation_cache_for(&command.entity);
111        Ok(affected)
112    }
113
114    pub async fn update(&self, command: &UpdateCommand) -> Result<u64, DataServiceError<E::Error>> {
115        let affected = self.data_service().update(command).await?;
116        self.invalidate_aggregation_cache_for(&command.entity);
117        Ok(affected)
118    }
119
120    pub async fn batch_insert(
121        &self,
122        command: &teaql_core::BatchInsertCommand,
123    ) -> Result<u64, DataServiceError<E::Error>> {
124        let affected = self.data_service().batch_insert(command).await?;
125        self.invalidate_aggregation_cache_for(&command.entity);
126        Ok(affected)
127    }
128
129    pub async fn batch_update(
130        &self,
131        command: &teaql_core::BatchUpdateCommand,
132    ) -> Result<u64, DataServiceError<E::Error>> {
133        let affected = self.data_service().batch_update(command).await?;
134        self.invalidate_aggregation_cache_for(&command.entity);
135        Ok(affected)
136    }
137
138    pub async fn delete(&self, command: &DeleteCommand) -> Result<u64, DataServiceError<E::Error>> {
139        let affected = self.data_service().delete(command).await?;
140        self.invalidate_aggregation_cache_for(&command.entity);
141        Ok(affected)
142    }
143
144    pub async fn recover(
145        &self,
146        command: &RecoverCommand,
147    ) -> Result<u64, DataServiceError<E::Error>> {
148        let affected = self.data_service().recover(command).await?;
149        self.invalidate_aggregation_cache_for(&command.entity);
150        Ok(affected)
151    }
152
153    pub(super) fn invalidate_aggregation_cache_for(&self, entity: &str) {
154        if let Some(cache) = self
155            .metadata
156            .context
157            .get_resource::<Arc<dyn AggregationCacheBackend>>()
158        {
159            invalidate_aggregation_cache_namespace(cache.as_ref(), entity);
160        }
161        if let Some(cache) = self
162            .metadata
163            .context
164            .get_resource::<InMemoryAggregationCache>()
165        {
166            invalidate_aggregation_cache_namespace(cache, entity);
167        }
168    }
169
170    pub(crate) fn resolve_final_comment(
171        &self,
172        trace_chain: &[teaql_core::TraceNode],
173        comment: Option<String>,
174    ) -> Option<String> {
175        let chain_str = if trace_chain.is_empty() {
176            None
177        } else {
178            let formatted = trace_chain
179                .iter()
180                .map(|n| {
181                    format!(
182                        "{}({}): {}",
183                        n.entity_type,
184                        n.entity_id
185                            .map(|id| id.to_string())
186                            .unwrap_or_else(|| "pending".to_owned()),
187                        n.comment
188                    )
189                })
190                .collect::<Vec<_>>()
191                .join(" -> ");
192            Some(formatted)
193        };
194
195        let business_comment = chain_str.or(comment);
196        let user_id = self
197            .metadata
198            .context
199            .user_identifier()
200            .map(|s| s.to_owned());
201
202        match (user_id, business_comment) {
203            (Some(user), Some(bus)) if !user.is_empty() && !bus.is_empty() => {
204                Some(format!("[{user}] {bus}"))
205            }
206            (Some(user), _) if !user.is_empty() => Some(format!("[{user}]")),
207            (_, Some(bus)) if !bus.is_empty() => Some(bus),
208            _ => None,
209        }
210    }
211}
teaql_runtime/data_service/context.rs

teaql_runtime/data_service/
context.rs
