rws_server/model/
query.rs

1//! Async fluent `QueryBuilder<T>` for constructing SQL queries.
2
3use std::marker::PhantomData;
4
5use super::backend::Backend;
6use super::pool::DbPool;
7use super::repository::{extract_count, placeholder};
8use super::{DbError, Model, ToColumn, Value};
9use crate::pagination::{CursorPage, Page};
10
11// ── Order ─────────────────────────────────────────────────────────────────────
12
13/// Sort direction for `order_by`.
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum Order {
16    Asc,
17    Desc,
18}
19
20impl Order {
21    fn as_sql(self) -> &'static str {
22        match self {
23            Order::Asc => "ASC",
24            Order::Desc => "DESC",
25        }
26    }
27}
28
29// ── QueryBuilder ──────────────────────────────────────────────────────────────
30
31/// Fluent async query builder for `SELECT`, `COUNT`, `DELETE`, and `UPDATE`.
32///
33/// ```rust,no_run
34/// # extern crate rws_server as rust_web_server;
35/// # #[cfg(feature = "model-sqlite")]
36/// # async fn example() -> Result<(), rust_web_server::model::DbError> {
37/// use rust_web_server::model::{DbPool, Order, QueryBuilder};
38///
39/// let pool = DbPool::memory().await?;
40/// // pool.execute("CREATE TABLE users …", &[]).await?;
41///
42/// // let users: Vec<User> = QueryBuilder::<User>::new(&pool)
43/// //     .where_eq("active", true)
44/// //     .order_by("name", Order::Asc)
45/// //     .limit(10)
46/// //     .fetch_all()
47/// //     .await?;
48/// # Ok(())
49/// # }
50/// ```
51pub struct QueryBuilder<'a, T: Model> {
52    pool: &'a DbPool,
53    /// Each filter is a (sql_fragment, params) pair.
54    /// sql_fragment may contain `__placeholder__` (from `where_eq`) or raw `?`.
55    filters: Vec<(String, Vec<Value>)>,
56    order: Option<(String, Order)>,
57    limit: Option<u64>,
58    offset: Option<u64>,
59    _phantom: PhantomData<T>,
60}
61
62// Implemented by hand rather than `#[derive(Clone)]`: the derive macro adds a
63// `T: Clone` bound even though `T` only appears behind `PhantomData`, which
64// would force every `QueryBuilder<T>` user's model type to be `Clone` just to
65// call `.paginate()`.
66impl<'a, T: Model> Clone for QueryBuilder<'a, T> {
67    fn clone(&self) -> Self {
68        QueryBuilder {
69            pool: self.pool,
70            filters: self.filters.clone(),
71            order: self.order.clone(),
72            limit: self.limit,
73            offset: self.offset,
74            _phantom: PhantomData,
75        }
76    }
77}
78
79impl<'a, T: Model> QueryBuilder<'a, T> {
80    pub fn new(pool: &'a DbPool) -> Self {
81        QueryBuilder {
82            pool,
83            filters: Vec::new(),
84            order: None,
85            limit: None,
86            offset: None,
87            _phantom: PhantomData,
88        }
89    }
90
91    /// Add a `col = ?` equality filter.
92    pub fn where_eq(mut self, col: &str, val: impl ToColumn) -> Self {
93        self.filters.push((
94            format!("{} = __placeholder__", col),
95            vec![val.to_column()],
96        ));
97        self
98    }
99
100    /// Add a raw SQL filter fragment with bind parameters.
101    ///
102    /// ```ignore
103    /// .filter("age >= ?", vec![Value::Int(18)])
104    /// ```
105    pub fn filter(mut self, expr: &str, params: Vec<Value>) -> Self {
106        self.filters.push((expr.to_owned(), params));
107        self
108    }
109
110    /// Set the ORDER BY clause.
111    pub fn order_by(mut self, col: &str, order: Order) -> Self {
112        self.order = Some((col.to_owned(), order));
113        self
114    }
115
116    /// Set the LIMIT clause.
117    pub fn limit(mut self, n: u64) -> Self {
118        self.limit = Some(n);
119        self
120    }
121
122    /// Set the OFFSET clause.
123    pub fn offset(mut self, n: u64) -> Self {
124        self.offset = Some(n);
125        self
126    }
127
128    // ── Terminal operations ───────────────────────────────────────────────────
129
130    /// Execute `SELECT * FROM table WHERE … ORDER BY … LIMIT … OFFSET …`.
131    pub async fn fetch_all(self) -> Result<Vec<T>, DbError> {
132        let backend = self.pool.backend();
133        let (sql, params) = build_select::<T>(
134            self.filters, self.order, self.limit, self.offset, "*", backend,
135        );
136        let rows = self.pool.query_rows(&sql, &params).await?;
137        rows.iter().map(|r| T::from_row(r)).collect()
138    }
139
140    /// Execute `SELECT * … LIMIT 1` and return the first result.
141    pub async fn fetch_one(self) -> Result<Option<T>, DbError> {
142        let backend = self.pool.backend();
143        let (sql, params) = build_select::<T>(
144            self.filters, self.order, Some(1), self.offset, "*", backend,
145        );
146        let rows = self.pool.query_rows(&sql, &params).await?;
147        match rows.into_iter().next() {
148            Some(row) => Ok(Some(T::from_row(&row)?)),
149            None => Ok(None),
150        }
151    }
152
153    /// Execute `SELECT COUNT(*) FROM table WHERE …`.
154    pub async fn count(self) -> Result<i64, DbError> {
155        let backend = self.pool.backend();
156        let (sql, params) = build_select::<T>(
157            self.filters, self.order, self.limit, self.offset, "COUNT(*)", backend,
158        );
159        let rows = self.pool.query_rows(&sql, &params).await?;
160        extract_count(rows)
161    }
162
163    /// Execute `DELETE FROM table WHERE …`.
164    pub async fn delete(self) -> Result<(), DbError> {
165        let backend = self.pool.backend();
166        let (where_clause, params, _) = build_where(self.filters, 0, backend);
167        let sql = format!("DELETE FROM {}{}", T::table_name(), where_clause);
168        self.pool.execute(&sql, &params).await?;
169        Ok(())
170    }
171
172    /// Execute `UPDATE table SET col = ? WHERE …`.
173    pub async fn update(self, col: &str, val: impl ToColumn) -> Result<(), DbError> {
174        let backend = self.pool.backend();
175        let set_ph = placeholder(backend, 1);
176        let set_val = val.to_column();
177        let (where_clause, mut where_params, _) = build_where(self.filters, 1, backend);
178        let mut params = vec![set_val];
179        params.append(&mut where_params);
180        let sql = format!(
181            "UPDATE {} SET {} = {}{}",
182            T::table_name(), col, set_ph, where_clause,
183        );
184        self.pool.execute(&sql, &params).await?;
185        Ok(())
186    }
187
188    /// Offset-paginated fetch: runs a `COUNT(*)` (with the same filters, no
189    /// limit/offset) and a `SELECT … LIMIT … OFFSET …`, and wraps both into a
190    /// [`Page`]. `page` is 1-based; both `page` and `per_page` are clamped to
191    /// a minimum of `1`. Overrides any `.limit()`/`.offset()` set earlier in
192    /// the chain — `page`/`per_page` are authoritative.
193    ///
194    /// Two queries, not one — if you're paginating a very large, frequently
195    /// appended-to table and don't need `total_pages`, [`Self::paginate_after`]
196    /// (keyset pagination) needs only one.
197    pub async fn paginate(self, page: u64, per_page: u64) -> Result<Page<T>, DbError> {
198        let page = page.max(1);
199        let per_page = per_page.max(1);
200        let offset = (page - 1) * per_page;
201
202        let total_items = self.clone().count().await? as u64;
203        let items = self.limit(per_page).offset(offset).fetch_all().await?;
204
205        Ok(Page::new(items, page, per_page, total_items))
206    }
207
208    /// Cursor (keyset) paginated fetch: orders by the primary key ascending,
209    /// fetches `per_page + 1` rows to cheaply detect whether there's a next
210    /// page, and returns a [`CursorPage`] whose `next_cursor` is the last
211    /// returned row's primary key (as a string) — pass it back as `cursor` to
212    /// get the next page. `cursor: None` fetches the first page.
213    ///
214    /// A single query, unlike [`Self::paginate`] — no `COUNT(*)`, and no
215    /// `OFFSET` to skip over on every subsequent page, which is what makes
216    /// keyset pagination scale on large tables where offset pagination gets
217    /// slower page by page. The tradeoff: no `total_items`/`total_pages`, and
218    /// only forward iteration (no jumping to an arbitrary page).
219    ///
220    /// Overrides any `.order_by()` set earlier in the chain — keyset
221    /// pagination requires ordering by the cursor column. Requires the
222    /// primary key to be numeric (parsed as `i64`); returns `Err` if `cursor`
223    /// isn't a valid integer.
224    pub async fn paginate_after(self, cursor: Option<&str>, per_page: u64) -> Result<CursorPage<T>, DbError> {
225        let per_page = per_page.max(1);
226        let pk_col = T::primary_key_name();
227
228        let mut builder = self;
229        if let Some(cursor_str) = cursor {
230            let cursor_val: i64 = cursor_str
231                .parse()
232                .map_err(|_| DbError::new(format!("invalid cursor '{}': expected an integer primary key", cursor_str)))?;
233            builder = builder.filter(&format!("{} > __placeholder__", pk_col), vec![Value::Int(cursor_val)]);
234        }
235
236        let mut rows = builder.order_by(pk_col, Order::Asc).limit(per_page + 1).fetch_all().await?;
237
238        let has_more = rows.len() as u64 > per_page;
239        if has_more {
240            rows.truncate(per_page as usize);
241        }
242
243        let next_cursor = if has_more {
244            rows.last().map(|item| match item.primary_key_value() {
245                Value::Int(n) => n.to_string(),
246                Value::Text(s) => s,
247                other => format!("{:?}", other),
248            })
249        } else {
250            None
251        };
252
253        Ok(CursorPage { items: rows, next_cursor })
254    }
255}
256
257// ── Internal SQL builders ─────────────────────────────────────────────────────
258
259fn build_select<T: Model>(
260    filters: Vec<(String, Vec<Value>)>,
261    order: Option<(String, Order)>,
262    limit: Option<u64>,
263    offset: Option<u64>,
264    projection: &str,
265    backend: Backend,
266) -> (String, Vec<Value>) {
267    let (where_clause, params, _) = build_where(filters, 0, backend);
268    let mut sql = format!("SELECT {} FROM {}{}", projection, T::table_name(), where_clause);
269    if let Some((col, ord)) = order {
270        sql.push_str(&format!(" ORDER BY {} {}", col, ord.as_sql()));
271    }
272    if let Some(n) = limit {
273        sql.push_str(&format!(" LIMIT {}", n));
274    }
275    if let Some(n) = offset {
276        sql.push_str(&format!(" OFFSET {}", n));
277    }
278    (sql, params)
279}
280
281/// Build the WHERE clause. `start_idx` offsets PostgreSQL `$N` placeholders.
282#[doc(hidden)]
283pub fn build_where(
284    filters: Vec<(String, Vec<Value>)>,
285    start_idx: usize,
286    backend: Backend,
287) -> (String, Vec<Value>, usize) {
288    let mut all_params: Vec<Value> = Vec::new();
289    let mut conditions: Vec<String> = Vec::new();
290    let mut idx = start_idx;
291
292    for (mut fragment, params) in filters {
293        for param in params {
294            idx += 1;
295            if fragment.contains("__placeholder__") {
296                fragment = fragment.replacen("__placeholder__", &placeholder(backend, idx), 1);
297            } else {
298                // A raw fragment written with a literal `?` (e.g. from
299                // `.filter("age >= ?", ...)`) — no-op on backends that
300                // already use `?`, rewrites it to `$N` on PostgreSQL.
301                fragment = fragment.replacen("?", &placeholder(backend, idx), 1);
302            }
303            all_params.push(param);
304        }
305        conditions.push(fragment);
306    }
307
308    let where_clause = if conditions.is_empty() {
309        String::new()
310    } else {
311        format!(" WHERE {}", conditions.join(" AND "))
312    };
313
314    (where_clause, all_params, idx)
315}
rws_server/model/query.rs

rws_server/model/
query.rs
