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, live_condition, placeholder, timestamp_placeholder};
8use super::timestamp::Timestamp;
9use super::{DbError, Model, ToColumn, Value};
10use crate::pagination::{CursorPage, Page};
11
12// ── Order ─────────────────────────────────────────────────────────────────────
13
14/// Sort direction for `order_by`.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum Order {
17    Asc,
18    Desc,
19}
20
21impl Order {
22    fn as_sql(self) -> &'static str {
23        match self {
24            Order::Asc => "ASC",
25            Order::Desc => "DESC",
26        }
27    }
28}
29
30// ── Deleted ───────────────────────────────────────────────────────────────────
31
32/// Which rows of a soft-delete model a query sees.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34enum Deleted {
35    /// Not soft-deleted (the default).
36    Exclude,
37    /// All.
38    Include,
39    /// Soft-deleted only.
40    Only,
41}
42
43// ── QueryBuilder ──────────────────────────────────────────────────────────────
44
45/// Fluent async query builder for `SELECT`, `COUNT`, `DELETE`, and `UPDATE`.
46///
47/// For a model with a soft-delete column (`#[soft_delete]`) every operation
48/// sees only rows not deleted, unless the chain has
49/// [`with_deleted`](Self::with_deleted) or
50/// [`only_deleted`](Self::only_deleted); [`delete`](Self::delete) sets the
51/// column and [`force_delete`](Self::force_delete) removes rows. With
52/// `#[timestamps]` / `#[version]`, [`update`](Self::update) also sets
53/// `updated_at` and increments the version (18.30.0).
54///
55/// ```rust,no_run
56/// # extern crate rws_server as rust_web_server;
57/// # #[cfg(feature = "model-sqlite")]
58/// # async fn example() -> Result<(), rust_web_server::model::DbError> {
59/// use rust_web_server::model::{DbPool, Order, QueryBuilder};
60///
61/// let pool = DbPool::memory().await?;
62/// // pool.execute("CREATE TABLE users …", &[]).await?;
63///
64/// // let users: Vec<User> = QueryBuilder::<User>::new(&pool)
65/// //     .where_eq("active", true)
66/// //     .order_by("name", Order::Asc)
67/// //     .limit(10)
68/// //     .fetch_all()
69/// //     .await?;
70/// # Ok(())
71/// # }
72/// ```
73pub struct QueryBuilder<'a, T: Model> {
74    pool: &'a DbPool,
75    /// Each filter is a (sql_fragment, params) pair.
76    /// sql_fragment may contain `__placeholder__` (from `where_eq`) or raw `?`.
77    filters: Vec<(String, Vec<Value>)>,
78    order: Option<(String, Order)>,
79    limit: Option<u64>,
80    offset: Option<u64>,
81    deleted: Deleted,
82    _phantom: PhantomData<T>,
83}
84
85// Implemented by hand rather than `#[derive(Clone)]`: the derive macro adds a
86// `T: Clone` bound even though `T` only appears behind `PhantomData`, which
87// would force every `QueryBuilder<T>` user's model type to be `Clone` just to
88// call `.paginate()`.
89impl<'a, T: Model> Clone for QueryBuilder<'a, T> {
90    fn clone(&self) -> Self {
91        QueryBuilder {
92            pool: self.pool,
93            filters: self.filters.clone(),
94            order: self.order.clone(),
95            limit: self.limit,
96            offset: self.offset,
97            deleted: self.deleted,
98            _phantom: PhantomData,
99        }
100    }
101}
102
103impl<'a, T: Model> QueryBuilder<'a, T> {
104    pub fn new(pool: &'a DbPool) -> Self {
105        QueryBuilder {
106            pool,
107            filters: Vec::new(),
108            order: None,
109            limit: None,
110            offset: None,
111            deleted: Deleted::Exclude,
112            _phantom: PhantomData,
113        }
114    }
115
116    /// Soft-deleted rows too (no effect on a model without `#[soft_delete]`).
117    pub fn with_deleted(mut self) -> Self {
118        self.deleted = Deleted::Include;
119        self
120    }
121
122    /// Soft-deleted rows only (none for a model without `#[soft_delete]`).
123    pub fn only_deleted(mut self) -> Self {
124        self.deleted = Deleted::Only;
125        self
126    }
127
128    /// The filters plus the soft-delete scope.
129    fn scoped_filters(&self) -> Vec<(String, Vec<Value>)> {
130        let mut filters = self.filters.clone();
131        if let Some(col) = T::soft_delete_column() {
132            match self.deleted {
133                Deleted::Exclude => filters.extend(live_condition::<T>().map(|c| (c, Vec::new()))),
134                Deleted::Include => {}
135                Deleted::Only => filters.push((format!("{col} IS NOT NULL"), Vec::new())),
136            }
137        } else if self.deleted == Deleted::Only {
138            filters.push(("1 = 0".to_string(), Vec::new()));
139        }
140        filters
141    }
142
143    /// Add a `col = ?` equality filter.
144    pub fn where_eq(mut self, col: &str, val: impl ToColumn) -> Self {
145        self.filters.push((
146            format!("{} = __placeholder__", col),
147            vec![val.to_column()],
148        ));
149        self
150    }
151
152    /// Add a `col IN (?, ?, …)` filter (18.30.0); no values match no rows.
153    pub fn where_in<I>(mut self, col: &str, values: I) -> Self
154    where
155        I: IntoIterator,
156        I::Item: ToColumn,
157    {
158        let values: Vec<Value> = values.into_iter().map(|v| v.to_column()).collect();
159        if values.is_empty() {
160            self.filters.push(("1 = 0".to_string(), Vec::new()));
161        } else {
162            let marks = vec!["__placeholder__"; values.len()].join(", ");
163            self.filters.push((format!("{col} IN ({marks})"), values));
164        }
165        self
166    }
167
168    /// Add a `col IS NULL` filter (18.30.0).
169    pub fn where_null(mut self, col: &str) -> Self {
170        self.filters.push((format!("{col} IS NULL"), Vec::new()));
171        self
172    }
173
174    /// Add a `col IS NOT NULL` filter (18.30.0).
175    pub fn where_not_null(mut self, col: &str) -> Self {
176        self.filters.push((format!("{col} IS NOT NULL"), Vec::new()));
177        self
178    }
179
180    /// Apply a scope (18.30.0): a reusable function from a query to a
181    /// narrower one. Scopes are plain functions (or closures), so they
182    /// take arguments and compose:
183    ///
184    /// ```rust,ignore
185    /// fn published(q: QueryBuilder<'_, Post>) -> QueryBuilder<'_, Post> {
186    ///     q.where_eq("published", true).where_null("hidden_at")
187    /// }
188    /// fn by(author: i64) -> impl FnOnce(QueryBuilder<'_, Post>) -> QueryBuilder<'_, Post> {
189    ///     move |q| q.where_eq("author_id", author)
190    /// }
191    /// let mine = Post::query(&pool).scope(published).scope(by(7)).fetch_all().await?;
192    /// ```
193    ///
194    /// For `Post::query(&pool).published()`, put them in an extension
195    /// trait:
196    ///
197    /// ```rust,ignore
198    /// pub trait PostScopes {
199    ///     fn published(self) -> Self;
200    /// }
201    /// impl PostScopes for QueryBuilder<'_, Post> {
202    ///     fn published(self) -> Self {
203    ///         self.where_eq("published", true)
204    ///     }
205    /// }
206    /// ```
207    pub fn scope(self, scope: impl FnOnce(Self) -> Self) -> Self {
208        scope(self)
209    }
210
211    /// Load `relation` for every row the query answers (18.30.0): one more
212    /// query per relation, however many rows. The `fetch_*` of the
213    /// [`EagerQuery`](super::EagerQuery) it returns answer
214    /// [`Loaded`](super::Loaded) models.
215    ///
216    /// ```rust,ignore
217    /// let posts = Post::query(&pool).with(Post::comments()).fetch_all().await?;
218    /// let first_comments: &[Comment] = posts[0].related(&Post::comments());
219    /// ```
220    pub fn with<R: Model + Clone + 'static>(self, relation: super::Relation<T, R>) -> super::EagerQuery<'a, T>
221    where
222        T: 'static,
223    {
224        super::EagerQuery::new(self).with(relation)
225    }
226
227    /// Load `relation` with relations of its own (18.35.0); see
228    /// [`EagerQuery::with_nested`](super::EagerQuery::with_nested).
229    pub fn with_nested<R: Model + Clone + 'static>(self, relation: super::Relation<T, R>, nested: impl FnOnce(super::Nested<R>) -> super::Nested<R>) -> super::EagerQuery<'a, T>
230    where
231        T: 'static,
232    {
233        super::EagerQuery::new(self).with_nested(relation, nested)
234    }
235
236    /// The pool the query runs on.
237    pub(crate) fn pool(&self) -> &'a DbPool {
238        self.pool
239    }
240
241    /// Add a raw SQL filter fragment with bind parameters.
242    ///
243    /// ```ignore
244    /// .filter("age >= ?", vec![Value::Int(18)])
245    /// ```
246    pub fn filter(mut self, expr: &str, params: Vec<Value>) -> Self {
247        self.filters.push((expr.to_owned(), params));
248        self
249    }
250
251    /// Set the ORDER BY clause.
252    pub fn order_by(mut self, col: &str, order: Order) -> Self {
253        self.order = Some((col.to_owned(), order));
254        self
255    }
256
257    /// Set the LIMIT clause.
258    pub fn limit(mut self, n: u64) -> Self {
259        self.limit = Some(n);
260        self
261    }
262
263    /// Set the OFFSET clause.
264    pub fn offset(mut self, n: u64) -> Self {
265        self.offset = Some(n);
266        self
267    }
268
269    // ── Terminal operations ───────────────────────────────────────────────────
270
271    /// Execute `SELECT * FROM table WHERE … ORDER BY … LIMIT … OFFSET …`.
272    pub async fn fetch_all(self) -> Result<Vec<T>, DbError> {
273        let backend = self.pool.backend();
274        let filters = self.scoped_filters();
275        let (sql, params) = build_select::<T>(
276            filters, self.order, self.limit, self.offset, "*", backend,
277        );
278        let rows = self.pool.query_rows(&sql, &params).await?;
279        rows.iter().map(|r| T::from_row(r)).collect()
280    }
281
282    /// Execute `SELECT * … LIMIT 1` and return the first result.
283    pub async fn fetch_one(self) -> Result<Option<T>, DbError> {
284        let backend = self.pool.backend();
285        let filters = self.scoped_filters();
286        let (sql, params) = build_select::<T>(
287            filters, self.order, Some(1), self.offset, "*", backend,
288        );
289        let rows = self.pool.query_rows(&sql, &params).await?;
290        match rows.into_iter().next() {
291            Some(row) => Ok(Some(T::from_row(&row)?)),
292            None => Ok(None),
293        }
294    }
295
296    /// Execute `SELECT COUNT(*) FROM table WHERE …` (without the chain's
297    /// `ORDER BY`, which PostgreSQL refuses next to an aggregate).
298    pub async fn count(self) -> Result<i64, DbError> {
299        let backend = self.pool.backend();
300        let filters = self.scoped_filters();
301        let (sql, params) = build_select::<T>(
302            filters, None, self.limit, self.offset, "COUNT(*)", backend,
303        );
304        let rows = self.pool.query_rows(&sql, &params).await?;
305        extract_count(rows)
306    }
307
308    /// Execute `DELETE FROM table WHERE …` — for a soft-delete model,
309    /// `UPDATE table SET deleted_at = now WHERE … AND deleted_at IS NULL`
310    /// instead ([`force_delete`](Self::force_delete) removes the rows).
311    pub async fn delete(self) -> Result<(), DbError> {
312        let Some(col) = T::soft_delete_column() else {
313            return self.force_delete().await.map(|_| ());
314        };
315        let backend = self.pool.backend();
316        let mut filters = self.filters.clone();
317        filters.extend(live_condition::<T>().map(|c| (c, Vec::new())));
318        let (where_clause, where_params, _) = build_where(filters, 1, backend);
319        let mut params = vec![Timestamp::now().to_value(backend)];
320        params.extend(where_params);
321        let sql = format!("UPDATE {} SET {} = {}{}", T::table_name(), col, timestamp_placeholder(backend, 1), where_clause);
322        self.pool.execute(&sql, &params).await?;
323        Ok(())
324    }
325
326    /// Execute `DELETE FROM table WHERE …`, removing the rows whatever the
327    /// soft-delete column says (within the query's scope: add
328    /// [`with_deleted`](Self::with_deleted) to include deleted rows).
329    /// Answers the number of rows removed.
330    pub async fn force_delete(self) -> Result<u64, DbError> {
331        let backend = self.pool.backend();
332        let (where_clause, params, _) = build_where(self.scoped_filters(), 0, backend);
333        let sql = format!("DELETE FROM {}{}", T::table_name(), where_clause);
334        self.pool.execute(&sql, &params).await
335    }
336
337    /// Clear the soft-delete column of the matching deleted rows; answers
338    /// how many. An error for a model without `#[soft_delete]`.
339    pub async fn restore(self) -> Result<u64, DbError> {
340        let Some(col) = T::soft_delete_column() else {
341            return Err(DbError::new(format!(
342                "restore: {} has no soft-delete column (add #[soft_delete] and a deleted_at field)",
343                T::table_name()
344            )));
345        };
346        let backend = self.pool.backend();
347        let mut filters = self.filters.clone();
348        filters.push((format!("{col} IS NOT NULL"), Vec::new()));
349        let (where_clause, params, _) = build_where(filters, 0, backend);
350        let sql = format!("UPDATE {} SET {} = NULL{}", T::table_name(), col, where_clause);
351        self.pool.execute(&sql, &params).await
352    }
353
354    /// Execute `UPDATE table SET col = ? WHERE …` (also `updated_at = now`
355    /// and `version = version + 1` when the model has them).
356    pub async fn update(self, col: &str, val: impl ToColumn) -> Result<(), DbError> {
357        let backend = self.pool.backend();
358        let (mut sets, mut params) = match val.to_column() {
359            // A literal NULL: PostgreSQL won't assign one bound as text to
360            // a column of another type.
361            Value::Null => (vec![format!("{col} = NULL")], Vec::new()),
362            value => (vec![format!("{} = {}", col, placeholder(backend, 1))], vec![value]),
363        };
364        if let Some(updated) = T::updated_at_column().filter(|c| *c != col) {
365            params.push(Timestamp::now().to_value(backend));
366            sets.push(format!("{} = {}", updated, timestamp_placeholder(backend, params.len())));
367        }
368        if let Some(version) = T::version_column().filter(|c| *c != col) {
369            sets.push(format!("{version} = {version} + 1"));
370        }
371        let (where_clause, where_params, _) = build_where(self.scoped_filters(), params.len(), backend);
372        params.extend(where_params);
373        let sql = format!("UPDATE {} SET {}{}", T::table_name(), sets.join(", "), where_clause);
374        self.pool.execute(&sql, &params).await?;
375        Ok(())
376    }
377
378    /// Offset-paginated fetch: runs a `COUNT(*)` (with the same filters, no
379    /// limit/offset) and a `SELECT … LIMIT … OFFSET …`, and wraps both into a
380    /// [`Page`]. `page` is 1-based; both `page` and `per_page` are clamped to
381    /// a minimum of `1`. Overrides any `.limit()`/`.offset()` set earlier in
382    /// the chain — `page`/`per_page` are authoritative.
383    ///
384    /// Two queries, not one — if you're paginating a very large, frequently
385    /// appended-to table and don't need `total_pages`, [`Self::paginate_after`]
386    /// (keyset pagination) needs only one.
387    pub async fn paginate(self, page: u64, per_page: u64) -> Result<Page<T>, DbError> {
388        let page = page.max(1);
389        let per_page = per_page.max(1);
390        let offset = (page - 1) * per_page;
391
392        let total_items = self.clone().count().await? as u64;
393        let items = self.limit(per_page).offset(offset).fetch_all().await?;
394
395        Ok(Page::new(items, page, per_page, total_items))
396    }
397
398    /// Cursor (keyset) paginated fetch: orders by the primary key ascending,
399    /// fetches `per_page + 1` rows to cheaply detect whether there's a next
400    /// page, and returns a [`CursorPage`] whose `next_cursor` is the last
401    /// returned row's primary key (as a string) — pass it back as `cursor` to
402    /// get the next page. `cursor: None` fetches the first page.
403    ///
404    /// A single query, unlike [`Self::paginate`] — no `COUNT(*)`, and no
405    /// `OFFSET` to skip over on every subsequent page, which is what makes
406    /// keyset pagination scale on large tables where offset pagination gets
407    /// slower page by page. The tradeoff: no `total_items`/`total_pages`, and
408    /// only forward iteration (no jumping to an arbitrary page).
409    ///
410    /// Overrides any `.order_by()` set earlier in the chain — keyset
411    /// pagination requires ordering by the cursor column. Requires the
412    /// primary key to be numeric (parsed as `i64`); returns `Err` if `cursor`
413    /// isn't a valid integer.
414    pub async fn paginate_after(self, cursor: Option<&str>, per_page: u64) -> Result<CursorPage<T>, DbError> {
415        let per_page = per_page.max(1);
416        let pk_col = T::primary_key_name();
417
418        let mut builder = self;
419        if let Some(cursor_str) = cursor {
420            let cursor_val: i64 = cursor_str
421                .parse()
422                .map_err(|_| DbError::new(format!("invalid cursor '{}': expected an integer primary key", cursor_str)))?;
423            builder = builder.filter(&format!("{} > __placeholder__", pk_col), vec![Value::Int(cursor_val)]);
424        }
425
426        let mut rows = builder.order_by(pk_col, Order::Asc).limit(per_page + 1).fetch_all().await?;
427
428        let has_more = rows.len() as u64 > per_page;
429        if has_more {
430            rows.truncate(per_page as usize);
431        }
432
433        let next_cursor = if has_more {
434            rows.last().map(|item| match item.primary_key_value() {
435                Value::Int(n) => n.to_string(),
436                Value::Text(s) => s,
437                other => format!("{:?}", other),
438            })
439        } else {
440            None
441        };
442
443        Ok(CursorPage { items: rows, next_cursor })
444    }
445}
446
447// ── Internal SQL builders ─────────────────────────────────────────────────────
448
449fn build_select<T: Model>(
450    filters: Vec<(String, Vec<Value>)>,
451    order: Option<(String, Order)>,
452    limit: Option<u64>,
453    offset: Option<u64>,
454    projection: &str,
455    backend: Backend,
456) -> (String, Vec<Value>) {
457    let (where_clause, params, _) = build_where(filters, 0, backend);
458    let mut sql = format!("SELECT {} FROM {}{}", projection, T::table_name(), where_clause);
459    if let Some((col, ord)) = order {
460        sql.push_str(&format!(" ORDER BY {} {}", col, ord.as_sql()));
461    }
462    if let Some(n) = limit {
463        sql.push_str(&format!(" LIMIT {}", n));
464    }
465    if let Some(n) = offset {
466        sql.push_str(&format!(" OFFSET {}", n));
467    }
468    (sql, params)
469}
470
471/// Build the WHERE clause. `start_idx` offsets PostgreSQL `$N` placeholders.
472#[doc(hidden)]
473pub fn build_where(
474    filters: Vec<(String, Vec<Value>)>,
475    start_idx: usize,
476    backend: Backend,
477) -> (String, Vec<Value>, usize) {
478    let mut all_params: Vec<Value> = Vec::new();
479    let mut conditions: Vec<String> = Vec::new();
480    let mut idx = start_idx;
481
482    for (mut fragment, params) in filters {
483        for param in params {
484            idx += 1;
485            if fragment.contains("__placeholder__") {
486                fragment = fragment.replacen("__placeholder__", &placeholder(backend, idx), 1);
487            } else {
488                // A raw fragment written with a literal `?` (e.g. from
489                // `.filter("age >= ?", ...)`) — no-op on backends that
490                // already use `?`, rewrites it to `$N` on PostgreSQL.
491                fragment = fragment.replacen("?", &placeholder(backend, idx), 1);
492            }
493            all_params.push(param);
494        }
495        conditions.push(fragment);
496    }
497
498    let where_clause = if conditions.is_empty() {
499        String::new()
500    } else {
501        format!(" WHERE {}", conditions.join(" AND "))
502    };
503
504    (where_clause, all_params, idx)
505}
rws_server/model/query.rs

rws_server/model/
query.rs
