1use std::sync::Arc;
2use std::time::{Instant, SystemTime};
3
4use teaql_core::{
5 DeleteCommand, Entity, InsertCommand, Record, RecoverCommand, SelectQuery, SmartList,
6 UpdateCommand,
7};
8use teaql_sql::{CompiledQuery, SqlDialect};
9
10use crate::{
11 ContextError, GraphMutationPlan, GraphNode, RepositoryError, RuntimeError, SqlLogOperation,
12 UserContext,
13};
14
15use super::{
16 AggregationCacheBackend, ContextRepository, InMemoryAggregationCache, QueryExecutor,
17 Repository, ResolvedRepository, UserContextMetadata,
18 helpers::invalidate_aggregation_cache_namespace,
19};
20
21impl UserContext {
22 pub fn repository<D, E>(&self) -> Result<ContextRepository<'_, D, E>, ContextError>
23 where
24 D: SqlDialect + Send + Sync + 'static,
25 E: QueryExecutor + Send + Sync + 'static,
26 {
27 if self.metadata.is_none() {
28 return Err(ContextError::MissingResource("metadata".to_owned()));
29 }
30
31 let dialect = self.require_resource::<D>()?;
32 let executor = self.require_resource::<E>()?;
33 Ok(ContextRepository {
34 metadata: UserContextMetadata { context: self },
35 dialect,
36 executor,
37 })
38 }
39
40 #[doc(hidden)]
41 pub fn resolve_repository<D, E>(
42 &self,
43 entity: impl Into<String>,
44 ) -> Result<ResolvedRepository<'_, D, E>, ContextError>
45 where
46 D: SqlDialect + Send + Sync + 'static,
47 E: QueryExecutor + Send + Sync + 'static,
48 {
49 let entity = entity.into();
50 if !self.has_repository(&entity) {
51 return Err(ContextError::MissingRepository(entity));
52 }
53 Ok(ResolvedRepository {
54 entity,
55 repository: self.repository::<D, E>()?,
56 trace_context: Vec::new(),
57 })
58 }
59
60 pub fn plan_for_save_graph<D, E>(
61 &self,
62 node: GraphNode,
63 ) -> Result<GraphMutationPlan, RepositoryError<E::Error>>
64 where
65 D: SqlDialect + Send + Sync + 'static,
66 E: QueryExecutor + Send + Sync + 'static,
67 {
68 let repository = self
69 .resolve_repository::<D, E>(node.entity.clone())
70 .map_err(|err| RepositoryError::Runtime(RuntimeError::Graph(err.to_string())))?;
71 repository.plan_graph(node)
72 }
73}
74
75impl<'a, D, E> ContextRepository<'a, D, E>
76where
77 D: SqlDialect,
78 E: QueryExecutor,
79{
80 fn repository(&self) -> Repository<'_, D, UserContextMetadata<'_>, E> {
81 Repository::new(self.dialect, &self.metadata, self.executor)
82 }
83
84 pub fn compile(&self, query: &SelectQuery) -> Result<CompiledQuery, RuntimeError> {
85 self.repository().compile(query)
86 }
87
88 pub fn fetch_all(&self, query: &SelectQuery) -> Result<Vec<Record>, RepositoryError<E::Error>> {
89 let mut compiled = self.compile(query).map_err(RepositoryError::Runtime)?;
90 let final_comment = self.resolve_final_comment(&query.trace_chain, query.comment.clone());
91 compiled.comment = final_comment;
92
93 let started_at = SystemTime::now();
94 let started = Instant::now();
95 let rows = self
96 .executor
97 .fetch_all(&compiled)
98 .map_err(RepositoryError::Executor)?;
99 self.log_sql_result(
100 SqlLogOperation::Select,
101 &compiled,
102 started_at,
103 started,
104 Some(rows.len()),
105 Some(query.entity.clone()),
106 None,
107 query.trace_chain.clone(),
108 );
109 Ok(rows)
110 }
111
112 pub fn fetch_smart_list(
113 &self,
114 query: &SelectQuery,
115 ) -> Result<SmartList<Record>, RepositoryError<E::Error>> {
116 self.repository().fetch_smart_list(query)
117 }
118
119 pub fn fetch_entities<T>(
120 &self,
121 query: &SelectQuery,
122 ) -> Result<SmartList<T>, RepositoryError<E::Error>>
123 where
124 T: Entity,
125 {
126 self.repository().fetch_entities(query)
127 }
128
129 pub fn fetch_enhanced_entities<T>(
130 &self,
131 query: &SelectQuery,
132 ) -> Result<SmartList<T>, RepositoryError<E::Error>>
133 where
134 T: Entity,
135 {
136 self.repository().fetch_enhanced_entities(query)
137 }
138
139 pub fn insert(&self, command: &InsertCommand) -> Result<u64, RepositoryError<E::Error>> {
140 let mut compiled = self
141 .repository()
142 .compile_insert(command)
143 .map_err(RepositoryError::Runtime)?;
144 let final_comment = self.resolve_final_comment(&command.trace_chain, None);
145 compiled.comment = final_comment;
146
147 let started_at = SystemTime::now();
148 let started = Instant::now();
149 let affected = self
150 .executor
151 .execute(&compiled)
152 .map_err(RepositoryError::Executor)?;
153 self.log_sql_result(
154 SqlLogOperation::Insert,
155 &compiled,
156 started_at,
157 started,
158 None,
159 None,
160 Some(affected),
161 command.trace_chain.clone(),
162 );
163 self.invalidate_aggregation_cache_for(&command.entity);
164 Ok(affected)
165 }
166
167 pub fn update(&self, command: &UpdateCommand) -> Result<u64, RepositoryError<E::Error>> {
168 let affected = self.execute_mutation(
169 SqlLogOperation::Update,
170 &command.entity,
171 self.repository()
172 .compile_update(command)
173 .map_err(RepositoryError::Runtime)?,
174 command.trace_chain.clone(),
175 )?;
176 if command.expected_version.is_some() && affected == 0 {
177 return Err(RepositoryError::Runtime(
178 RuntimeError::OptimisticLockConflict {
179 entity: command.entity.clone(),
180 id: format!("{:?}", command.id),
181 },
182 ));
183 }
184 Ok(affected)
185 }
186
187 pub fn delete(&self, command: &DeleteCommand) -> Result<u64, RepositoryError<E::Error>> {
188 let affected = self.execute_mutation(
189 SqlLogOperation::Delete,
190 &command.entity,
191 self.repository()
192 .compile_delete(command)
193 .map_err(RepositoryError::Runtime)?,
194 command.trace_chain.clone(),
195 )?;
196 if command.expected_version.is_some() && affected == 0 {
197 return Err(RepositoryError::Runtime(
198 RuntimeError::OptimisticLockConflict {
199 entity: command.entity.clone(),
200 id: format!("{:?}", command.id),
201 },
202 ));
203 }
204 Ok(affected)
205 }
206
207 pub fn recover(&self, command: &RecoverCommand) -> Result<u64, RepositoryError<E::Error>> {
208 let affected = self.execute_mutation(
209 SqlLogOperation::Recover,
210 &command.entity,
211 self.repository()
212 .compile_recover(command)
213 .map_err(RepositoryError::Runtime)?,
214 command.trace_chain.clone(),
215 )?;
216 if affected == 0 {
217 return Err(RepositoryError::Runtime(
218 RuntimeError::OptimisticLockConflict {
219 entity: command.entity.clone(),
220 id: format!("{:?}", command.id),
221 },
222 ));
223 }
224 Ok(affected)
225 }
226
227 fn execute_mutation(
228 &self,
229 operation: SqlLogOperation,
230 entity: &str,
231 mut compiled: CompiledQuery,
232 trace_chain: Vec<teaql_core::TraceNode>,
233 ) -> Result<u64, RepositoryError<E::Error>> {
234 let final_comment = self.resolve_final_comment(&trace_chain, None);
235 compiled.comment = final_comment;
236
237 let started_at = SystemTime::now();
238 let started = Instant::now();
239 let affected = self
240 .executor
241 .execute(&compiled)
242 .map_err(RepositoryError::Executor)?;
243 self.log_sql_result(
244 operation,
245 &compiled,
246 started_at,
247 started,
248 None,
249 None,
250 Some(affected),
251 trace_chain,
252 );
253 self.invalidate_aggregation_cache_for(entity);
254 Ok(affected)
255 }
256
257 pub(super) fn log_sql_result(
258 &self,
259 operation: SqlLogOperation,
260 compiled: &CompiledQuery,
261 started_at: SystemTime,
262 started: Instant,
263 result_count: Option<usize>,
264 result_type: Option<String>,
265 affected_rows: Option<u64>,
266 trace_chain: Vec<teaql_core::TraceNode>,
267 ) {
268 self.metadata.context.record_sql_log(
269 operation,
270 compiled,
271 self.dialect.kind(),
272 started_at,
273 SystemTime::now(),
274 started.elapsed(),
275 result_count,
276 result_type,
277 affected_rows,
278 trace_chain,
279 );
280 }
281
282 pub(super) fn invalidate_aggregation_cache_for(&self, entity: &str) {
283 if let Some(cache) = self
284 .metadata
285 .context
286 .get_resource::<Arc<dyn AggregationCacheBackend>>()
287 {
288 invalidate_aggregation_cache_namespace(cache.as_ref(), entity);
289 }
290 if let Some(cache) = self
291 .metadata
292 .context
293 .get_resource::<InMemoryAggregationCache>()
294 {
295 invalidate_aggregation_cache_namespace(cache, entity);
296 }
297 }
298
299 pub(crate) fn resolve_final_comment(&self, trace_chain: &[teaql_core::TraceNode], comment: Option<String>) -> Option<String> {
300 let chain_str = if trace_chain.is_empty() {
301 None
302 } else {
303 let formatted = trace_chain.iter().map(|n| {
304 format!("{}({}): {}", n.entity_type, n.entity_id.map(|id| id.to_string()).unwrap_or_else(|| "pending".to_owned()), n.comment)
305 }).collect::<Vec<_>>().join(" -> ");
306 Some(formatted)
307 };
308
309 let business_comment = chain_str.or(comment);
310 let user_id = self.metadata.context.user_identifier().map(|s| s.to_owned());
311
312 match (user_id, business_comment) {
313 (Some(user), Some(bus)) if !user.is_empty() && !bus.is_empty() => {
314 Some(format!("[{user}] {bus}"))
315 }
316 (Some(user), _) if !user.is_empty() => {
317 Some(format!("[{user}]"))
318 }
319 (_, Some(bus)) if !bus.is_empty() => {
320 Some(bus)
321 }
322 _ => None,
323 }
324 }
325}