1use std::collections::HashMap;
7
8use rusqlite::types::Value;
9use serde_json::Value as JsonValue;
10use tracing::{debug, instrument, trace};
11
12use prax_query::filter::FilterValue;
13use prax_query::types::SortOrder;
14
15use crate::error::SqliteError;
16use crate::pool::SqlitePool;
17use crate::types::filter_value_to_sqlite;
18
19#[derive(Clone)]
21pub struct SqliteRawEngine {
22 pool: SqlitePool,
23}
24
25#[derive(Debug, Clone)]
27pub struct SqliteJsonRow {
28 pub data: JsonValue,
30}
31
32impl SqliteJsonRow {
33 pub fn new(data: JsonValue) -> Self {
35 Self { data }
36 }
37
38 pub fn json(&self) -> &JsonValue {
40 &self.data
41 }
42
43 pub fn into_json(self) -> JsonValue {
45 self.data
46 }
47}
48
49impl SqliteRawEngine {
50 pub fn new(pool: SqlitePool) -> Self {
52 Self { pool }
53 }
54
55 pub fn pool(&self) -> &SqlitePool {
57 &self.pool
58 }
59
60 fn build_select(
62 &self,
63 table: &str,
64 columns: &[String],
65 filters: &HashMap<String, FilterValue>,
66 sort: &[(String, SortOrder)],
67 limit: Option<u64>,
68 offset: Option<u64>,
69 ) -> (String, Vec<Value>) {
70 let mut sql = String::new();
71 let mut params: Vec<Value> = Vec::new();
72
73 let cols = if columns.is_empty() {
75 "*".to_string()
76 } else {
77 columns
78 .iter()
79 .map(|c| format!("\"{}\"", c))
80 .collect::<Vec<_>>()
81 .join(", ")
82 };
83 sql.push_str(&format!("SELECT {} FROM \"{}\"", cols, table));
84
85 if !filters.is_empty() {
87 let mut conditions = Vec::new();
88 for (field, value) in filters {
89 match value {
90 FilterValue::Null => {
91 conditions.push(format!("\"{}\" IS NULL", field));
92 }
93 _ => {
94 conditions.push(format!("\"{}\" = ?", field));
95 params.push(filter_value_to_sqlite(value));
96 }
97 }
98 }
99 sql.push_str(" WHERE ");
100 sql.push_str(&conditions.join(" AND "));
101 }
102
103 if !sort.is_empty() {
105 let order_parts: Vec<String> = sort
106 .iter()
107 .map(|(col, dir)| {
108 let direction = match dir {
109 SortOrder::Asc => "ASC",
110 SortOrder::Desc => "DESC",
111 };
112 format!("\"{}\" {}", col, direction)
113 })
114 .collect();
115 sql.push_str(" ORDER BY ");
116 sql.push_str(&order_parts.join(", "));
117 }
118
119 if let Some(lim) = limit {
121 sql.push_str(&format!(" LIMIT {}", lim));
122 }
123 if let Some(off) = offset {
124 sql.push_str(&format!(" OFFSET {}", off));
125 }
126
127 (sql, params)
128 }
129
130 fn build_insert(
132 &self,
133 table: &str,
134 data: &HashMap<String, FilterValue>,
135 ) -> (String, Vec<Value>) {
136 let mut columns = Vec::new();
137 let mut placeholders = Vec::new();
138 let mut params: Vec<Value> = Vec::new();
139
140 for (col, val) in data {
141 columns.push(format!("\"{}\"", col));
142 placeholders.push("?".to_string());
143 params.push(filter_value_to_sqlite(val));
144 }
145
146 let sql = format!(
147 "INSERT INTO \"{}\" ({}) VALUES ({})",
148 table,
149 columns.join(", "),
150 placeholders.join(", ")
151 );
152
153 (sql, params)
154 }
155
156 fn build_update(
158 &self,
159 table: &str,
160 data: &HashMap<String, FilterValue>,
161 filters: &HashMap<String, FilterValue>,
162 ) -> (String, Vec<Value>) {
163 let mut params: Vec<Value> = Vec::new();
164
165 let set_parts: Vec<String> = data
167 .iter()
168 .map(|(col, val)| {
169 params.push(filter_value_to_sqlite(val));
170 format!("\"{}\" = ?", col)
171 })
172 .collect();
173
174 let mut sql = format!("UPDATE \"{}\" SET {}", table, set_parts.join(", "));
175
176 if !filters.is_empty() {
178 let mut conditions = Vec::new();
179 for (field, value) in filters {
180 match value {
181 FilterValue::Null => {
182 conditions.push(format!("\"{}\" IS NULL", field));
183 }
184 _ => {
185 conditions.push(format!("\"{}\" = ?", field));
186 params.push(filter_value_to_sqlite(value));
187 }
188 }
189 }
190 sql.push_str(" WHERE ");
191 sql.push_str(&conditions.join(" AND "));
192 }
193
194 (sql, params)
195 }
196
197 fn build_delete(
199 &self,
200 table: &str,
201 filters: &HashMap<String, FilterValue>,
202 ) -> (String, Vec<Value>) {
203 let mut sql = format!("DELETE FROM \"{}\"", table);
204 let mut params: Vec<Value> = Vec::new();
205
206 if !filters.is_empty() {
207 let mut conditions = Vec::new();
208 for (field, value) in filters {
209 match value {
210 FilterValue::Null => {
211 conditions.push(format!("\"{}\" IS NULL", field));
212 }
213 _ => {
214 conditions.push(format!("\"{}\" = ?", field));
215 params.push(filter_value_to_sqlite(value));
216 }
217 }
218 }
219 sql.push_str(" WHERE ");
220 sql.push_str(&conditions.join(" AND "));
221 }
222
223 (sql, params)
224 }
225
226 #[instrument(skip(self, columns, filters, sort), fields(table = %table))]
228 pub async fn query_many(
229 &self,
230 table: &str,
231 columns: &[String],
232 filters: &HashMap<String, FilterValue>,
233 sort: &[(String, SortOrder)],
234 limit: Option<u64>,
235 offset: Option<u64>,
236 ) -> Result<Vec<SqliteJsonRow>, SqliteError> {
237 let (sql, params) = self.build_select(table, columns, filters, sort, limit, offset);
238 trace!(sql = %sql, "Executing query_many");
239
240 let conn = self.pool.get().await?;
241
242 let results = conn.query_params(&sql, params).await?;
243
244 Ok(results.into_iter().map(SqliteJsonRow::new).collect())
245 }
246
247 #[instrument(skip(self, columns, filters), fields(table = %table))]
249 pub async fn query_one(
250 &self,
251 table: &str,
252 columns: &[String],
253 filters: &HashMap<String, FilterValue>,
254 ) -> Result<SqliteJsonRow, SqliteError> {
255 let (sql, params) = self.build_select(table, columns, filters, &[], Some(1), None);
256 trace!(sql = %sql, "Executing query_one");
257
258 let conn = self.pool.get().await?;
259
260 let results = conn.query_params(&sql, params).await?;
261
262 results
263 .into_iter()
264 .next()
265 .map(SqliteJsonRow::new)
266 .ok_or_else(|| {
267 SqliteError::query(format!(
268 "No row found in table '{}' with the given filters",
269 table
270 ))
271 })
272 }
273
274 #[instrument(skip(self, columns, filters), fields(table = %table))]
276 pub async fn query_optional(
277 &self,
278 table: &str,
279 columns: &[String],
280 filters: &HashMap<String, FilterValue>,
281 ) -> Result<Option<SqliteJsonRow>, SqliteError> {
282 let (sql, params) = self.build_select(table, columns, filters, &[], Some(1), None);
283 trace!(sql = %sql, "Executing query_optional");
284
285 let conn = self.pool.get().await?;
286
287 let results = conn.query_params(&sql, params).await?;
288
289 Ok(results.into_iter().next().map(SqliteJsonRow::new))
290 }
291
292 #[instrument(skip(self, data), fields(table = %table))]
294 pub async fn execute_insert(
295 &self,
296 table: &str,
297 data: &HashMap<String, FilterValue>,
298 ) -> Result<SqliteJsonRow, SqliteError> {
299 let (sql, params) = self.build_insert(table, data);
300 trace!(sql = %sql, "Executing insert");
301
302 let conn = self.pool.get().await?;
303
304 let last_rowid = conn.execute_insert_params(&sql, params).await?;
305
306 let mut result = data.clone();
308 if !result.contains_key("id") {
309 result.insert("id".to_string(), FilterValue::Int(last_rowid));
310 }
311
312 let json = result
313 .into_iter()
314 .map(|(k, v)| (k, filter_value_to_json(&v)))
315 .collect::<serde_json::Map<_, _>>();
316
317 Ok(SqliteJsonRow::new(JsonValue::Object(json)))
318 }
319
320 #[instrument(skip(self, data, filters), fields(table = %table))]
322 pub async fn execute_update(
323 &self,
324 table: &str,
325 data: &HashMap<String, FilterValue>,
326 filters: &HashMap<String, FilterValue>,
327 ) -> Result<u64, SqliteError> {
328 let (sql, params) = self.build_update(table, data, filters);
329 trace!(sql = %sql, "Executing update");
330
331 let conn = self.pool.get().await?;
332
333 let affected = conn.execute_params(&sql, params).await?;
334
335 Ok(affected as u64)
336 }
337
338 #[instrument(skip(self, filters), fields(table = %table))]
340 pub async fn execute_delete(
341 &self,
342 table: &str,
343 filters: &HashMap<String, FilterValue>,
344 ) -> Result<u64, SqliteError> {
345 let (sql, params) = self.build_delete(table, filters);
346 trace!(sql = %sql, "Executing delete");
347
348 let conn = self.pool.get().await?;
349
350 let affected = conn.execute_params(&sql, params).await?;
351
352 Ok(affected as u64)
353 }
354
355 #[instrument(skip(self, params), fields(sql = %sql))]
357 pub async fn execute_raw(
358 &self,
359 sql: &str,
360 params: &[FilterValue],
361 ) -> Result<Vec<SqliteJsonRow>, SqliteError> {
362 debug!("Executing raw SQL");
363
364 let sqlite_params: Vec<Value> = params.iter().map(filter_value_to_sqlite).collect();
365
366 let conn = self.pool.get().await?;
367
368 let results = conn.query_params(sql, sqlite_params).await?;
369
370 Ok(results.into_iter().map(SqliteJsonRow::new).collect())
371 }
372
373 #[instrument(skip(self, sql))]
392 pub async fn raw_sql(
393 &self,
394 sql: prax_query::raw::Sql,
395 ) -> Result<Vec<SqliteJsonRow>, SqliteError> {
396 let (query_string, params) = sql.build();
397 debug!(sql = %query_string, "Executing raw SQL from builder");
398 self.raw_sql_query(&query_string, ¶ms).await
399 }
400
401 #[instrument(skip(self, params), fields(sql = %sql))]
412 pub async fn raw_sql_query(
413 &self,
414 sql: &str,
415 params: &[FilterValue],
416 ) -> Result<Vec<SqliteJsonRow>, SqliteError> {
417 debug!("Executing raw SQL query");
418
419 let sqlite_params: Vec<Value> = params.iter().map(filter_value_to_sqlite).collect();
420
421 let conn = self.pool.get().await?;
422
423 let results = conn.query_params(sql, sqlite_params).await?;
424
425 Ok(results.into_iter().map(SqliteJsonRow::new).collect())
426 }
427
428 #[instrument(skip(self, params), fields(sql = %sql))]
442 pub async fn raw_sql_execute(
443 &self,
444 sql: &str,
445 params: &[FilterValue],
446 ) -> Result<u64, SqliteError> {
447 debug!("Executing raw SQL statement");
448
449 let sqlite_params: Vec<Value> = params.iter().map(filter_value_to_sqlite).collect();
450
451 let conn = self.pool.get().await?;
452
453 let affected = conn.execute_params(sql, sqlite_params).await?;
454
455 Ok(affected as u64)
456 }
457
458 #[instrument(skip(self, params), fields(sql = %sql))]
471 pub async fn raw_sql_first(
472 &self,
473 sql: &str,
474 params: &[FilterValue],
475 ) -> Result<SqliteJsonRow, SqliteError> {
476 debug!("Executing raw SQL first");
477
478 let sqlite_params: Vec<Value> = params.iter().map(filter_value_to_sqlite).collect();
479
480 let conn = self.pool.get().await?;
481
482 let results = conn.query_params(sql, sqlite_params).await?;
483
484 results
485 .into_iter()
486 .next()
487 .map(SqliteJsonRow::new)
488 .ok_or_else(|| SqliteError::query("raw_sql_first returned no rows"))
489 }
490
491 #[instrument(skip(self, params), fields(sql = %sql))]
502 pub async fn raw_sql_optional(
503 &self,
504 sql: &str,
505 params: &[FilterValue],
506 ) -> Result<Option<SqliteJsonRow>, SqliteError> {
507 debug!("Executing raw SQL optional");
508
509 let sqlite_params: Vec<Value> = params.iter().map(filter_value_to_sqlite).collect();
510
511 let conn = self.pool.get().await?;
512
513 let results = conn.query_params(sql, sqlite_params).await?;
514
515 Ok(results.into_iter().next().map(SqliteJsonRow::new))
516 }
517
518 #[instrument(skip(self, params), fields(sql = %sql))]
529 pub async fn raw_sql_scalar<T>(
530 &self,
531 sql: &str,
532 params: &[FilterValue],
533 ) -> Result<T, SqliteError>
534 where
535 T: for<'a> serde::Deserialize<'a>,
536 {
537 debug!("Executing raw SQL scalar");
538
539 let sqlite_params: Vec<Value> = params.iter().map(filter_value_to_sqlite).collect();
540
541 let conn = self.pool.get().await?;
542
543 let results = conn.query_params(sql, sqlite_params).await?;
544
545 let row = results
546 .into_iter()
547 .next()
548 .ok_or_else(|| SqliteError::query("raw_sql_scalar returned no rows"))?;
549
550 let value = row
552 .as_object()
553 .and_then(|obj| obj.values().next())
554 .ok_or_else(|| SqliteError::query("raw_sql_scalar returned empty row"))?;
555
556 serde_json::from_value(value.clone()).map_err(|e| {
557 SqliteError::deserialization(format!("failed to deserialize scalar: {}", e))
558 })
559 }
560
561 #[instrument(skip(self), fields(sql_len = %sql.len()))]
574 pub async fn raw_sql_batch(&self, sql: &str) -> Result<(), SqliteError> {
575 debug!("Executing raw SQL batch");
576
577 let conn = self.pool.get().await?;
578
579 conn.execute_batch(sql).await
580 }
581
582 #[instrument(skip(self, filters), fields(table = %table))]
584 pub async fn count(
585 &self,
586 table: &str,
587 filters: &HashMap<String, FilterValue>,
588 ) -> Result<u64, SqliteError> {
589 let mut sql = format!("SELECT COUNT(*) as count FROM \"{}\"", table);
590 let mut params: Vec<Value> = Vec::new();
591
592 if !filters.is_empty() {
593 let mut conditions = Vec::new();
594 for (field, value) in filters {
595 match value {
596 FilterValue::Null => {
597 conditions.push(format!("\"{}\" IS NULL", field));
598 }
599 _ => {
600 conditions.push(format!("\"{}\" = ?", field));
601 params.push(filter_value_to_sqlite(value));
602 }
603 }
604 }
605 sql.push_str(" WHERE ");
606 sql.push_str(&conditions.join(" AND "));
607 }
608
609 trace!(sql = %sql, "Executing count");
610
611 let conn = self.pool.get().await?;
612
613 let results = conn.query_params(&sql, params).await?;
614
615 let count = results
617 .first()
618 .and_then(|row| row.get("count"))
619 .and_then(|v| v.as_i64())
620 .unwrap_or(0);
621
622 Ok(count as u64)
623 }
624}
625
626fn filter_value_to_json(value: &FilterValue) -> JsonValue {
628 match value {
629 FilterValue::Null => JsonValue::Null,
630 FilterValue::Bool(b) => JsonValue::Bool(*b),
631 FilterValue::Int(i) => JsonValue::Number((*i).into()),
632 FilterValue::Float(f) => serde_json::Number::from_f64(*f)
633 .map(JsonValue::Number)
634 .unwrap_or(JsonValue::Null),
635 FilterValue::String(s) => JsonValue::String(s.clone()),
636 FilterValue::Json(j) => j.clone(),
637 FilterValue::List(list) => {
638 JsonValue::Array(list.iter().map(filter_value_to_json).collect())
639 }
640 }
641}
642
643#[cfg(test)]
644mod tests {
645 use super::*;
646
647 #[test]
648 fn test_filter_value_to_json() {
649 assert_eq!(filter_value_to_json(&FilterValue::Null), JsonValue::Null);
650 assert_eq!(
651 filter_value_to_json(&FilterValue::Bool(true)),
652 JsonValue::Bool(true)
653 );
654 assert_eq!(
655 filter_value_to_json(&FilterValue::Int(42)),
656 JsonValue::Number(42.into())
657 );
658 assert_eq!(
659 filter_value_to_json(&FilterValue::String("test".to_string())),
660 JsonValue::String("test".to_string())
661 );
662 }
663
664 #[test]
665 fn test_build_select_simple() {
666 let sql = "SELECT * FROM \"users\"";
667 assert!(sql.contains("SELECT"));
668 assert!(sql.contains("users"));
669 }
670
671 #[test]
672 fn test_query_result() {
673 let result = SqliteJsonRow::new(JsonValue::Object(serde_json::Map::new()));
674 assert!(result.json().is_object());
675 }
676
677 #[test]
678 fn test_query_result_into_json() {
679 let json = JsonValue::Object(serde_json::Map::new());
680 let result = SqliteJsonRow::new(json.clone());
681 assert_eq!(result.into_json(), json);
682 }
683
684 #[test]
685 fn test_sql_builder_integration() {
686 use prax_query::raw::Sql;
687
688 let sql = Sql::new("SELECT * FROM users WHERE age > ")
689 .bind(18)
690 .push(" AND active = ")
691 .bind(true);
692
693 let (query, params) = sql.build();
694 assert!(query.contains("SELECT"));
695 assert!(query.contains("users"));
696 assert_eq!(params.len(), 2);
697 }
698}