rws_server/model/
repository.rs

1//! Async `Repository` trait and `ModelRepository` implementation.
2//!
3//! # Primary keys (18.30.0)
4//!
5//! `ModelRepository<T, ID>` implements [`Repository<T, ID>`] for `ID` =
6//! `i64`, `i32`, `String` and [`Uuid`] (the [`PrimaryKey`] types). What
7//! [`save`](Repository::save) does depends on the key:
8//!
9//! - `#[primary_key(auto_increment)]`: a key of `0` (or `NULL`) inserts and
10//!   the database picks the key; any other key updates that row.
11//! - Any other key: `save` looks the key up (soft-deleted rows included)
12//!   and updates the row when it is there, inserts it otherwise. A
13//!   [`Uuid::NIL`] key (`Uuid::default()`) is replaced by a fresh
14//!   [`Uuid::now_v7`] and inserted; an empty `String` key is an error. The
15//!   look-up and the write are two statements: two processes saving the
16//!   same new key at once can both decide to insert, and the second gets
17//!   the database's unique-key error. [`ModelRepository::insert`] always
18//!   inserts.
19//!
20//! # Conventions
21//!
22//! The columns a model names through [`Model::created_at_column`],
23//! [`Model::updated_at_column`], [`Model::soft_delete_column`] and
24//! [`Model::version_column`] (`#[timestamps]`, `#[soft_delete]`,
25//! `#[version]` on `#[derive(Model)]`) are written by the repository, never
26//! from the entity's fields: on insert `created_at` = `updated_at` = now,
27//! `deleted_at` = `NULL`, `version` = 1; on update `updated_at` = now and
28//! `version` = `version + 1`, only while the row still has the entity's
29//! version — otherwise `save` fails with
30//! [`DbErrorKind::StaleVersion`](super::DbErrorKind::StaleVersion) and
31//! changes nothing. With a soft-delete column `delete_by_id` sets it,
32//! `find_by_id` / `find_all` / `count` / `exists_by_id` skip such rows, and
33//! [`ModelRepository::restore`] / [`ModelRepository::force_delete_by_id`]
34//! undo or finish it. Times are UTC ([`Timestamp`]); see
35//! [`timestamp`](super::timestamp) for the column types per backend.
36
37use std::marker::PhantomData;
38
39use super::backend::Backend;
40use super::pool::DbPool;
41use super::query::QueryBuilder;
42use super::timestamp::Timestamp;
43use super::uuid::Uuid;
44use super::{DbError, Model, ToColumn, Value};
45
46// ── Repository trait ──────────────────────────────────────────────────────────
47
48/// Async CRUD operations for a model type.
49pub trait Repository<T: Model, ID> {
50    async fn find_by_id(&self, id: ID) -> Result<Option<T>, DbError>;
51    async fn find_all(&self) -> Result<Vec<T>, DbError>;
52    async fn save(&self, entity: &T) -> Result<T, DbError>;
53    async fn save_all(&self, entities: &[T]) -> Result<Vec<T>, DbError>;
54    async fn delete_by_id(&self, id: ID) -> Result<(), DbError>;
55    async fn delete_all_by_id(&self, ids: &[ID]) -> Result<(), DbError>;
56    async fn count(&self) -> Result<i64, DbError>;
57    async fn exists_by_id(&self, id: ID) -> Result<bool, DbError>;
58}
59
60// ── PrimaryKey ────────────────────────────────────────────────────────────────
61
62mod sealed {
63    pub trait Sealed {}
64    impl Sealed for i64 {}
65    impl Sealed for i32 {}
66    impl Sealed for String {}
67    impl Sealed for super::Uuid {}
68}
69
70/// A type [`ModelRepository`] can key rows by: `i64`, `i32`, `String` or
71/// [`Uuid`] (18.30.0). Sealed — these are the types `Repository` is
72/// implemented for.
73pub trait PrimaryKey: ToColumn + Send + Sync + sealed::Sealed {}
74impl PrimaryKey for i64 {}
75impl PrimaryKey for i32 {}
76impl PrimaryKey for String {}
77impl PrimaryKey for Uuid {}
78
79// ── ModelRepository ───────────────────────────────────────────────────────────
80
81/// Repository tied to a model type and a pool.
82pub struct ModelRepository<'a, T: Model, ID> {
83    #[doc(hidden)]
84    pub pool: &'a DbPool,
85    _phantom: PhantomData<(T, ID)>,
86}
87
88impl<'a, T: Model, ID> ModelRepository<'a, T, ID> {
89    pub fn new(pool: &'a DbPool) -> Self {
90        ModelRepository { pool, _phantom: PhantomData }
91    }
92}
93
94impl<'a, T: Model, ID: PrimaryKey> ModelRepository<'a, T, ID> {
95    /// Insert `entity` (never an update, unlike [`Repository::save`]) and
96    /// answer the row as stored — with the generated key, timestamps and
97    /// version. A key that exists already is the database's unique-key
98    /// error.
99    pub async fn insert(&self, entity: &T) -> Result<T, DbError> {
100        insert_entity(self.pool, entity).await
101    }
102
103    /// The row with key `id`, soft-deleted or not.
104    pub async fn find_by_id_with_deleted(&self, id: ID) -> Result<Option<T>, DbError> {
105        select_by_pk(self.pool, id.to_column(), false).await
106    }
107
108    /// Clear the soft-delete column of row `id`: `true` when it was
109    /// deleted, `false` when it wasn't (or doesn't exist). An error for a
110    /// model without one.
111    pub async fn restore(&self, id: ID) -> Result<bool, DbError> {
112        let column = soft_delete_column_or_err::<T>("restore")?;
113        let backend = self.pool.backend();
114        let sql = format!(
115            "UPDATE {} SET {column} = NULL WHERE {} = {} AND {column} IS NOT NULL",
116            T::table_name(),
117            T::primary_key_name(),
118            placeholder(backend, 1),
119        );
120        Ok(self.pool.execute(&sql, &[id.to_column()]).await? > 0)
121    }
122
123    /// Delete row `id` for real (`DELETE`), soft-delete column or not.
124    pub async fn force_delete_by_id(&self, id: ID) -> Result<(), DbError> {
125        let sql = format!("DELETE FROM {} WHERE {} = {}", T::table_name(), T::primary_key_name(), placeholder(self.pool.backend(), 1));
126        self.pool.execute(&sql, &[id.to_column()]).await?;
127        Ok(())
128    }
129
130    /// A [`QueryBuilder`] over the same table and pool.
131    pub fn query(&self) -> QueryBuilder<'a, T> {
132        QueryBuilder::new(self.pool)
133    }
134}
135
136// ── impl Repository<T, ID> ────────────────────────────────────────────────────
137
138macro_rules! impl_repository {
139    ($($id:ty),*) => {$(
140        impl<'a, T: Model> Repository<T, $id> for ModelRepository<'a, T, $id> {
141            async fn find_by_id(&self, id: $id) -> Result<Option<T>, DbError> {
142                select_by_pk(self.pool, id.to_column(), true).await
143            }
144
145            async fn find_all(&self) -> Result<Vec<T>, DbError> {
146                let sql = format!("SELECT * FROM {}{}", T::table_name(), live_where::<T>());
147                let rows = self.pool.query_rows(&sql, &[]).await?;
148                rows.iter().map(|r| T::from_row(r)).collect()
149            }
150
151            async fn save(&self, entity: &T) -> Result<T, DbError> {
152                save_entity(self.pool, entity).await
153            }
154
155            async fn save_all(&self, entities: &[T]) -> Result<Vec<T>, DbError> {
156                let mut result = Vec::with_capacity(entities.len());
157                for e in entities {
158                    result.push(save_entity(self.pool, e).await?);
159                }
160                Ok(result)
161            }
162
163            async fn delete_by_id(&self, id: $id) -> Result<(), DbError> {
164                delete_entity::<T>(self.pool, id.to_column()).await
165            }
166
167            async fn delete_all_by_id(&self, ids: &[$id]) -> Result<(), DbError> {
168                for id in ids {
169                    delete_entity::<T>(self.pool, id.to_column()).await?;
170                }
171                Ok(())
172            }
173
174            async fn count(&self) -> Result<i64, DbError> {
175                let sql = format!("SELECT COUNT(*) FROM {}{}", T::table_name(), live_where::<T>());
176                let rows = self.pool.query_rows(&sql, &[]).await?;
177                extract_count(rows)
178            }
179
180            async fn exists_by_id(&self, id: $id) -> Result<bool, DbError> {
181                Ok(select_by_pk::<T>(self.pool, id.to_column(), true).await?.is_some())
182            }
183        }
184    )*};
185}
186
187impl_repository!(i64, i32, String, Uuid);
188
189// ── Placeholder helpers ───────────────────────────────────────────────────────
190
191/// Return the DB-appropriate placeholder for position `pos` (1-indexed) on
192/// `backend`. SQLite and MySQL use `?`, PostgreSQL uses `$1`, `$2`, …
193#[doc(hidden)]
194pub fn placeholder(backend: Backend, _pos: usize) -> String {
195    match backend {
196        #[cfg(feature = "model-sqlite")]
197        Backend::Sqlite => "?".to_string(),
198        #[cfg(feature = "model-postgres")]
199        Backend::Postgres => format!("${}", _pos),
200        #[cfg(feature = "model-mysql")]
201        Backend::MySql => "?".to_string(),
202    }
203}
204
205/// Build a comma-separated list of placeholders.
206#[doc(hidden)]
207pub fn placeholders(backend: Backend, count: usize) -> String {
208    (1..=count).map(|i| placeholder(backend, i)).collect::<Vec<_>>().join(", ")
209}
210
211/// The placeholder for a timestamp parameter: rws binds text, which
212/// PostgreSQL only assigns to a `TIMESTAMPTZ` column through a cast (and
213/// assigns the cast's result to a text column too).
214#[doc(hidden)]
215pub fn timestamp_placeholder(backend: Backend, pos: usize) -> String {
216    match backend {
217        #[cfg(feature = "model-postgres")]
218        Backend::Postgres => format!("CAST(${} AS TIMESTAMPTZ)", pos),
219        #[allow(unreachable_patterns)]
220        _ => placeholder(backend, pos),
221    }
222}
223
224/// `col IS NULL` for a soft-delete model (the rows not deleted).
225#[doc(hidden)]
226pub fn live_condition<T: Model>() -> Option<String> {
227    T::soft_delete_column().map(|c| format!("{c} IS NULL"))
228}
229
230fn live_where<T: Model>() -> String {
231    live_condition::<T>().map(|c| format!(" WHERE {c}")).unwrap_or_default()
232}
233
234fn soft_delete_column_or_err<T: Model>(what: &str) -> Result<&'static str, DbError> {
235    T::soft_delete_column().ok_or_else(|| {
236        DbError::new(format!("{what}: {} has no soft-delete column (add #[soft_delete] and a deleted_at field)", T::table_name()))
237    })
238}
239
240fn is_timestamp_column<T: Model>(col: &str) -> bool {
241    [T::created_at_column(), T::updated_at_column(), T::soft_delete_column()].contains(&Some(col))
242}
243
244fn is_nil_uuid(v: &Value) -> bool {
245    matches!(v, Value::Text(s) if s == "00000000-0000-0000-0000-000000000000")
246}
247
248// ── Insert / Update helpers ───────────────────────────────────────────────────
249
250/// The row with primary key `pk` (`live_only`: not soft-deleted).
251#[doc(hidden)]
252pub async fn select_by_pk<T: Model>(pool: &DbPool, pk: Value, live_only: bool) -> Result<Option<T>, DbError> {
253    let mut sql = format!("SELECT * FROM {} WHERE {} = {}", T::table_name(), T::primary_key_name(), placeholder(pool.backend(), 1));
254    if live_only {
255        if let Some(c) = live_condition::<T>() {
256            sql.push_str(" AND ");
257            sql.push_str(&c);
258        }
259    }
260    let rows = pool.query_rows(&sql, &[pk]).await?;
261    match rows.into_iter().next() {
262        Some(row) => Ok(Some(T::from_row(&row)?)),
263        None => Ok(None),
264    }
265}
266
267/// `save`: insert or update, as the module docs describe.
268#[doc(hidden)]
269pub async fn save_entity<T: Model>(pool: &DbPool, entity: &T) -> Result<T, DbError> {
270    // On the primary throughout: whether the row exists, and the row as
271    // saved, must not come from a replica that lags behind.
272    let pool = pool.primary();
273    let pk = entity.primary_key_value();
274    if T::primary_key_auto_increment() {
275        return match pk {
276            Value::Null | Value::Int(0) => insert_entity(pool, entity).await,
277            _ => update_entity(pool, entity).await,
278        };
279    }
280    match &pk {
281        Value::Null => Err(DbError::new(format!(
282            "save: {}.{} is NULL — set the key, or mark it #[primary_key(auto_increment)]",
283            T::table_name(),
284            T::primary_key_name()
285        ))),
286        Value::Text(s) if s.is_empty() => Err(DbError::new(format!(
287            "save: {}.{} is empty — set the key before saving",
288            T::table_name(),
289            T::primary_key_name()
290        ))),
291        key if is_nil_uuid(key) => insert_entity(pool, entity).await,
292        _ => {
293            let sql = format!("SELECT 1 FROM {} WHERE {} = {}", T::table_name(), T::primary_key_name(), placeholder(pool.backend(), 1));
294            if pool.query_rows(&sql, &[pk]).await?.is_empty() {
295                insert_entity(pool, entity).await
296            } else {
297                update_entity(pool, entity).await
298            }
299        }
300    }
301}
302
303#[doc(hidden)]
304pub async fn insert_entity<T: Model>(pool: &DbPool, entity: &T) -> Result<T, DbError> {
305    let pool = pool.primary();
306    let backend = pool.backend();
307    let now = Timestamp::now().to_value(backend);
308    let pk_name = T::primary_key_name();
309    let mut key = entity.primary_key_value();
310
311    let mut cols: Vec<&str> = Vec::new();
312    let mut params: Vec<Value> = Vec::new();
313    let mut marks: Vec<String> = Vec::new();
314    for (col, value) in entity.to_values() {
315        let value = if col == pk_name {
316            if T::primary_key_auto_increment() {
317                continue;
318            }
319            if is_nil_uuid(&value) {
320                key = Value::Text(Uuid::now_v7().to_string());
321            }
322            key.clone()
323        } else if Some(col) == T::created_at_column() || Some(col) == T::updated_at_column() {
324            now.clone()
325        } else if Some(col) == T::soft_delete_column() {
326            Value::Null
327        } else if Some(col) == T::version_column() {
328            Value::Int(1)
329        } else {
330            value
331        };
332        cols.push(col);
333        if value == Value::Null {
334            // A literal: PostgreSQL won't assign a NULL bound as text to a
335            // column of another type.
336            marks.push("NULL".to_string());
337            continue;
338        }
339        params.push(value);
340        let pos = params.len();
341        marks.push(if is_timestamp_column::<T>(col) { timestamp_placeholder(backend, pos) } else { placeholder(backend, pos) });
342    }
343
344    let insert_sql = if cols.is_empty() {
345        match backend {
346            #[cfg(feature = "model-mysql")]
347            Backend::MySql => format!("INSERT INTO {} () VALUES ()", T::table_name()),
348            #[allow(unreachable_patterns)]
349            _ => format!("INSERT INTO {} DEFAULT VALUES", T::table_name()),
350        }
351    } else {
352        format!("INSERT INTO {} ({}) VALUES ({})", T::table_name(), cols.join(", "), marks.join(", "))
353    };
354
355    if T::primary_key_auto_increment() {
356        key = Value::Int(pool.insert_returning_id(&insert_sql, &params, pk_name).await?);
357    } else {
358        pool.execute(&insert_sql, &params).await?;
359    }
360    select_by_pk(pool, key, false).await?.ok_or_else(|| DbError::new("inserted row not found"))
361}
362
363#[doc(hidden)]
364pub async fn update_entity<T: Model>(pool: &DbPool, entity: &T) -> Result<T, DbError> {
365    let pool = pool.primary();
366    let backend = pool.backend();
367    let pk = entity.primary_key_value();
368    let pk_name = T::primary_key_name();
369    let mut sets: Vec<String> = Vec::new();
370    let mut params: Vec<Value> = Vec::new();
371    let mut expected_version: Option<Value> = None;
372    for (col, value) in entity.to_values() {
373        if col == pk_name || Some(col) == T::created_at_column() || Some(col) == T::updated_at_column() || Some(col) == T::soft_delete_column() {
374            continue;
375        }
376        if Some(col) == T::version_column() {
377            expected_version = Some(value);
378            continue;
379        }
380        if value == Value::Null {
381            sets.push(format!("{col} = NULL"));
382            continue;
383        }
384        params.push(value);
385        sets.push(format!("{} = {}", col, placeholder(backend, params.len())));
386    }
387    if let Some(col) = T::updated_at_column() {
388        params.push(Timestamp::now().to_value(backend));
389        sets.push(format!("{} = {}", col, timestamp_placeholder(backend, params.len())));
390    }
391    if let Some(col) = T::version_column() {
392        sets.push(format!("{col} = {col} + 1"));
393    }
394    if sets.is_empty() {
395        // Nothing but the key: nothing to write.
396        return select_by_pk(pool, pk.clone(), false)
397            .await?
398            .ok_or_else(|| DbError::not_found(format!("{} {}={}: no such row to update", T::table_name(), pk_name, show(&pk))));
399    }
400
401    params.push(pk.clone());
402    let mut sql = format!(
403        "UPDATE {} SET {} WHERE {} = {}",
404        T::table_name(),
405        sets.join(", "),
406        pk_name,
407        placeholder(backend, params.len()),
408    );
409    if let (Some(col), Some(version)) = (T::version_column(), &expected_version) {
410        params.push(version.clone());
411        sql.push_str(&format!(" AND {} = {}", col, placeholder(backend, params.len())));
412    }
413
414    if pool.execute(&sql, &params).await? == 0 {
415        let current = select_by_pk::<T>(pool, pk.clone(), false).await?;
416        let (Some(row), Some(col), Some(expected)) = (current, T::version_column(), expected_version) else {
417            return Err(DbError::not_found(format!("{} {}={}: no such row to update", T::table_name(), pk_name, show(&pk))));
418        };
419        let now_at = row.to_values().into_iter().find(|(c, _)| *c == col).map(|(_, v)| v).unwrap_or(Value::Null);
420        return Err(DbError::stale_version(format!(
421            "{} {}={}: the row is at {} {}, this copy has {} — someone saved it since it was read; reload it and apply the change again",
422            T::table_name(),
423            pk_name,
424            show(&pk),
425            col,
426            show(&now_at),
427            show(&expected),
428        )));
429    }
430
431    select_by_pk(pool, pk, false).await?.ok_or_else(|| DbError::new("updated row not found"))
432}
433
434/// `delete_by_id`: `UPDATE … SET deleted_at` for a soft-delete model (rows
435/// deleted already keep their time), `DELETE` otherwise.
436#[doc(hidden)]
437pub async fn delete_entity<T: Model>(pool: &DbPool, pk: Value) -> Result<(), DbError> {
438    let backend = pool.backend();
439    match T::soft_delete_column() {
440        Some(col) => {
441            let sql = format!(
442                "UPDATE {} SET {col} = {} WHERE {} = {} AND {col} IS NULL",
443                T::table_name(),
444                timestamp_placeholder(backend, 1),
445                T::primary_key_name(),
446                placeholder(backend, 2),
447            );
448            pool.execute(&sql, &[Timestamp::now().to_value(backend), pk]).await?;
449        }
450        None => {
451            let sql = format!("DELETE FROM {} WHERE {} = {}", T::table_name(), T::primary_key_name(), placeholder(backend, 1));
452            pool.execute(&sql, &[pk]).await?;
453        }
454    }
455    Ok(())
456}
457
458fn show(v: &Value) -> String {
459    match v {
460        Value::Null => "NULL".to_string(),
461        Value::Bool(b) => b.to_string(),
462        Value::Int(n) => n.to_string(),
463        Value::Float(f) => f.to_string(),
464        Value::Text(s) => format!("{s:?}"),
465        Value::Bytes(b) => format!("<{} bytes>", b.len()),
466    }
467}
468
469#[doc(hidden)]
470pub fn extract_count(rows: Vec<super::ModelRow>) -> Result<i64, DbError> {
471    match rows.into_iter().next() {
472        Some(row) => {
473            if let Some((_, val)) = row.columns.first() {
474                match val.clone() {
475                    Value::Int(n) => return Ok(n),
476                    other => return Err(DbError::new(format!("unexpected count value: {:?}", other))),
477                }
478            }
479            Err(DbError::new("count query returned empty row"))
480        }
481        None => Ok(0),
482    }
483}
rws_server/model/repository.rs

rws_server/model/
repository.rs
