use async_trait::async_trait;
use sqlx::PgPool;
{% if has_map %}use sqlx::types::Json;
use std::collections::HashMap;
{% endif %}
use business::domain::shared::pagination::{Page, PageRequest};
use business::domain::shared::repository_error::RepositoryError;
use business::domain::{{ snake }}::{
model::{{ "{" }}{{ pascal }}, {{ pascal }}Id},
repository::{{ pascal }}RepositoryTrait,
};
use super::entity::{{ pascal }}Db;
use crate::db::map_sqlx_error;
pub struct Pg{{ pascal }}Repository {
pub pool: PgPool,
}
impl Pg{{ pascal }}Repository {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
/// A row as the domain's entity. A row the domain rejects is stored data gone bad, not the
/// caller's fault: it is logged with the domain's reason and reported as `InvalidData`.
fn to_domain(row: {{ pascal }}Db) -> Result<{{ pascal }}, RepositoryError> {
{{ pascal }}::try_from(row).map_err(|e| {
tracing::error!(error = %e, "stored row is not a valid {{ snake }}");
RepositoryError::InvalidData
})
}
#[async_trait]
impl {{ pascal }}RepositoryTrait for Pg{{ pascal }}Repository {
async fn find_all(&self, page: PageRequest) -> Result<Page<{{ pascal }}>, RepositoryError> {
tracing::debug!(limit = page.limit(), offset = page.offset(), "find_all");
let rows = sqlx::query_as!(
{{ pascal }}Db,
"SELECT {{ select_cols }}
FROM {{ plural }} WHERE deleted = false
ORDER BY created_at DESC, id LIMIT $1 OFFSET $2",
i64::from(page.limit()),
i64::from(page.offset())
)
.fetch_all(&self.pool)
.await
.map_err(|e| {
tracing::error!(error = %e, "find_all failed");
map_sqlx_error(&e)
})?;
let total = sqlx::query_scalar!(
r#"SELECT COUNT(*) AS "total!" FROM {{ plural }} WHERE deleted = false"#
)
.fetch_one(&self.pool)
.await
.map_err(|e| {
tracing::error!(error = %e, "find_all count failed");
map_sqlx_error(&e)
})?;
let items = rows.into_iter().map(to_domain).collect::<Result<Vec<_>, _>>()?;
// COUNT(*) is never negative.
let total = u64::try_from(total).unwrap_or_default();
Ok(Page::new(items, total, page))
}
async fn find_by_id(&self, id: {{ pascal }}Id) -> Result<Option<{{ pascal }}>, RepositoryError> {
tracing::debug!(%id, "find_by_id");
let row = sqlx::query_as!(
{{ pascal }}Db,
"SELECT {{ select_cols }}
FROM {{ plural }} WHERE id = $1 AND deleted = false",
id.value()
)
.fetch_optional(&self.pool)
.await
.map_err(|e| {
tracing::error!(error = %e, "find_by_id failed");
map_sqlx_error(&e)
})?;
row.map(to_domain).transpose()
}
async fn save(&self, entity: &{{ pascal }}) -> Result<(), RepositoryError> {
tracing::debug!(id = %entity.id, "save");
let db = {{ pascal }}Db::from(entity);
sqlx::query!(
"INSERT INTO {{ plural }} ({{ all_cols }})
VALUES ({{ all_params }})
ON CONFLICT (id) DO UPDATE
SET {{ all_updates }}",
db.id,
db.created_at,
db.updated_at,
db.deleted,
db.deleted_at,
{% for f in fields %} {% if f.is_array %}&db.{{ f.name }}{% elif f.is_map %}db.{{ f.name }} as _{% else %}db.{{ f.name }}{% endif %},
{% endfor %} )
.execute(&self.pool)
.await
.map_err(|e| {
tracing::error!(error = %e, "save failed");
map_sqlx_error(&e)
})?;
Ok(())
}
}
#[cfg(test)]
mod integration_tests {
use super::*;
use business::domain::{{ snake }}::model::{{ pascal }}Props;
{% for import in test_imports %} use {{ import }};
{% endfor %}
fn test_repo(pool: PgPool) -> Pg{{ pascal }}Repository {
Pg{{ pascal }}Repository::new(pool)
}
async fn seed(pool: &PgPool) -> {{ pascal }} {
let entity = {{ pascal }}::new({{ pascal }}Props {
{% for f in fields %} {{ f.name }}: {{ f.seed_literal }},
{% endfor %} }).unwrap();
test_repo(pool.clone()).save(&entity).await.unwrap();
entity
}
#[sqlx::test(migrations = "./migrations")]
async fn should_persist_and_retrieve_by_id(pool: PgPool) {
// Arrange
let entity = seed(&pool).await;
// Act
let found = test_repo(pool).find_by_id(entity.id).await.unwrap().unwrap();
// Assert
assert_eq!(found.id, entity.id);
{% for f in fields %} assert_eq!(found.{{ f.name }}, entity.{{ f.name }});
{% endfor %} assert!(!found.deleted);
}
#[sqlx::test(migrations = "./migrations")]
async fn should_return_none_for_nonexistent_id(pool: PgPool) {
// Act
let result = test_repo(pool).find_by_id({{ pascal }}Id::generate()).await.unwrap();
// Assert
assert!(result.is_none());
}
#[sqlx::test(migrations = "./migrations")]
async fn should_update_entity_on_save_conflict(pool: PgPool) {
// Arrange
{% set required = fields | filter(attribute="required_string", value=true) %}{% set string_vos = fields | filter(attribute="string_vo", value=true) %}{% if required or string_vos %}{% if required %}{% set f = required.0 %}{% else %}{% set f = string_vos.0 %}{% endif %} let mut entity = seed(&pool).await;
entity.{{ f.name }} = {% if f.is_vo %}{{ f.vo }}::new("updated".to_string()).unwrap(){% else %}"updated".to_string(){% endif %};
entity.updated_at = chrono::Utc::now();
// Act
test_repo(pool.clone()).save(&entity).await.unwrap();
// Assert
let found = test_repo(pool).find_by_id(entity.id).await.unwrap().unwrap();
assert_eq!(found.{{ f.name }}{% if f.is_vo %}.value(){% endif %}, "updated");
{% else %} let entity = seed(&pool).await;
// Act
test_repo(pool.clone()).save(&entity).await.unwrap();
// Assert
let found = test_repo(pool).find_by_id(entity.id).await.unwrap();
assert!(found.is_some());
{% endif %} }
#[sqlx::test(migrations = "./migrations")]
async fn should_list_all_{{ plural }}_excluding_deleted(pool: PgPool) {
// Arrange
seed(&pool).await;
let mut deleted = seed(&pool).await;
deleted.deleted = true;
deleted.deleted_at = Some(deleted.updated_at);
test_repo(pool.clone()).save(&deleted).await.unwrap();
// Act
let page = test_repo(pool).find_all(PageRequest::default()).await.unwrap();
// Assert
assert_eq!(page.items.len(), 1);
assert_eq!(page.total, 1);
}
#[sqlx::test(migrations = "./migrations")]
async fn should_return_the_requested_page_of_a_larger_list(pool: PgPool) {
// Arrange
for _ in 0..3 {
seed(&pool).await;
}
// Act
let page = test_repo(pool)
.find_all(PageRequest::new(Some(2), Some(2)))
.await
.unwrap();
// Assert
assert_eq!(page.items.len(), 1);
assert_eq!(page.total, 3);
assert_eq!((page.limit, page.offset), (2, 2));
}
{% set uniques = fields | filter(attribute="unique", value=true) %}{% if uniques %}{% set u = uniques.0 %}
#[sqlx::test(migrations = "./migrations")]
async fn should_reject_a_duplicate_{{ u.name }}(pool: PgPool) {
// Arrange
let first = seed(&pool).await;
let mut second = {{ pascal }}::new({{ pascal }}Props {
{% for f in fields %} {{ f.name }}: {{ f.seed_literal }},
{% endfor %} }).unwrap();
second.{{ u.name }} = first.{{ u.name }}{% if u.needs_clone %}.clone(){% endif %};
// Act
let result = test_repo(pool).save(&second).await;
// Assert
assert_eq!(result.unwrap_err(), RepositoryError::Duplicate);
}
{% endif %}}