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    let pk = entity.primary_key_value();
271    if T::primary_key_auto_increment() {
272        return match pk {
273            Value::Null | Value::Int(0) => insert_entity(pool, entity).await,
274            _ => update_entity(pool, entity).await,
275        };
276    }
277    match &pk {
278        Value::Null => Err(DbError::new(format!(
279            "save: {}.{} is NULL — set the key, or mark it #[primary_key(auto_increment)]",
280            T::table_name(),
281            T::primary_key_name()
282        ))),
283        Value::Text(s) if s.is_empty() => Err(DbError::new(format!(
284            "save: {}.{} is empty — set the key before saving",
285            T::table_name(),
286            T::primary_key_name()
287        ))),
288        key if is_nil_uuid(key) => insert_entity(pool, entity).await,
289        _ => {
290            let sql = format!("SELECT 1 FROM {} WHERE {} = {}", T::table_name(), T::primary_key_name(), placeholder(pool.backend(), 1));
291            if pool.query_rows(&sql, &[pk]).await?.is_empty() {
292                insert_entity(pool, entity).await
293            } else {
294                update_entity(pool, entity).await
295            }
296        }
297    }
298}
299
300#[doc(hidden)]
301pub async fn insert_entity<T: Model>(pool: &DbPool, entity: &T) -> Result<T, DbError> {
302    let backend = pool.backend();
303    let now = Timestamp::now().to_value(backend);
304    let pk_name = T::primary_key_name();
305    let mut key = entity.primary_key_value();
306
307    let mut cols: Vec<&str> = Vec::new();
308    let mut params: Vec<Value> = Vec::new();
309    let mut marks: Vec<String> = Vec::new();
310    for (col, value) in entity.to_values() {
311        let value = if col == pk_name {
312            if T::primary_key_auto_increment() {
313                continue;
314            }
315            if is_nil_uuid(&value) {
316                key = Value::Text(Uuid::now_v7().to_string());
317            }
318            key.clone()
319        } else if Some(col) == T::created_at_column() || Some(col) == T::updated_at_column() {
320            now.clone()
321        } else if Some(col) == T::soft_delete_column() {
322            Value::Null
323        } else if Some(col) == T::version_column() {
324            Value::Int(1)
325        } else {
326            value
327        };
328        cols.push(col);
329        if value == Value::Null {
330            // A literal: PostgreSQL won't assign a NULL bound as text to a
331            // column of another type.
332            marks.push("NULL".to_string());
333            continue;
334        }
335        params.push(value);
336        let pos = params.len();
337        marks.push(if is_timestamp_column::<T>(col) { timestamp_placeholder(backend, pos) } else { placeholder(backend, pos) });
338    }
339
340    let insert_sql = if cols.is_empty() {
341        match backend {
342            #[cfg(feature = "model-mysql")]
343            Backend::MySql => format!("INSERT INTO {} () VALUES ()", T::table_name()),
344            #[allow(unreachable_patterns)]
345            _ => format!("INSERT INTO {} DEFAULT VALUES", T::table_name()),
346        }
347    } else {
348        format!("INSERT INTO {} ({}) VALUES ({})", T::table_name(), cols.join(", "), marks.join(", "))
349    };
350
351    if T::primary_key_auto_increment() {
352        key = Value::Int(pool.insert_returning_id(&insert_sql, &params, pk_name).await?);
353    } else {
354        pool.execute(&insert_sql, &params).await?;
355    }
356    select_by_pk(pool, key, false).await?.ok_or_else(|| DbError::new("inserted row not found"))
357}
358
359#[doc(hidden)]
360pub async fn update_entity<T: Model>(pool: &DbPool, entity: &T) -> Result<T, DbError> {
361    let backend = pool.backend();
362    let pk = entity.primary_key_value();
363    let pk_name = T::primary_key_name();
364    let mut sets: Vec<String> = Vec::new();
365    let mut params: Vec<Value> = Vec::new();
366    let mut expected_version: Option<Value> = None;
367    for (col, value) in entity.to_values() {
368        if col == pk_name || Some(col) == T::created_at_column() || Some(col) == T::updated_at_column() || Some(col) == T::soft_delete_column() {
369            continue;
370        }
371        if Some(col) == T::version_column() {
372            expected_version = Some(value);
373            continue;
374        }
375        if value == Value::Null {
376            sets.push(format!("{col} = NULL"));
377            continue;
378        }
379        params.push(value);
380        sets.push(format!("{} = {}", col, placeholder(backend, params.len())));
381    }
382    if let Some(col) = T::updated_at_column() {
383        params.push(Timestamp::now().to_value(backend));
384        sets.push(format!("{} = {}", col, timestamp_placeholder(backend, params.len())));
385    }
386    if let Some(col) = T::version_column() {
387        sets.push(format!("{col} = {col} + 1"));
388    }
389    if sets.is_empty() {
390        // Nothing but the key: nothing to write.
391        return select_by_pk(pool, pk.clone(), false)
392            .await?
393            .ok_or_else(|| DbError::not_found(format!("{} {}={}: no such row to update", T::table_name(), pk_name, show(&pk))));
394    }
395
396    params.push(pk.clone());
397    let mut sql = format!(
398        "UPDATE {} SET {} WHERE {} = {}",
399        T::table_name(),
400        sets.join(", "),
401        pk_name,
402        placeholder(backend, params.len()),
403    );
404    if let (Some(col), Some(version)) = (T::version_column(), &expected_version) {
405        params.push(version.clone());
406        sql.push_str(&format!(" AND {} = {}", col, placeholder(backend, params.len())));
407    }
408
409    if pool.execute(&sql, &params).await? == 0 {
410        let current = select_by_pk::<T>(pool, pk.clone(), false).await?;
411        let (Some(row), Some(col), Some(expected)) = (current, T::version_column(), expected_version) else {
412            return Err(DbError::not_found(format!("{} {}={}: no such row to update", T::table_name(), pk_name, show(&pk))));
413        };
414        let now_at = row.to_values().into_iter().find(|(c, _)| *c == col).map(|(_, v)| v).unwrap_or(Value::Null);
415        return Err(DbError::stale_version(format!(
416            "{} {}={}: the row is at {} {}, this copy has {} — someone saved it since it was read; reload it and apply the change again",
417            T::table_name(),
418            pk_name,
419            show(&pk),
420            col,
421            show(&now_at),
422            show(&expected),
423        )));
424    }
425
426    select_by_pk(pool, pk, false).await?.ok_or_else(|| DbError::new("updated row not found"))
427}
428
429/// `delete_by_id`: `UPDATE … SET deleted_at` for a soft-delete model (rows
430/// deleted already keep their time), `DELETE` otherwise.
431#[doc(hidden)]
432pub async fn delete_entity<T: Model>(pool: &DbPool, pk: Value) -> Result<(), DbError> {
433    let backend = pool.backend();
434    match T::soft_delete_column() {
435        Some(col) => {
436            let sql = format!(
437                "UPDATE {} SET {col} = {} WHERE {} = {} AND {col} IS NULL",
438                T::table_name(),
439                timestamp_placeholder(backend, 1),
440                T::primary_key_name(),
441                placeholder(backend, 2),
442            );
443            pool.execute(&sql, &[Timestamp::now().to_value(backend), pk]).await?;
444        }
445        None => {
446            let sql = format!("DELETE FROM {} WHERE {} = {}", T::table_name(), T::primary_key_name(), placeholder(backend, 1));
447            pool.execute(&sql, &[pk]).await?;
448        }
449    }
450    Ok(())
451}
452
453fn show(v: &Value) -> String {
454    match v {
455        Value::Null => "NULL".to_string(),
456        Value::Bool(b) => b.to_string(),
457        Value::Int(n) => n.to_string(),
458        Value::Float(f) => f.to_string(),
459        Value::Text(s) => format!("{s:?}"),
460        Value::Bytes(b) => format!("<{} bytes>", b.len()),
461    }
462}
463
464#[doc(hidden)]
465pub fn extract_count(rows: Vec<super::ModelRow>) -> Result<i64, DbError> {
466    match rows.into_iter().next() {
467        Some(row) => {
468            if let Some((_, val)) = row.columns.first() {
469                match val.clone() {
470                    Value::Int(n) => return Ok(n),
471                    other => return Err(DbError::new(format!("unexpected count value: {:?}", other))),
472                }
473            }
474            Err(DbError::new("count query returned empty row"))
475        }
476        None => Ok(0),
477    }
478}
rws_server/model/repository.rs

rws_server/model/
repository.rs
