ri/database/orm/
repository.rs

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

ri/database/orm/
repository.rs
