dmsc/database/orm/
repository.rs

1//! Copyright © 2025-2026 Wenze Wei. All Rights Reserved.
2//!
3//! This file is part of DMSC.
4//! The DMSC project belongs to the Dunimd Team.
5//!
6//! Licensed under the Apache License, Version 2.0 (the "License");
7//! You may not use this file except in compliance with the License.
8//! You may obtain a copy of the License at
9//!
10//!     http://www.apache.org/licenses/LICENSE-2.0
11//!
12//! Unless required by applicable law or agreed to in writing, software
13//! distributed under the License is distributed on an "AS IS" BASIS,
14//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15//! See the License for the specific language governing permissions and
16//! limitations under the License.
17
18//! # ORM Repository
19//!
20//! This module provides repository implementations for ORM operations.
21
22use super::*;
23use async_trait::async_trait;
24use std::collections::HashMap;
25
26#[async_trait]
27pub trait DMSCORMRepository<E: for<'de> serde::Deserialize<'de> + serde::Serialize + Clone + Send + Sync>: Send + Sync {
28    fn table_name(&self) -> &'static str;
29    
30    async fn find_all(&self, db: &dyn DMSCDatabase) -> DMSCResult<Vec<E>>;
31    async fn find_by_id(&self, db: &dyn DMSCDatabase, id: &str) -> DMSCResult<Option<E>>;
32    async fn find_one(&self, db: &dyn DMSCDatabase, criteria: &Criteria) -> DMSCResult<Option<E>>;
33    async fn find_many(&self, db: &dyn DMSCDatabase, criteria: Vec<Criteria>) -> DMSCResult<Vec<E>>;
34    async fn find_paginated(&self, db: &dyn DMSCDatabase, pagination: Pagination, criteria: Vec<Criteria>) -> DMSCResult<(Vec<E>, u64)>;
35    async fn count(&self, db: &dyn DMSCDatabase, criteria: Vec<Criteria>) -> DMSCResult<u64>;
36    
37    async fn save(&self, db: &dyn DMSCDatabase, entity: &E) -> DMSCResult<E>;
38    async fn save_many(&self, db: &dyn DMSCDatabase, entities: &[E]) -> DMSCResult<Vec<E>>;
39    async fn update(&self, db: &dyn DMSCDatabase, entity: &E) -> DMSCResult<E>;
40    async fn delete(&self, db: &dyn DMSCDatabase, entity: &E) -> DMSCResult<()>;
41    async fn delete_by_id(&self, db: &dyn DMSCDatabase, id: &str) -> DMSCResult<()>;
42    async fn delete_many(&self, db: &dyn DMSCDatabase, criteria: Vec<Criteria>) -> DMSCResult<u64>;
43    
44    async fn exists(&self, db: &dyn DMSCDatabase, id: &str) -> DMSCResult<bool>;
45    async fn exists_by(&self, db: &dyn DMSCDatabase, criteria: &Criteria) -> DMSCResult<bool>;
46    
47    async fn batch_insert(&self, db: &dyn DMSCDatabase, entities: &[E], batch_size: usize) -> DMSCResult<Vec<E>>;
48    async fn upsert(&self, db: &dyn DMSCDatabase, entity: &E, conflict_columns: &[&str]) -> DMSCResult<E>;
49}
50
51#[async_trait]
52pub trait DMSCORMCrudRepository<E: for<'de> serde::Deserialize<'de> + serde::Serialize + Clone + Send + Sync>: Send + Sync {
53    fn table_name(&self) -> &'static str;
54    
55    async fn find_all(&self) -> DMSCResult<Vec<E>>;
56    async fn find_by_id(&self, id: &str) -> DMSCResult<Option<E>>;
57    async fn save(&self, entity: &E) -> DMSCResult<E>;
58    async fn delete(&self, entity: &E) -> DMSCResult<()>;
59}
60
61#[derive(Debug, Clone)]
62pub struct DMSCORMSimpleRepository<E: for<'de> serde::Deserialize<'de> + serde::Serialize + Clone + Send + Sync> {
63    table_name: &'static str,
64    _phantom: std::marker::PhantomData<E>,
65}
66
67impl<E: for<'de> serde::Deserialize<'de> + serde::Serialize + Clone + Send + Sync> DMSCORMSimpleRepository<E> {
68    pub fn new(table_name: &'static str) -> Self {
69        Self {
70            table_name,
71            _phantom: std::marker::PhantomData,
72        }
73    }
74}
75
76#[async_trait]
77impl<E: for<'de> serde::Deserialize<'de> + serde::Serialize + Clone + Send + Sync> DMSCORMRepository<E> for DMSCORMSimpleRepository<E> {
78    fn table_name(&self) -> &'static str {
79        self.table_name
80    }
81
82    async fn find_all(&self, db: &dyn DMSCDatabase) -> DMSCResult<Vec<E>> {
83        let sql = format!("SELECT * FROM {}", self.table_name);
84        let result = db.query(&sql).await?;
85        
86        let mut entities = Vec::new();
87        for row in result {
88            let json_value = serde_json::to_value(row.to_map())?;
89            let entity: E = serde_json::from_value(json_value)?;
90            entities.push(entity);
91        }
92        
93        Ok(entities)
94    }
95
96    async fn find_by_id(&self, db: &dyn DMSCDatabase, id: &str) -> DMSCResult<Option<E>> {
97        let sql = format!("SELECT * FROM {} WHERE id = ?", self.table_name);
98        
99        let result = db.query_with_params(&sql, &[serde_json::json!(id)]).await?;
100        
101        if let Some(row) = result.first() {
102            let json_value = serde_json::to_value(row.to_map())?;
103            let entity: E = serde_json::from_value(json_value)?;
104            Ok(Some(entity))
105        } else {
106            Ok(None)
107        }
108    }
109
110    async fn find_one(&self, db: &dyn DMSCDatabase, criteria: &Criteria) -> DMSCResult<Option<E>> {
111        let mut query = QueryBuilder::new(self.table_name);
112        query.where_criteria(criteria.clone());
113        
114        let (sql, params) = query.build();
115        let result = db.query_with_params(&sql, &params).await?;
116        
117        if let Some(row) = result.first() {
118            let json_value = serde_json::to_value(row.to_map())?;
119            let entity: E = serde_json::from_value(json_value)?;
120            Ok(Some(entity))
121        } else {
122            Ok(None)
123        }
124    }
125
126    async fn find_many(&self, db: &dyn DMSCDatabase, criteria: Vec<Criteria>) -> DMSCResult<Vec<E>> {
127        let mut query = QueryBuilder::new(self.table_name);
128        
129        for criteria in criteria {
130            query.and_where(criteria);
131        }
132        
133        let (sql, params) = query.build();
134        let result = db.query_with_params(&sql, &params).await?;
135        
136        let mut entities = Vec::new();
137        for row in result {
138            let json_value = serde_json::to_value(row.to_map())?;
139            let entity: E = serde_json::from_value(json_value)?;
140            entities.push(entity);
141        }
142        
143        Ok(entities)
144    }
145
146    async fn find_paginated(&self, db: &dyn DMSCDatabase, pagination: Pagination, criteria: Vec<Criteria>) -> DMSCResult<(Vec<E>, u64)> {
147        let count_sql = format!("SELECT COUNT(*) as total FROM {}", self.table_name);
148        
149        let total: u64 = if criteria.is_empty() {
150            if let Some(row) = db.query_one(&count_sql).await? {
151                row.get::<i64>("total").map(|v| v as u64).unwrap_or(0)
152            } else {
153                0
154            }
155        } else {
156            let mut count_query = QueryBuilder::new(self.table_name);
157            for c in &criteria {
158                count_query.and_where(c.clone());
159            }
160            let (sql, params) = count_query.build();
161            let count_sql = format!("SELECT COUNT(*) as total FROM ({}) as subquery", sql);
162            
163            let result = db.query_with_params(&count_sql, &params).await?;
164            let row = result.first().ok_or_else(|| DMSCError::Other("Query returned no rows".to_string()))?;
165            row.get_i64("total").map(|v| v as u64).unwrap_or(0)
166        };
167        
168        let mut data_query = QueryBuilder::new(self.table_name);
169        for c in criteria {
170            data_query.and_where(c);
171        }
172        data_query.paginate(pagination.page, pagination.page_size);
173        
174        let (sql, params) = data_query.build();
175        let result = db.query_with_params(&sql, &params).await?;
176        
177        let mut entities = Vec::new();
178        for row in result {
179            let json_value = serde_json::to_value(row.to_map())?;
180            let entity: E = serde_json::from_value(json_value)?;
181            entities.push(entity);
182        }
183        
184        Ok((entities, total))
185    }
186
187    async fn count(&self, db: &dyn DMSCDatabase, criteria: Vec<Criteria>) -> DMSCResult<u64> {
188        let mut query = QueryBuilder::new(self.table_name);
189        for c in criteria {
190            query.and_where(c);
191        }
192        let (sql, _) = query.build();
193        let count_sql = sql.replace("*", "COUNT(*) as total");
194        
195        if let Some(row) = db.query_one(&count_sql).await? {
196            row.get::<i64>("total").map(|v| v as u64).ok_or_else(|| DMSCError::Other("Failed to get count".to_string()))
197        } else {
198            Ok(0)
199        }
200    }
201
202    async fn save(&self, db: &dyn DMSCDatabase, entity: &E) -> DMSCResult<E> {
203        let json_value = serde_json::to_value(entity)?;
204        let values: HashMap<String, serde_json::Value> = serde_json::from_value(json_value)?;
205        
206        let columns: Vec<&str> = values.keys().map(|s| s.as_str()).collect();
207        let placeholders: Vec<String> = (0..columns.len()).map(|_| "?".to_string()).collect();
208        
209        let sql = format!(
210            "INSERT INTO {} ({}) VALUES ({})",
211            self.table_name,
212            columns.join(", "),
213            placeholders.join(", ")
214        );
215        
216        let params: Vec<serde_json::Value> = columns.iter()
217            .map(|&col| values.get(col).cloned().unwrap_or(serde_json::Value::Null))
218            .collect();
219        
220        db.execute_with_params(&sql, &params).await?;
221        
222        Ok(entity.clone())
223    }
224
225    async fn save_many(&self, db: &dyn DMSCDatabase, entities: &[E]) -> DMSCResult<Vec<E>> {
226        let mut saved = Vec::with_capacity(entities.len());
227        
228        for entity in entities {
229            saved.push(self.save(db, entity).await?);
230        }
231        
232        Ok(saved)
233    }
234
235    async fn update(&self, db: &dyn DMSCDatabase, entity: &E) -> DMSCResult<E> {
236        let json_value = serde_json::to_value(entity)?;
237        let values: HashMap<String, serde_json::Value> = serde_json::from_value(json_value)?;
238        
239        let updates: Vec<String> = values.keys()
240            .filter(|&col| col != "id")
241            .map(|col| format!("{} = ?", col))
242            .collect();
243        
244        if updates.is_empty() {
245            return Ok(entity.clone());
246        }
247        
248        let sql = format!(
249            "UPDATE {} SET {} WHERE id = ?",
250            self.table_name,
251            updates.join(", ")
252        );
253        
254        let mut params: Vec<serde_json::Value> = values.iter()
255            .filter(|(col, _)| *col != "id")
256            .map(|(_, v)| v.clone())
257            .collect();
258        
259        if let Some(id) = values.get("id") {
260            params.push(id.clone());
261        }
262        
263        db.execute_with_params(&sql, &params).await?;
264        
265        Ok(entity.clone())
266    }
267
268    async fn delete(&self, db: &dyn DMSCDatabase, entity: &E) -> DMSCResult<()> {
269        let json_value = serde_json::to_value(entity)?;
270        let values: HashMap<String, serde_json::Value> = serde_json::from_value(json_value)?;
271        
272        if let Some(id) = values.get("id") {
273            self.delete_by_id(db, &id.to_string()).await
274        } else {
275            Err(DMSCError::Other("Entity has no id field".to_string()))
276        }
277    }
278
279    async fn delete_by_id(&self, db: &dyn DMSCDatabase, id: &str) -> DMSCResult<()> {
280        let sql = format!("DELETE FROM {} WHERE id = ?", self.table_name);
281        
282        db.execute_with_params(&sql, &[serde_json::json!(id)]).await?;
283        Ok(())
284    }
285
286    async fn delete_many(&self, db: &dyn DMSCDatabase, criteria: Vec<Criteria>) -> DMSCResult<u64> {
287        if criteria.is_empty() {
288            return Err(DMSCError::Other("Criteria required for delete_many operation".to_string()));
289        }
290        
291        let mut query = QueryBuilder::new(self.table_name);
292        for c in criteria {
293            query.and_where(c);
294        }
295        
296        let (sql, params) = query.build();
297        let delete_sql = format!("DELETE FROM {}", sql.split("FROM").nth(1).unwrap_or(&sql));
298        
299        db.execute_with_params(&delete_sql, &params).await.map_err(|e| e.into())
300    }
301    
302    async fn batch_insert(&self, db: &dyn DMSCDatabase, entities: &[E], batch_size: usize) -> DMSCResult<Vec<E>> {
303        let mut inserted = Vec::with_capacity(entities.len());
304        
305        for chunk in entities.chunks(batch_size) {
306            let json_values: Vec<serde_json::Value> = chunk.iter()
307                .map(|e| serde_json::to_value(e))
308                .collect::<Result<_, _>>()?;
309            
310            let mut all_columns: std::collections::HashSet<&str> = std::collections::HashSet::new();
311            for json_value in &json_values {
312                if let serde_json::Value::Object(map) = json_value {
313                    for key in map.keys() {
314                        all_columns.insert(key);
315                    }
316                }
317            }
318            
319            let columns: Vec<&str> = all_columns.iter().copied().collect();
320            let placeholders: Vec<String> = (0..columns.len()).map(|_| "?".to_string()).collect();
321            
322            let sql = format!(
323                "INSERT INTO {} ({}) VALUES ({})",
324                self.table_name,
325                columns.join(", "),
326                placeholders.join(", ")
327            );
328            
329            for json_value in chunk {
330                let json_val = serde_json::to_value(json_value)?;
331                let values: HashMap<String, serde_json::Value> = serde_json::from_value(json_val)?;
332                
333                let params: Vec<serde_json::Value> = columns.iter()
334                    .map(|&col| values.get(col).cloned().unwrap_or(serde_json::Value::Null))
335                    .collect();
336                
337                db.execute_with_params(&sql, &params).await?;
338                inserted.push(json_value.clone());
339            }
340        }
341        
342        Ok(inserted)
343    }
344    
345    async fn upsert(&self, db: &dyn DMSCDatabase, entity: &E, conflict_columns: &[&str]) -> DMSCResult<E> {
346        let json_value = serde_json::to_value(entity)?;
347        let values: HashMap<String, serde_json::Value> = serde_json::from_value(json_value)?;
348        
349        let columns: Vec<&str> = values.keys().map(|s| s.as_str()).collect();
350        let placeholders: Vec<String> = (0..columns.len()).map(|_| "?".to_string()).collect();
351        
352        let update_parts: Vec<String> = columns.iter()
353            .filter(|&&col| !conflict_columns.contains(&col))
354            .map(|col| format!("{} = EXCLUDED.{}", col, col))
355            .collect();
356        
357        let conflict_cols = conflict_columns.join(", ");
358        let update_set = if update_parts.is_empty() {
359            String::new()
360        } else {
361            format!("ON CONFLICT ({}) DO UPDATE SET {}", conflict_cols, update_parts.join(", "))
362        };
363        
364        let sql = format!(
365            "INSERT INTO {} ({}) VALUES ({}) {}",
366            self.table_name,
367            columns.join(", "),
368            placeholders.join(", "),
369            update_set
370        );
371        
372        let params: Vec<serde_json::Value> = columns.iter()
373            .map(|&col| values.get(col).cloned().unwrap_or(serde_json::Value::Null))
374            .collect();
375        
376        db.execute_with_params(&sql, &params).await?;
377        
378        Ok(entity.clone())
379    }
380
381    async fn exists(&self, db: &dyn DMSCDatabase, id: &str) -> DMSCResult<bool> {
382        let sql = format!("SELECT 1 FROM {} WHERE id = ? LIMIT 1", self.table_name);
383        
384        let result = db.query_with_params(&sql, &[serde_json::json!(id)]).await?;
385        Ok(!result.is_empty())
386    }
387
388    async fn exists_by(&self, db: &dyn DMSCDatabase, criteria: &Criteria) -> DMSCResult<bool> {
389        let mut query = QueryBuilder::new(self.table_name);
390        query.select(vec!["1"]);
391        query.where_criteria(criteria.clone());
392        
393        let (sql, params) = query.build();
394        let sql = format!("{} LIMIT 1", sql);
395        
396        let result = db.query_with_params(&sql, &params).await?;
397        Ok(!result.is_empty())
398    }
399}
400
401impl<E: for<'de> serde::Deserialize<'de> + serde::Serialize + Clone + Send + Sync> DMSCORMSimpleRepository<E> {
402    pub fn default() -> Self {
403        Self::new("unknown")
404    }
405}
dmsc/database/orm/repository.rs

dmsc/database/orm/
repository.rs
