use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{de::DeserializeOwned, Serialize};
use std::collections::HashMap;
use std::fmt::Debug;
#[derive(Debug, thiserror::Error)]
pub enum RepositoryError {
#[error("Entity not found")]
NotFound,
#[error("Entity already exists: {0}")]
AlreadyExists(String),
#[error("Validation error: {0}")]
ValidationError(String),
#[error("Database error: {0}")]
DatabaseError(String),
#[error("Serialization error: {0}")]
SerializationError(String),
#[error("Conflict: {0}")]
Conflict(String),
#[error("Internal error: {0}")]
InternalError(String),
#[error("validation failed: {}", .0.iter().map(|v| v.message.as_str()).collect::<Vec<_>>().join("; "))]
Violations(Vec<crate::violation::Violation>),
}
impl RepositoryError {
pub fn from_source(e: impl Into<anyhow::Error>) -> Self {
let e: anyhow::Error = e.into();
let e = match e.downcast::<backbone_orm::QueryParamError>() {
Ok(refused) => {
return RepositoryError::Violations(vec![crate::violation::Violation::invalid_query(
&refused.parameter,
refused.reason.as_str(),
refused.message,
)])
}
Err(e) => e,
};
#[cfg(feature = "postgres")]
let e = match e.downcast::<sqlx::Error>() {
Ok(db) => return RepositoryError::from(db),
Err(e) => e,
};
RepositoryError::DatabaseError(e.to_string())
}
pub fn from_batch_source(e: impl Into<anyhow::Error>) -> Self {
let e: anyhow::Error = e.into();
match e.downcast::<backbone_orm::BatchItemError>() {
Ok(item) => Self::from_source(item.error).at_batch_index(item.index),
Err(e) => Self::from_source(e),
}
}
pub fn at_batch_index(self, index: usize) -> Self {
let item = |message: String| format!("item {index}: {message}");
match self {
RepositoryError::Violations(violations) => RepositoryError::Violations(
violations
.into_iter()
.map(|mut v| {
v.path = crate::violation::batch_item_path(index, &v.path);
v
})
.collect(),
),
RepositoryError::NotFound => RepositoryError::ValidationError(item(
"the record no longer exists or was deleted".to_string(),
)),
RepositoryError::AlreadyExists(m) => RepositoryError::AlreadyExists(item(m)),
RepositoryError::ValidationError(m) => RepositoryError::ValidationError(item(m)),
RepositoryError::DatabaseError(m) => RepositoryError::DatabaseError(item(m)),
RepositoryError::SerializationError(m) => RepositoryError::SerializationError(item(m)),
RepositoryError::Conflict(m) => RepositoryError::Conflict(item(m)),
RepositoryError::InternalError(m) => RepositoryError::InternalError(item(m)),
}
}
}
impl From<serde_json::Error> for RepositoryError {
fn from(e: serde_json::Error) -> Self {
RepositoryError::SerializationError(e.to_string())
}
}
#[cfg(feature = "postgres")]
impl From<sqlx::Error> for RepositoryError {
fn from(e: sqlx::Error) -> Self {
match e {
sqlx::Error::RowNotFound => RepositoryError::NotFound,
sqlx::Error::Database(db_err) => {
if let Some(v) = scope_refusal(db_err.code().as_deref(), db_err.message()) {
return RepositoryError::Violations(vec![v]);
}
let pg = db_err.try_downcast_ref::<sqlx::postgres::PgDatabaseError>();
if let Some(v) = constraint_violation(
db_err.kind(),
db_err.code().as_deref(),
db_err.constraint(),
pg.and_then(|p| p.detail()).unwrap_or(""),
pg.and_then(|p| p.column()),
pg.and_then(|p| p.table()),
) {
return RepositoryError::Violations(vec![v]);
}
let msg = db_err.message().to_string();
if msg.contains("duplicate key") || msg.contains("unique constraint") {
RepositoryError::Violations(vec![taken_violation("", "taken")])
} else {
RepositoryError::DatabaseError(msg)
}
}
_ => RepositoryError::DatabaseError(e.to_string()),
}
}
}
#[cfg(feature = "postgres")]
fn scope_refusal(sqlstate: Option<&str>, message: &str) -> Option<crate::violation::Violation> {
(sqlstate == Some("42501") && message.contains("row-level security policy")).then(|| {
crate::violation::Violation::new(
"",
"scope_refused",
"The record is outside the company or unit this request may write to",
)
})
}
#[cfg(feature = "postgres")]
fn taken_violation(path: &str, code: &str) -> crate::violation::Violation {
let message = match path {
"" => "This value is already taken by another record".to_string(),
p => format!("{p} is already taken by another record"),
};
crate::violation::Violation::new(path, code, message)
}
#[cfg(feature = "postgres")]
fn constraint_violation(
kind: sqlx::error::ErrorKind,
sqlstate: Option<&str>,
constraint: Option<&str>,
detail: &str,
column: Option<&str>,
table: Option<&str>,
) -> Option<crate::violation::Violation> {
use crate::violation::{code_from_constraint, Violation};
use sqlx::error::ErrorKind;
if kind == ErrorKind::NotNullViolation || sqlstate == Some("23502") {
let column = column.filter(|c| !c.is_empty())?;
return Some(Violation::new(
column,
"required",
format!("{column} is required"),
));
}
if kind == ErrorKind::UniqueViolation || sqlstate == Some("23505") {
let generated = constraint.and_then(generated_code);
let path = column
.filter(|c| !c.is_empty())
.map(str::to_string)
.or_else(|| key_column(detail))
.or_else(|| {
let (c, t) = (constraint?, table?);
constraint_column(c, t, generated.as_deref().unwrap_or(""))
})
.unwrap_or_default();
return Some(taken_violation(&path, generated.as_deref().unwrap_or("taken")));
}
let constraint = constraint?;
let code = code_from_constraint(constraint);
let path = column
.filter(|c| !c.is_empty())
.map(str::to_string)
.or_else(|| key_column(detail))
.or_else(|| table.and_then(|t| constraint_column(constraint, t, &code)))
.unwrap_or_default();
Some(match kind {
ErrorKind::ForeignKeyViolation if detail.contains("still referenced") => Violation::new(
"",
"still_referenced",
"This record is still referenced by another record, so it cannot be removed",
),
ErrorKind::ForeignKeyViolation => {
let message = match path.as_str() {
"" => "The record refers to a record that does not exist".to_string(),
p => format!("{p} refers to a record that does not exist"),
};
Violation::new(path, "reference_missing", message)
}
ErrorKind::CheckViolation => {
let message = check_sentence(&path, &code);
Violation::new(path, code, message)
}
_ if sqlstate == Some("23P01") => Violation::new(
"",
code,
"The record overlaps another one it may not overlap",
),
_ => return None,
})
}
#[cfg(feature = "postgres")]
fn check_sentence(path: &str, code: &str) -> String {
let field = if path.is_empty() {
"A value".to_string()
} else {
path.to_string()
};
let predicate = match code {
"required" => "is required",
"max_length" => "is too long",
"min_length" => "is too short",
"length" => "does not have the required length",
"max" => "is too large",
"min" => "is too small",
"range" => "is out of the allowed range",
"positive" => "must be positive",
"non_negative" => "must not be negative",
"pattern" => "does not have the expected form",
"alpha_dash" => "may hold only letters, digits, '_' and '-'",
"alpha_num" => "may hold only letters and digits",
"values" => "is not one of the allowed values",
"email" => "must be an email address",
"url" => "must be an absolute URL",
"state_transition_not_declared" => "cannot move to that state from its current one",
_ if path.is_empty() => return "The record does not satisfy one of its rules".to_string(),
_ => "does not satisfy one of its rules",
};
format!("{field} {predicate}")
}
#[cfg(feature = "postgres")]
fn generated_code(constraint: &str) -> Option<String> {
let code = crate::violation::code_from_constraint(constraint);
(code != constraint).then_some(code)
}
#[cfg(feature = "postgres")]
const TENANT_KEYS: &[&str] = &["company_id", "tenant_id", "org_unit_id", "organization_id"];
#[cfg(feature = "postgres")]
fn constraint_column(constraint: &str, table: &str, code: &str) -> Option<String> {
let ident = |s: &str| {
(!s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_'))
.then(|| s.to_string())
};
if let Some(rest) = constraint.strip_prefix("fk_") {
let column = rest.strip_prefix(table)?.strip_prefix('_')?;
return ident(column.strip_suffix("_fk").unwrap_or(column));
}
let head = ["ck_", "uq_", "ex_", "tg_"]
.iter()
.find_map(|p| constraint.strip_prefix(p))?
.split_once("__")?
.0;
if ends_in_digest(head) {
return None;
}
let mut rest = head.strip_prefix(table)?.strip_prefix('_')?;
if !code.is_empty() {
if let Some(field) = rest.strip_suffix(code).and_then(|f| f.strip_suffix('_')) {
if !field.is_empty() {
rest = field;
}
}
}
while let Some(next) = TENANT_KEYS.iter().find_map(|key| {
rest.strip_prefix(key)?.strip_prefix('_').filter(|r| !r.is_empty())
}) {
rest = next;
}
ident(rest)
}
#[cfg(feature = "postgres")]
fn ends_in_digest(head: &str) -> bool {
let bytes = head.as_bytes();
bytes.len() > 9
&& bytes[bytes.len() - 9] == b'_'
&& bytes[bytes.len() - 8..].iter().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(b))
}
#[cfg(feature = "postgres")]
const SCOPE_COLUMNS: &[&str] = &[
"company_id",
"tenant_id",
"org_unit_id",
"organization_id",
"deleted_at",
];
#[cfg(feature = "postgres")]
fn key_column(detail: &str) -> Option<String> {
let rest = detail.strip_prefix("Key (")?;
let cols = &rest[..rest.find(")=")?];
let names: Vec<String> = cols
.split(',')
.filter_map(|part| {
let inner = part.rsplit('(').next().unwrap_or(part);
let ident: String = inner
.trim()
.trim_matches('"')
.chars()
.take_while(|c| c.is_ascii_alphanumeric() || *c == '_')
.collect();
(!ident.is_empty()).then_some(ident)
})
.collect();
names
.iter()
.rev()
.find(|n| !SCOPE_COLUMNS.contains(&n.as_str()))
.or_else(|| (names.len() == 1).then(|| &names[0]))
.cloned()
}
#[cfg(all(test, feature = "postgres"))]
mod constraint_violation_tests {
use super::*;
use sqlx::error::ErrorKind;
#[test]
fn a_missing_reference_names_the_column_and_never_the_table() {
let v = constraint_violation(
ErrorKind::ForeignKeyViolation,
Some("23503"),
Some("fk_subdistricts_district_id"),
"Key (district_id)=(1484f475-14be-4abc-985b-7291543d32bc) is not present in table \"districts\".",
None,
None,
)
.unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("district_id", "reference_missing"));
assert!(!v.message.contains("districts\"") && !v.message.contains("fk_"), "{}", v.message);
}
#[test]
fn removing_a_referenced_record_is_still_referenced() {
let v = constraint_violation(
ErrorKind::ForeignKeyViolation,
Some("23503"),
Some("fk_provinces_country_id"),
"Key (id)=(866f1b4e-a9f3-4626-95d1-13c183916b4b) is still referenced from table \"provinces\".",
None,
None,
)
.unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("", "still_referenced"));
}
#[test]
fn a_rule_constraint_answers_its_rules_code() {
let check = constraint_violation(ErrorKind::CheckViolation, Some("23514"), Some("ck_countries_name_max_length__max_length"), "", None, Some("countries")).unwrap();
assert_eq!((check.path.as_str(), check.code.as_str()), ("name", "max_length"));
let unique = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_accounts__account_code_taken"), "Key (company_id, code)=(a, b) already exists.", None, Some("accounts")).unwrap();
assert_eq!((unique.path.as_str(), unique.code.as_str()), ("code", "account_code_taken"));
let overlap = constraint_violation(ErrorKind::Other, Some("23P01"), Some("ex_tax_rates__overlapping_validity"), "", None, Some("tax_rates")).unwrap();
assert_eq!(overlap.code, "overlapping_validity");
}
#[test]
fn a_missing_required_value_is_a_required_violation_on_its_column() {
let v = constraint_violation(
ErrorKind::NotNullViolation,
Some("23502"),
None,
"",
Some("employee_id"),
Some("payslips"),
)
.unwrap();
assert_eq!(
(v.path.as_str(), v.code.as_str()),
("employee_id", "required")
);
assert_eq!(v.message, "employee_id is required");
}
#[test]
fn a_composite_unique_key_names_its_own_column_not_the_tenant() {
let v = constraint_violation(
ErrorKind::UniqueViolation,
Some("23505"),
Some("uq_accounts__code_taken"),
"Key (company_id, code)=(7c0a…, 1100) already exists.",
None,
Some("accounts"),
)
.unwrap();
assert_eq!(v.path, "code");
assert!(
v.message.starts_with("code "),
"a sentence with a subject: {}",
v.message
);
let scoped = constraint_violation(
ErrorKind::UniqueViolation,
Some("23505"),
Some("uq_items__sku_taken"),
"Key (org_unit_id, lower(sku::text))=(a, b) already exists.",
None,
Some("items"),
)
.unwrap();
assert_eq!(scoped.path, "sku");
let bare = constraint_violation(
ErrorKind::UniqueViolation,
Some("23505"),
Some("uq_x__taken"),
"",
None,
None,
)
.unwrap();
assert_eq!(bare.path, "");
assert!(
bare.message.starts_with(char::is_uppercase),
"{}",
bare.message
);
}
#[test]
fn a_check_reads_as_a_sentence_never_as_the_rule_identifier() {
let v = constraint_violation(
ErrorKind::CheckViolation,
Some("23514"),
Some("ck_tickets_title_max_length__max_length"),
"",
None,
Some("tickets"),
)
.unwrap();
assert_eq!(v.code, "max_length", "the code stays machine-readable");
assert!(
!v.message.contains('`') && !v.message.contains("max_length"),
"{}",
v.message
);
assert!(v.message.starts_with("title "), "{}", v.message);
let row = constraint_violation(
ErrorKind::CheckViolation,
Some("23514"),
Some("ck_journals_lines_must_balance_1f2e3d4c__unbalanced"),
"",
None,
Some("journals"),
)
.unwrap();
assert_eq!((row.path.as_str(), row.code.as_str()), ("", "unbalanced"));
assert!(
!row.message.contains("unbalanced") && !row.message.contains('`'),
"{}",
row.message
);
let tg = constraint_violation(
ErrorKind::CheckViolation,
Some("23514"),
Some("tg_journals__state_transition_not_declared"),
"",
Some("status"),
Some("journals"),
)
.unwrap();
assert_eq!(
(tg.path.as_str(), tg.code.as_str()),
("status", "state_transition_not_declared")
);
assert!(
!tg.message.contains("state_transition_not_declared"),
"{}",
tg.message
);
}
#[test]
fn a_database_error_inside_any_error_converts_as_the_database_error_does() {
let wrapped = anyhow::Error::new(sqlx::Error::RowNotFound).context("loading the row");
assert!(matches!(RepositoryError::from_source(wrapped), RepositoryError::NotFound));
assert!(matches!(RepositoryError::from_source(sqlx::Error::RowNotFound), RepositoryError::NotFound));
let other = RepositoryError::from_source(anyhow::anyhow!("pool closed"));
assert!(matches!(other, RepositoryError::DatabaseError(ref m) if m == "pool closed"), "{other:?}");
}
#[test]
fn every_unique_key_answers_taken_and_never_names_its_index() {
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("users_email_key"), "Key (email)=(a@b.c) already exists.", None, Some("users")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("email", "taken"));
assert!(!v.message.contains("users_email_key"), "{}", v.message);
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("idx_projects_contact_email"), "", None, Some("projects")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("", "taken"));
assert!(!v.message.contains("idx_"), "{}", v.message);
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_projects_contact_email__unique"), "", None, Some("projects")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("contact_email", "unique"));
assert!(!v.message.contains("uq_"), "{}", v.message);
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_tickets_title__unique"), "", None, Some("tickets")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("title", "unique"));
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_projects_company_id_code__code_taken"), "", None, Some("projects")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("code", "code_taken"));
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_items_company_id_org_unit_id_sku__unique"), "", None, Some("items")).unwrap();
assert_eq!(v.path, "sku");
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_members_org_unit_id__unique"), "", None, Some("members")).unwrap();
assert_eq!(v.path, "org_unit_id");
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_projects_company_id_some_long_fiel_0a1b2c3d__code_taken"), "", None, Some("projects")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("", "code_taken"));
let v = constraint_violation(ErrorKind::UniqueViolation, Some("23505"), Some("uq_tickets_org_unit_id_reference"), "", None, Some("tickets")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("", "taken"));
assert!(!v.message.contains("uq_"), "{}", v.message);
}
#[test]
fn a_foreign_key_on_a_table_with_rls_names_its_column_from_the_constraint() {
let v = constraint_violation(ErrorKind::ForeignKeyViolation, Some("23503"), Some("fk_tickets_project_id"), "", None, Some("tickets")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("project_id", "reference_missing"));
assert!(!v.message.contains("fk_"), "{}", v.message);
let link = constraint_violation(ErrorKind::ForeignKeyViolation, Some("23503"), Some("fk_user_roles_role_id_fk"), "", None, Some("user_roles")).unwrap();
assert_eq!(link.path, "role_id");
let other = constraint_violation(ErrorKind::ForeignKeyViolation, Some("23503"), Some("fk_other_project_id"), "", None, Some("tickets")).unwrap();
assert_eq!(other.path, "");
}
#[test]
fn a_rule_check_about_a_field_answers_on_that_field() {
let v = constraint_violation(ErrorKind::CheckViolation, Some("23514"), Some("ck_projects_kind__website_required_for_client"), "", None, Some("projects")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("kind", "website_required_for_client"));
assert!(!v.message.contains("ck_") && !v.message.contains("website_required"), "{}", v.message);
let v = constraint_violation(ErrorKind::CheckViolation, Some("23514"), Some("ck_shifts_call_note_max_length__max_length"), "", None, Some("shifts")).unwrap();
assert_eq!(v.path, "call_note");
let v = constraint_violation(ErrorKind::CheckViolation, Some("23514"), Some("ck_projects_a_very_long_field_name_dead00ff__website_required_for_client"), "", None, Some("projects")).unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("", "website_required_for_client"));
}
#[test]
fn a_row_security_refusal_is_scope_refused_and_a_missing_grant_is_not() {
let v = scope_refusal(Some("42501"), "new row violates row-level security policy for table \"projects\"").unwrap();
assert_eq!((v.path.as_str(), v.code.as_str()), ("", "scope_refused"));
assert!(!v.message.contains("projects") && !v.message.contains("policy"), "{}", v.message);
assert!(scope_refusal(Some("42501"), "permission denied for table projects").is_none());
assert!(scope_refusal(Some("23505"), "row-level security policy").is_none());
}
#[test]
fn an_error_that_names_no_rule_keeps_its_old_answer() {
assert!(
constraint_violation(ErrorKind::Other, Some("42P01"), None, "", None, None).is_none()
);
assert!(constraint_violation(
ErrorKind::NotNullViolation,
Some("23502"),
None,
"",
None,
None
)
.is_none());
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Reference {
pub field: &'static str,
pub table: &'static str,
pub column: &'static str,
}
impl Reference {
pub const fn new(field: &'static str, table: &'static str, column: &'static str) -> Self {
Self { field, table, column }
}
}
pub trait PersistentEntity: Clone + Send + Sync + Debug + Serialize + DeserializeOwned + 'static {
fn entity_id(&self) -> String;
fn set_entity_id(&mut self, id: String);
fn created_at(&self) -> Option<DateTime<Utc>>;
fn set_created_at(&mut self, ts: DateTime<Utc>);
fn updated_at(&self) -> Option<DateTime<Utc>>;
fn set_updated_at(&mut self, ts: DateTime<Utc>);
fn deleted_at(&self) -> Option<DateTime<Utc>>;
fn set_deleted_at(&mut self, ts: Option<DateTime<Utc>>);
fn is_deleted(&self) -> bool {
self.deleted_at().is_some()
}
fn mark_deleted(&mut self) {
self.set_deleted_at(Some(Utc::now()));
self.set_updated_at(Utc::now());
}
fn restore(&mut self) {
self.set_deleted_at(None);
self.set_updated_at(Utc::now());
}
fn touch(&mut self) {
self.set_updated_at(Utc::now());
}
fn generate_id() -> String {
uuid::Uuid::new_v4().to_string()
}
fn write_protected_fields() -> &'static [&'static str] {
&[]
}
fn state_fields() -> &'static [&'static str] {
&[]
}
fn references() -> &'static [Reference] {
&[]
}
}
pub trait PartialUpdatable: PersistentEntity {
fn apply_partial_update(&mut self, fields: &HashMap<String, serde_json::Value>) -> Result<(), RepositoryError>;
}
pub trait Versioned {
fn version(&self) -> u64;
fn set_version(&mut self, version: u64);
fn increment_version(&mut self) {
self.set_version(self.version() + 1);
}
}
#[async_trait]
pub trait CrudRepository<E>: Send + Sync
where
E: PersistentEntity,
{
async fn create(&self, entity: E) -> Result<E, RepositoryError>;
async fn find_by_id(&self, id: &str) -> Result<Option<E>, RepositoryError>;
async fn find_by_id_including_deleted(&self, id: &str) -> Result<Option<E>, RepositoryError>;
async fn update(&self, entity: E) -> Result<E, RepositoryError>;
async fn update_changed(&self, before: &E, after: E) -> Result<Option<E>, RepositoryError> {
let _ = before;
match self.update(after).await {
Ok(saved) => Ok(Some(saved)),
Err(RepositoryError::NotFound) => Ok(None),
Err(e) => Err(e),
}
}
async fn soft_delete(&self, id: &str) -> Result<bool, RepositoryError>;
async fn restore(&self, id: &str) -> Result<Option<E>, RepositoryError>;
async fn hard_delete(&self, id: &str) -> Result<bool, RepositoryError>;
async fn list(&self, page: u32, limit: u32) -> Result<(Vec<E>, u64), RepositoryError>;
async fn list_deleted(&self, page: u32, limit: u32) -> Result<(Vec<E>, u64), RepositoryError>;
async fn list_deleted_filtered(
&self,
page: u32,
limit: u32,
filters: HashMap<String, String>,
) -> Result<(Vec<E>, u64), RepositoryError> {
let _ = filters;
self.list_deleted(page, limit).await
}
async fn count(&self) -> Result<u64, RepositoryError>;
async fn count_deleted(&self) -> Result<u64, RepositoryError>;
async fn bulk_create(&self, entities: Vec<E>) -> Result<Vec<E>, RepositoryError>;
async fn empty_trash(&self) -> Result<u64, RepositoryError>;
async fn list_filtered(
&self,
page: u32,
limit: u32,
filters: HashMap<String, String>,
) -> Result<(Vec<E>, u64), RepositoryError> {
let _ = filters; self.list(page, limit).await
}
async fn list_filtered_with_info(
&self,
page: u32,
limit: u32,
filters: HashMap<String, String>,
) -> Result<(Vec<E>, backbone_orm::repository::PaginationInfo), RepositoryError> {
let (rows, total) = self.list_filtered(page, limit, filters).await?;
Ok((
rows,
backbone_orm::repository::PaginationInfo::new(page, limit, total),
))
}
fn table_name(&self) -> Option<&str> {
None
}
fn write_pool(&self) -> Option<&backbone_orm::PgPool> {
None
}
async fn aggregate_filtered(
&self,
spec: &backbone_orm::repository::AggregateSpec,
filters: HashMap<String, String>,
) -> Result<backbone_orm::repository::AggregateResult, RepositoryError> {
let _ = (spec, filters);
Err(RepositoryError::DatabaseError(
"aggregate is not supported by this repository".to_string(),
))
}
async fn fetch_related_json(
&self,
_table: &str,
_ids: &[String],
) -> Vec<serde_json::Value> {
Vec::new()
}
async fn reference_visible(
&self,
table: &str,
column: &str,
value: &serde_json::Value,
) -> Result<Option<bool>, RepositoryError> {
let _ = (table, column, value);
Ok(None)
}
async fn exists(&self, id: &str) -> Result<bool, RepositoryError> {
Ok(self.find_by_id(id).await?.is_some())
}
async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64, RepositoryError> {
let mut n = 0;
for id in ids {
if self.soft_delete(id).await? {
n += 1;
}
}
Ok(n)
}
async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<E>, RepositoryError> {
let mut out = Vec::with_capacity(ids.len());
for id in ids {
if let Some(e) = self.restore(id).await? {
out.push(e);
}
}
Ok(out)
}
async fn bulk_hard_delete(&self, ids: &[String]) -> Result<u64, RepositoryError> {
let mut n = 0;
for id in ids {
if self.hard_delete(id).await? {
n += 1;
}
}
Ok(n)
}
async fn restore_all(&self) -> Result<Vec<E>, RepositoryError> {
let mut restored = Vec::new();
loop {
let (batch, _) = self.list_deleted(1, 500).await?;
if batch.is_empty() {
break;
}
for entity in &batch {
if let Some(e) = self.restore(&entity.entity_id()).await? {
restored.push(e);
}
}
}
Ok(restored)
}
async fn bulk_update(&self, entities: Vec<E>) -> Result<Vec<E>, RepositoryError> {
let mut out = Vec::with_capacity(entities.len());
for entity in entities {
out.push(self.update(entity).await?);
}
Ok(out)
}
async fn bulk_update_changed(&self, pairs: Vec<(E, E)>) -> Result<Vec<E>, RepositoryError> {
let mut out = Vec::with_capacity(pairs.len());
for (index, (before, after)) in pairs.into_iter().enumerate() {
match self.update_changed(&before, after).await {
Ok(Some(saved)) => out.push(saved),
Ok(None) => return Err(RepositoryError::NotFound.at_batch_index(index)),
Err(e) => return Err(e.at_batch_index(index)),
}
}
Ok(out)
}
}
#[async_trait]
pub trait SearchableRepository<E>: CrudRepository<E>
where
E: PersistentEntity,
{
async fn search(
&self,
filters: HashMap<String, String>,
page: u32,
limit: u32,
) -> Result<(Vec<E>, u64), RepositoryError>;
async fn find_by_field(&self, field: &str, value: &str) -> Result<Option<E>, RepositoryError>;
async fn find_all_by_field(
&self,
field: &str,
value: &str,
page: u32,
limit: u32,
) -> Result<(Vec<E>, u64), RepositoryError>;
}
#[cfg(feature = "postgres")]
pub use postgres_traits::*;
#[cfg(feature = "postgres")]
mod postgres_traits {
use super::*;
use sqlx::postgres::PgRow;
use sqlx::FromRow;
pub trait PostgresEntity: PersistentEntity + for<'r> FromRow<'r, PgRow> + Unpin {
fn table_name() -> &'static str;
fn id_column() -> &'static str {
"id"
}
fn select_columns() -> &'static [&'static str];
fn insert_columns() -> &'static [&'static str];
fn update_columns() -> &'static [&'static str] {
Self::insert_columns()
}
fn bind_for_insert<'q>(
entity: &'q Self,
query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>;
fn bind_for_update<'q>(
entity: &'q Self,
query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> {
Self::bind_for_insert(entity, query)
}
fn select_by_id_query() -> String {
format!(
"SELECT {} FROM {} WHERE {} = $1 AND deleted_at IS NULL",
Self::select_columns().join(", "),
Self::table_name(),
Self::id_column()
)
}
fn select_by_id_including_deleted_query() -> String {
format!(
"SELECT {} FROM {} WHERE {} = $1",
Self::select_columns().join(", "),
Self::table_name(),
Self::id_column()
)
}
fn list_query() -> String {
format!(
"SELECT {} FROM {} WHERE deleted_at IS NULL ORDER BY created_at DESC, {} LIMIT $1 OFFSET $2",
Self::select_columns().join(", "),
Self::table_name(),
Self::id_column()
)
}
fn count_query() -> String {
format!(
"SELECT COUNT(*) FROM {} WHERE deleted_at IS NULL",
Self::table_name()
)
}
fn list_deleted_query() -> String {
format!(
"SELECT {} FROM {} WHERE deleted_at IS NOT NULL ORDER BY deleted_at DESC, {} LIMIT $1 OFFSET $2",
Self::select_columns().join(", "),
Self::table_name(),
Self::id_column()
)
}
fn count_deleted_query() -> String {
format!(
"SELECT COUNT(*) FROM {} WHERE deleted_at IS NOT NULL",
Self::table_name()
)
}
fn soft_delete_query() -> String {
format!(
"UPDATE {} SET deleted_at = NOW(), updated_at = NOW() WHERE {} = $1 AND deleted_at IS NULL",
Self::table_name(),
Self::id_column()
)
}
fn restore_query() -> String {
format!(
"UPDATE {} SET deleted_at = NULL, updated_at = NOW() WHERE {} = $1 AND deleted_at IS NOT NULL RETURNING {}",
Self::table_name(),
Self::id_column(),
Self::select_columns().join(", ")
)
}
fn hard_delete_query() -> String {
format!(
"DELETE FROM {} WHERE {} = $1",
Self::table_name(),
Self::id_column()
)
}
fn empty_trash_query() -> String {
format!(
"DELETE FROM {} WHERE deleted_at IS NOT NULL",
Self::table_name()
)
}
}
}