1use async_trait::async_trait;
7use chrono::{DateTime, Utc};
8use serde::{de::DeserializeOwned, Serialize};
9use std::collections::HashMap;
10use std::fmt::Debug;
11
12#[derive(Debug, thiserror::Error)]
18pub enum RepositoryError {
19 #[error("Entity not found")]
20 NotFound,
21
22 #[error("Entity already exists: {0}")]
23 AlreadyExists(String),
24
25 #[error("Validation error: {0}")]
26 ValidationError(String),
27
28 #[error("Database error: {0}")]
29 DatabaseError(String),
30
31 #[error("Serialization error: {0}")]
32 SerializationError(String),
33
34 #[error("Conflict: {0}")]
35 Conflict(String),
36
37 #[error("Internal error: {0}")]
38 InternalError(String),
39
40 #[error("validation failed: {}", .0.iter().map(|v| v.message.as_str()).collect::<Vec<_>>().join("; "))]
44 Violations(Vec<crate::violation::Violation>),
45}
46
47impl RepositoryError {
48 pub fn from_source(e: impl Into<anyhow::Error>) -> Self {
53 let e: anyhow::Error = e.into();
54 #[cfg(feature = "postgres")]
55 let e = match e.downcast::<sqlx::Error>() {
56 Ok(db) => return RepositoryError::from(db),
57 Err(e) => e,
58 };
59 RepositoryError::DatabaseError(e.to_string())
60 }
61}
62
63impl From<serde_json::Error> for RepositoryError {
64 fn from(e: serde_json::Error) -> Self {
65 RepositoryError::SerializationError(e.to_string())
66 }
67}
68
69#[cfg(feature = "postgres")]
70impl From<sqlx::Error> for RepositoryError {
71 fn from(e: sqlx::Error) -> Self {
72 match e {
73 sqlx::Error::RowNotFound => RepositoryError::NotFound,
74 sqlx::Error::Database(db_err) => {
75 let pg = db_err.try_downcast_ref::<sqlx::postgres::PgDatabaseError>();
76 if let Some(v) = constraint_violation(
77 db_err.kind(),
78 db_err.code().as_deref(),
79 db_err.constraint(),
80 pg.and_then(|p| p.detail()).unwrap_or(""),
81 pg.and_then(|p| p.column()),
82 ) {
83 return RepositoryError::Violations(vec![v]);
84 }
85 let msg = db_err.message().to_string();
86 if msg.contains("duplicate key") || msg.contains("unique constraint") {
87 RepositoryError::AlreadyExists(msg)
88 } else {
89 RepositoryError::DatabaseError(msg)
90 }
91 }
92 _ => RepositoryError::DatabaseError(e.to_string()),
93 }
94 }
95}
96
97#[cfg(feature = "postgres")]
105fn constraint_violation(
106 kind: sqlx::error::ErrorKind,
107 sqlstate: Option<&str>,
108 constraint: Option<&str>,
109 detail: &str,
110 column: Option<&str>,
111) -> Option<crate::violation::Violation> {
112 use crate::violation::{code_from_constraint, Violation};
113 use sqlx::error::ErrorKind;
114 let constraint = constraint?;
115 let path = column.map(str::to_string).or_else(|| key_column(detail)).unwrap_or_default();
116 let code = code_from_constraint(constraint);
117 let on = |path: &str| if path.is_empty() { String::new() } else { format!("`{path}` ") };
118 Some(match kind {
119 ErrorKind::ForeignKeyViolation if detail.contains("still referenced") => Violation::new(
120 "",
121 "still_referenced",
122 "This record is still referenced by another record, so it cannot be removed",
123 ),
124 ErrorKind::ForeignKeyViolation => {
125 let message = format!("{}refers to a record that does not exist", on(&path));
126 Violation::new(path, "reference_missing", message)
127 }
128 ErrorKind::CheckViolation => {
129 let message = format!("{}breaks the rule `{code}`", on(&path));
130 Violation::new(path, code, message)
131 }
132 ErrorKind::UniqueViolation if constraint.starts_with("uq_") => {
133 let message = format!("{}is already taken under the rule `{code}`", on(&path));
134 Violation::new(path, code, message)
135 }
136 _ if sqlstate == Some("23P01") => {
137 let message = format!("The record overlaps another one under the rule `{code}`");
138 Violation::new("", code, message)
139 }
140 _ => return None,
141 })
142}
143
144#[cfg(feature = "postgres")]
147fn key_column(detail: &str) -> Option<String> {
148 let rest = detail.strip_prefix("Key (")?;
149 let cols = &rest[..rest.find(")=")?];
150 (!cols.contains(',')).then(|| cols.trim().to_string())
151}
152
153#[cfg(all(test, feature = "postgres"))]
154mod constraint_violation_tests {
155 use super::*;
156 use sqlx::error::ErrorKind;
157
158 #[test]
159 fn a_missing_reference_names_the_column_and_never_the_table() {
160 let v = constraint_violation(
161 ErrorKind::ForeignKeyViolation,
162 Some("23503"),
163 Some("fk_subdistricts_district_id"),
164 "Key (district_id)=(1484f475-14be-4abc-985b-7291543d32bc) is not present in table \"districts\".",
165 None,
166 )
167 .unwrap();
168 assert_eq!((v.path.as_str(), v.code.as_str()), ("district_id", "reference_missing"));
169 assert!(!v.message.contains("districts\"") && !v.message.contains("fk_"), "{}", v.message);
170 }
171
172 #[test]
173 fn removing_a_referenced_record_is_still_referenced() {
174 let v = constraint_violation(
175 ErrorKind::ForeignKeyViolation,
176 Some("23503"),
177 Some("fk_provinces_country_id"),
178 "Key (id)=(866f1b4e-a9f3-4626-95d1-13c183916b4b) is still referenced from table \"provinces\".",
179 None,
180 )
181 .unwrap();
182 assert_eq!((v.path.as_str(), v.code.as_str()), ("", "still_referenced"));
183 }
184
185 #[test]
186 fn a_rule_constraint_answers_its_rules_code() {
187 let check = constraint_violation(ErrorKind::CheckViolation, Some("23514"), Some("ck_countries_name__max_length"), "", Some("name")).unwrap();
188 assert_eq!((check.path.as_str(), check.code.as_str()), ("name", "max_length"));
189 let unique = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_accounts__account_code_taken"), "Key (company_id, code)=(a, b) already exists.", None).unwrap();
190 assert_eq!((unique.path.as_str(), unique.code.as_str()), ("", "account_code_taken"));
191 let overlap = constraint_violation(ErrorKind::Other, Some("23P01"), Some("ex_tax_rates__overlapping_validity"), "", None).unwrap();
192 assert_eq!(overlap.code, "overlapping_validity");
193 }
194
195 #[test]
196 fn a_database_error_inside_any_error_converts_as_the_database_error_does() {
197 let wrapped = anyhow::Error::new(sqlx::Error::RowNotFound).context("loading the row");
198 assert!(matches!(RepositoryError::from_source(wrapped), RepositoryError::NotFound));
199 assert!(matches!(RepositoryError::from_source(sqlx::Error::RowNotFound), RepositoryError::NotFound));
200 let other = RepositoryError::from_source(anyhow::anyhow!("pool closed"));
201 assert!(matches!(other, RepositoryError::DatabaseError(ref m) if m == "pool closed"), "{other:?}");
202 }
203
204 #[test]
205 fn an_error_that_names_no_rule_keeps_its_old_answer() {
206 assert!(constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("users_email_key"), "", None).is_none());
207 assert!(constraint_violation(ErrorKind::Other, Some("42P01"), None, "", None).is_none());
208 assert!(constraint_violation(ErrorKind::NotNullViolation, Some("23502"), Some("x"), "", None).is_none());
209 }
210}
211
212pub trait PersistentEntity: Clone + Send + Sync + Debug + Serialize + DeserializeOwned + 'static {
221 fn entity_id(&self) -> String;
223
224 fn set_entity_id(&mut self, id: String);
226
227 fn created_at(&self) -> Option<DateTime<Utc>>;
229
230 fn set_created_at(&mut self, ts: DateTime<Utc>);
232
233 fn updated_at(&self) -> Option<DateTime<Utc>>;
235
236 fn set_updated_at(&mut self, ts: DateTime<Utc>);
238
239 fn deleted_at(&self) -> Option<DateTime<Utc>>;
241
242 fn set_deleted_at(&mut self, ts: Option<DateTime<Utc>>);
244
245 fn is_deleted(&self) -> bool {
247 self.deleted_at().is_some()
248 }
249
250 fn mark_deleted(&mut self) {
252 self.set_deleted_at(Some(Utc::now()));
253 self.set_updated_at(Utc::now());
254 }
255
256 fn restore(&mut self) {
258 self.set_deleted_at(None);
259 self.set_updated_at(Utc::now());
260 }
261
262 fn touch(&mut self) {
264 self.set_updated_at(Utc::now());
265 }
266
267 fn generate_id() -> String {
269 uuid::Uuid::new_v4().to_string()
270 }
271
272 fn write_protected_fields() -> &'static [&'static str] {
283 &[]
284 }
285}
286
287pub trait PartialUpdatable: PersistentEntity {
289 fn apply_partial_update(&mut self, fields: &HashMap<String, serde_json::Value>) -> Result<(), RepositoryError>;
291}
292
293pub trait Versioned {
295 fn version(&self) -> u64;
296 fn set_version(&mut self, version: u64);
297 fn increment_version(&mut self) {
298 self.set_version(self.version() + 1);
299 }
300}
301
302#[async_trait]
311pub trait CrudRepository<E>: Send + Sync
312where
313 E: PersistentEntity,
314{
315 async fn create(&self, entity: E) -> Result<E, RepositoryError>;
317
318 async fn find_by_id(&self, id: &str) -> Result<Option<E>, RepositoryError>;
320
321 async fn find_by_id_including_deleted(&self, id: &str) -> Result<Option<E>, RepositoryError>;
323
324 async fn update(&self, entity: E) -> Result<E, RepositoryError>;
326
327 async fn soft_delete(&self, id: &str) -> Result<bool, RepositoryError>;
329
330 async fn restore(&self, id: &str) -> Result<Option<E>, RepositoryError>;
332
333 async fn hard_delete(&self, id: &str) -> Result<bool, RepositoryError>;
335
336 async fn list(&self, page: u32, limit: u32) -> Result<(Vec<E>, u64), RepositoryError>;
338
339 async fn list_deleted(&self, page: u32, limit: u32) -> Result<(Vec<E>, u64), RepositoryError>;
341
342 async fn count(&self) -> Result<u64, RepositoryError>;
344
345 async fn count_deleted(&self) -> Result<u64, RepositoryError>;
347
348 async fn bulk_create(&self, entities: Vec<E>) -> Result<Vec<E>, RepositoryError>;
350
351 async fn empty_trash(&self) -> Result<u64, RepositoryError>;
353
354 async fn list_filtered(
359 &self,
360 page: u32,
361 limit: u32,
362 filters: HashMap<String, String>,
363 ) -> Result<(Vec<E>, u64), RepositoryError> {
364 let _ = filters; self.list(page, limit).await
366 }
367
368 async fn list_filtered_with_info(
373 &self,
374 page: u32,
375 limit: u32,
376 filters: HashMap<String, String>,
377 ) -> Result<(Vec<E>, backbone_orm::repository::PaginationInfo), RepositoryError> {
378 let (rows, total) = self.list_filtered(page, limit, filters).await?;
379 Ok((
380 rows,
381 backbone_orm::repository::PaginationInfo::new(page, limit, total),
382 ))
383 }
384
385 fn table_name(&self) -> Option<&str> {
397 None
398 }
399
400 async fn aggregate_filtered(
401 &self,
402 spec: &backbone_orm::repository::AggregateSpec,
403 filters: HashMap<String, String>,
404 ) -> Result<backbone_orm::repository::AggregateResult, RepositoryError> {
405 let _ = (spec, filters);
406 Err(RepositoryError::DatabaseError(
407 "aggregate is not supported by this repository".to_string(),
408 ))
409 }
410
411 async fn fetch_related_json(
416 &self,
417 _table: &str,
418 _ids: &[String],
419 ) -> Vec<serde_json::Value> {
420 Vec::new()
421 }
422
423 async fn exists(&self, id: &str) -> Result<bool, RepositoryError> {
425 Ok(self.find_by_id(id).await?.is_some())
426 }
427
428 async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64, RepositoryError> {
438 let mut n = 0;
439 for id in ids {
440 if self.soft_delete(id).await? {
441 n += 1;
442 }
443 }
444 Ok(n)
445 }
446
447 async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<E>, RepositoryError> {
449 let mut out = Vec::with_capacity(ids.len());
450 for id in ids {
451 if let Some(e) = self.restore(id).await? {
452 out.push(e);
453 }
454 }
455 Ok(out)
456 }
457
458 async fn bulk_hard_delete(&self, ids: &[String]) -> Result<u64, RepositoryError> {
460 let mut n = 0;
461 for id in ids {
462 if self.hard_delete(id).await? {
463 n += 1;
464 }
465 }
466 Ok(n)
467 }
468
469 async fn restore_all(&self) -> Result<Vec<E>, RepositoryError> {
471 let mut restored = Vec::new();
472 loop {
475 let (batch, _) = self.list_deleted(1, 500).await?;
476 if batch.is_empty() {
477 break;
478 }
479 for entity in &batch {
480 if let Some(e) = self.restore(&entity.entity_id()).await? {
481 restored.push(e);
482 }
483 }
484 }
485 Ok(restored)
486 }
487
488 async fn bulk_update(&self, entities: Vec<E>) -> Result<Vec<E>, RepositoryError> {
490 let mut out = Vec::with_capacity(entities.len());
491 for entity in entities {
492 out.push(self.update(entity).await?);
493 }
494 Ok(out)
495 }
496}
497
498#[async_trait]
500pub trait SearchableRepository<E>: CrudRepository<E>
501where
502 E: PersistentEntity,
503{
504 async fn search(
506 &self,
507 filters: HashMap<String, String>,
508 page: u32,
509 limit: u32,
510 ) -> Result<(Vec<E>, u64), RepositoryError>;
511
512 async fn find_by_field(&self, field: &str, value: &str) -> Result<Option<E>, RepositoryError>;
514
515 async fn find_all_by_field(
517 &self,
518 field: &str,
519 value: &str,
520 page: u32,
521 limit: u32,
522 ) -> Result<(Vec<E>, u64), RepositoryError>;
523}
524
525#[cfg(feature = "postgres")]
530pub use postgres_traits::*;
531
532#[cfg(feature = "postgres")]
533mod postgres_traits {
534 use super::*;
535 use sqlx::postgres::PgRow;
536 use sqlx::FromRow;
537
538 pub trait PostgresEntity: PersistentEntity + for<'r> FromRow<'r, PgRow> + Unpin {
576 fn table_name() -> &'static str;
578
579 fn id_column() -> &'static str {
581 "id"
582 }
583
584 fn select_columns() -> &'static [&'static str];
586
587 fn insert_columns() -> &'static [&'static str];
589
590 fn update_columns() -> &'static [&'static str] {
592 Self::insert_columns()
593 }
594
595 fn bind_for_insert<'q>(
599 entity: &'q Self,
600 query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
601 ) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>;
602
603 fn bind_for_update<'q>(
607 entity: &'q Self,
608 query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
609 ) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> {
610 Self::bind_for_insert(entity, query)
612 }
613
614 fn select_by_id_query() -> String {
616 format!(
617 "SELECT {} FROM {} WHERE {} = $1 AND deleted_at IS NULL",
618 Self::select_columns().join(", "),
619 Self::table_name(),
620 Self::id_column()
621 )
622 }
623
624 fn select_by_id_including_deleted_query() -> String {
626 format!(
627 "SELECT {} FROM {} WHERE {} = $1",
628 Self::select_columns().join(", "),
629 Self::table_name(),
630 Self::id_column()
631 )
632 }
633
634 fn list_query() -> String {
636 format!(
637 "SELECT {} FROM {} WHERE deleted_at IS NULL ORDER BY created_at DESC LIMIT $1 OFFSET $2",
638 Self::select_columns().join(", "),
639 Self::table_name()
640 )
641 }
642
643 fn count_query() -> String {
645 format!(
646 "SELECT COUNT(*) FROM {} WHERE deleted_at IS NULL",
647 Self::table_name()
648 )
649 }
650
651 fn list_deleted_query() -> String {
653 format!(
654 "SELECT {} FROM {} WHERE deleted_at IS NOT NULL ORDER BY deleted_at DESC LIMIT $1 OFFSET $2",
655 Self::select_columns().join(", "),
656 Self::table_name()
657 )
658 }
659
660 fn count_deleted_query() -> String {
662 format!(
663 "SELECT COUNT(*) FROM {} WHERE deleted_at IS NOT NULL",
664 Self::table_name()
665 )
666 }
667
668 fn soft_delete_query() -> String {
670 format!(
671 "UPDATE {} SET deleted_at = NOW(), updated_at = NOW() WHERE {} = $1 AND deleted_at IS NULL",
672 Self::table_name(),
673 Self::id_column()
674 )
675 }
676
677 fn restore_query() -> String {
679 format!(
680 "UPDATE {} SET deleted_at = NULL, updated_at = NOW() WHERE {} = $1 AND deleted_at IS NOT NULL RETURNING {}",
681 Self::table_name(),
682 Self::id_column(),
683 Self::select_columns().join(", ")
684 )
685 }
686
687 fn hard_delete_query() -> String {
689 format!(
690 "DELETE FROM {} WHERE {} = $1",
691 Self::table_name(),
692 Self::id_column()
693 )
694 }
695
696 fn empty_trash_query() -> String {
698 format!(
699 "DELETE FROM {} WHERE deleted_at IS NOT NULL",
700 Self::table_name()
701 )
702 }
703 }
704}