feat: import store aggregate from inventory into billing domain
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
ci/woodpecker/pr/woodpecker Pipeline failed

This commit is contained in:
Aravinth Manivannan 2024-09-16 17:58:28 +05:30
parent 6b6084a0e9
commit 5e48eed486
Signed by: realaravinth
GPG key ID: F8F50389936984FF
23 changed files with 1574 additions and 0 deletions

View file

@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS (\n SELECT 1\n FROM cqrs_billing_store_query\n WHERE\n name = $1\n AND\n deleted = false\n );",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "03b31d653e96f3b28ff15c86ff5801fe319ba86a15c09aacc4990a7a58849081"
}

View file

@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n store_id, version\n FROM\n cqrs_billing_store_query\n WHERE\n store_id = $1;",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "store_id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "version",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false
]
},
"hash": "0bf69e4af7657572bf84633ecf8e2649a7b5baa3fc3abbb6ee7f522fedf3062e"
}

View file

@ -0,0 +1,19 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE\n cqrs_billing_store_query\n SET\n version = $1,\n name = $2,\n address = $3,\n store_id = $4,\n owner = $5,\n deleted = $6;",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Text",
"Text",
"Uuid",
"Uuid",
"Bool"
]
},
"nullable": []
},
"hash": "3811531518316435c32223582f9de6bdfefc19839129577407a1e4071f5c49f6"
}

View file

@ -0,0 +1,19 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO cqrs_billing_store_query (\n version, name, address, store_id, owner, deleted\n ) VALUES (\n $1, $2, $3, $4, $5, $6\n );",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Text",
"Text",
"Uuid",
"Uuid",
"Bool"
]
},
"nullable": []
},
"hash": "78008f1b0d3f366937c72a181b19217cd9f28f1ebe73ddc081ec2f7bec5dc1b3"
}

View file

@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS (\n SELECT 1\n FROM cqrs_billing_store_query\n WHERE\n store_id = $1\n );",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "ba895b4310f182a73d88df08255dc89374f95ea0e3967124cb1e414c52d0428a"
}

View file

@ -0,0 +1,46 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n name, address, store_id, owner, deleted\n FROM\n cqrs_billing_store_query\n WHERE\n store_id = $1;",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "name",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "address",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "store_id",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "owner",
"type_info": "Uuid"
},
{
"ordinal": 4,
"name": "deleted",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
true,
false,
false,
false
]
},
"hash": "c3f26f3816104a510a4630d5d3d27178f9bcf4055cbf176b5e8fb2376a5972ba"
}

View file

@ -0,0 +1,16 @@
--- SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
--
-- SPDX-License-Identifier: AGPL-3.0-or-later
CREATE TABLE IF NOT EXISTS cqrs_billing_store_query
(
version bigint CHECK (version >= 0) NOT NULL,
name TEXT NOT NULL,
address TEXT,
owner UUID NOT NULL,
store_id UUID NOT NULL UNIQUE,
deleted BOOLEAN NOT NULL DEFAULT FALSE,
PRIMARY KEY (store_id)
);

View file

@ -0,0 +1,4 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
mod postgres;

View file

@ -0,0 +1,79 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use std::borrow::Cow;
use cqrs_es::persist::PersistenceError;
use sqlx::Error as SqlxError;
use crate::billing::application::port::output::db::errors::BillingDBError;
impl From<SqlxError> for BillingDBError {
fn from(e: SqlxError) -> Self {
log::error!("[postgres] err: {}", e);
if let SqlxError::Database(err) = e {
if err.code() == Some(Cow::from("23505")) {
let msg = err.message();
if msg.contains("cqrs_inventory_store_query_store_id_key") {
return Self::DuplicateStoreID;
} else {
println!("{msg}");
}
}
}
Self::InternalError
}
}
/// map custom row not found error to DB error
pub fn map_row_not_found_err(e: SqlxError, row_not_found: BillingDBError) -> BillingDBError {
if let SqlxError::RowNotFound = e {
row_not_found
} else {
e.into()
}
}
#[derive(Debug)]
pub enum PostgresAggregateError {
OptimisticLock,
ConnectionError(Box<dyn std::error::Error + Send + Sync + 'static>),
DeserializationError(Box<dyn std::error::Error + Send + Sync + 'static>),
UnknownError(Box<dyn std::error::Error + Send + Sync + 'static>),
}
impl From<SqlxError> for PostgresAggregateError {
fn from(err: SqlxError) -> Self {
// TODO: improve error handling
match &err {
SqlxError::Database(database_error) => {
if let Some(code) = database_error.code() {
if code.as_ref() == "23505" {
return PostgresAggregateError::OptimisticLock;
}
}
PostgresAggregateError::UnknownError(Box::new(err))
}
SqlxError::Io(_) | SqlxError::Tls(_) => {
PostgresAggregateError::ConnectionError(Box::new(err))
}
_ => PostgresAggregateError::UnknownError(Box::new(err)),
}
}
}
impl From<PostgresAggregateError> for PersistenceError {
fn from(err: PostgresAggregateError) -> Self {
match err {
PostgresAggregateError::OptimisticLock => PersistenceError::OptimisticLockError,
PostgresAggregateError::ConnectionError(error) => {
PersistenceError::ConnectionError(error)
}
PostgresAggregateError::DeserializationError(error) => {
PersistenceError::UnknownError(error)
}
PostgresAggregateError::UnknownError(error) => PersistenceError::UnknownError(error),
}
}
}

View file

@ -0,0 +1,28 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use std::sync::Arc;
use sqlx::postgres::PgPool;
use crate::db::{migrate::RunMigrations, sqlx_postgres::Postgres};
mod errors;
mod store_id_exists;
mod store_name_exists;
mod store_view;
#[derive(Clone)]
pub struct BillingDBPostgresAdapter {
pool: PgPool,
}
impl BillingDBPostgresAdapter {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
pub fn migratable(&self) -> Arc<dyn RunMigrations> {
Arc::new(Postgres::new(self.pool.clone()))
}
}

View file

@ -0,0 +1,87 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use uuid::Uuid;
use super::BillingDBPostgresAdapter;
use crate::billing::application::port::output::db::{errors::*, store_id_exists::*};
use crate::billing::domain::store_aggregate::*;
#[async_trait::async_trait]
impl StoreIDExistsDBPort for BillingDBPostgresAdapter {
async fn store_id_exists(&self, store_id: &Uuid) -> BillingDBResult<bool> {
let res = sqlx::query!(
"SELECT EXISTS (
SELECT 1
FROM cqrs_billing_store_query
WHERE
store_id = $1
);",
store_id
)
.fetch_one(&self.pool)
.await?;
if let Some(x) = res.exists {
Ok(x)
} else {
Ok(false)
}
}
}
#[cfg(test)]
pub mod tests {
use uuid::Uuid;
use crate::utils::uuid::tests::UUID;
use super::*;
pub async fn create_dummy_store_record(s: &Store, db: &BillingDBPostgresAdapter) {
sqlx::query!(
"INSERT INTO cqrs_billing_store_query
(version, name, address, store_id, owner, deleted)
VALUES ($1, $2, $3, $4, $5 ,$6);",
1,
s.name(),
s.address().as_ref().unwrap(),
s.store_id(),
s.owner(),
false
)
.execute(&db.pool)
.await
.unwrap();
}
#[actix_rt::test]
async fn test_postgres_store_exists() {
let store_id = Uuid::new_v4();
let settings = crate::settings::tests::get_settings().await;
settings.create_db().await;
let db = super::BillingDBPostgresAdapter::new(
sqlx::postgres::PgPool::connect(&settings.database.url)
.await
.unwrap(),
);
let store = StoreBuilder::default()
.name("store_name".into())
.owner(UUID)
.address(Some("store_address".into()))
.store_id(store_id)
.build()
.unwrap();
// state doesn't exist
assert!(!db.store_id_exists(store.store_id()).await.unwrap());
create_dummy_store_record(&store, &db).await;
// state exists
assert!(db.store_id_exists(store.store_id()).await.unwrap());
settings.drop_db().await;
}
}

View file

@ -0,0 +1,81 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use super::BillingDBPostgresAdapter;
use crate::billing::application::port::output::db::{errors::*, store_name_exists::*};
use crate::billing::domain::store_aggregate::*;
#[async_trait::async_trait]
impl StoreNameExistsDBPort for BillingDBPostgresAdapter {
async fn store_name_exists(&self, s: &Store) -> BillingDBResult<bool> {
let res = sqlx::query!(
"SELECT EXISTS (
SELECT 1
FROM cqrs_billing_store_query
WHERE
name = $1
AND
deleted = false
);",
s.name(),
)
.fetch_one(&self.pool)
.await?;
if let Some(x) = res.exists {
Ok(x)
} else {
Ok(false)
}
}
}
#[cfg(test)]
mod tests {
use uuid::Uuid;
use crate::utils::uuid::tests::UUID;
use super::*;
use crate::billing::adapters::output::db::postgres::store_id_exists::tests::create_dummy_store_record;
#[actix_rt::test]
async fn test_postgres_store_exists() {
let store_id = Uuid::new_v4();
let settings = crate::settings::tests::get_settings().await;
settings.create_db().await;
let db = super::BillingDBPostgresAdapter::new(
sqlx::postgres::PgPool::connect(&settings.database.url)
.await
.unwrap(),
);
let store = StoreBuilder::default()
.name("store_name".into())
.owner(UUID)
.address(Some("store_address".into()))
.store_id(store_id)
.build()
.unwrap();
// state doesn't exist
assert!(!db.store_name_exists(&store).await.unwrap());
create_dummy_store_record(&store, &db).await;
// state exists
assert!(db.store_name_exists(&store).await.unwrap());
// Set store.deleted = true; now db.store_name_exists must return false
sqlx::query!(
"UPDATE cqrs_billing_store_query SET deleted = true WHERE store_id = $1;",
store.store_id()
)
.execute(&db.pool)
.await
.unwrap();
assert!(!db.store_name_exists(&store).await.unwrap());
settings.drop_db().await;
}
}

View file

@ -0,0 +1,302 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use async_trait::async_trait;
use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
use cqrs_es::{EventEnvelope, Query, View};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use super::errors::*;
use super::BillingDBPostgresAdapter;
use crate::billing::domain::events::BillingEvent;
use crate::billing::domain::store_aggregate::Store;
use crate::utils::parse_aggregate_id::parse_aggregate_id;
pub const NEW_STORE_NON_UUID: &str = "billing_new_store_non_uuid-asdfa";
// The view for a Store query, for a standard http application this should
// be designed to reflect the response dto that will be returned to a user.
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct StoreView {
name: String,
address: Option<String>,
store_id: Uuid,
owner: Uuid,
deleted: bool,
}
// This updates the view with events as they are committed.
// The logic should be minimal here, e.g., don't calculate the account balance,
// design the events to carry the balance information instead.
impl View<Store> for StoreView {
fn update(&mut self, event: &EventEnvelope<Store>) {
if let BillingEvent::StoreAdded(val) = &event.payload {
self.name = val.name().into();
self.address = val.address().clone();
self.store_id = *val.store_id();
self.owner = *val.owner();
self.deleted = false;
}
}
}
#[async_trait]
impl ViewRepository<StoreView, Store> for BillingDBPostgresAdapter {
async fn load(&self, store_id: &str) -> Result<Option<StoreView>, PersistenceError> {
let store_id = match parse_aggregate_id(store_id, NEW_STORE_NON_UUID)? {
Some((val, _)) => return Ok(Some(val)),
None => Uuid::parse_str(store_id).unwrap(),
};
let res = sqlx::query_as!(
StoreView,
"SELECT
name, address, store_id, owner, deleted
FROM
cqrs_billing_store_query
WHERE
store_id = $1;",
store_id
)
.fetch_one(&self.pool)
.await
.map_err(PostgresAggregateError::from)?;
Ok(Some(res))
}
async fn load_with_context(
&self,
store_id: &str,
) -> Result<Option<(StoreView, ViewContext)>, PersistenceError> {
let store_id = match parse_aggregate_id(store_id, NEW_STORE_NON_UUID)? {
Some(val) => return Ok(Some(val)),
None => Uuid::parse_str(store_id).unwrap(),
};
let res = sqlx::query_as!(
StoreView,
"SELECT
name, address, store_id, owner, deleted
FROM
cqrs_billing_store_query
WHERE
store_id = $1;",
&store_id,
)
.fetch_one(&self.pool)
.await
.map_err(PostgresAggregateError::from)?;
struct Context {
version: i64,
store_id: Uuid,
}
let ctx = sqlx::query_as!(
Context,
"SELECT
store_id, version
FROM
cqrs_billing_store_query
WHERE
store_id = $1;",
store_id
)
.fetch_one(&self.pool)
.await
.map_err(PostgresAggregateError::from)?;
let view_context = ViewContext::new(ctx.store_id.to_string(), ctx.version);
Ok(Some((res, view_context)))
}
async fn update_view(
&self,
view: StoreView,
context: ViewContext,
) -> Result<(), PersistenceError> {
match context.version {
0 => {
let version = context.version + 1;
sqlx::query!(
"INSERT INTO cqrs_billing_store_query (
version, name, address, store_id, owner, deleted
) VALUES (
$1, $2, $3, $4, $5, $6
);",
version,
view.name,
view.address,
view.store_id,
view.owner,
view.deleted,
)
.execute(&self.pool)
.await
.map_err(PostgresAggregateError::from)?;
}
_ => {
let version = context.version + 1;
sqlx::query!(
"UPDATE
cqrs_billing_store_query
SET
version = $1,
name = $2,
address = $3,
store_id = $4,
owner = $5,
deleted = $6;",
version,
view.name,
view.address,
view.store_id,
view.owner,
view.deleted,
)
.execute(&self.pool)
.await
.map_err(PostgresAggregateError::from)?;
}
}
Ok(())
}
}
pub struct SimpleLoggingQuery {}
// Our simplest query, this is great for debugging but absolutely useless in production.
// This query just pretty prints the events as they are processed.
#[async_trait]
impl Query<Store> for SimpleLoggingQuery {
async fn dispatch(&self, aggregate_id: &str, events: &[EventEnvelope<Store>]) {
for event in events {
let payload = serde_json::to_string_pretty(&event.payload).unwrap();
println!("{}-{}\n{}", aggregate_id, event.sequence, payload);
}
}
}
#[async_trait]
impl Query<Store> for BillingDBPostgresAdapter {
async fn dispatch(&self, store_id: &str, events: &[EventEnvelope<Store>]) {
let res = self
.load_with_context(store_id)
.await
.unwrap_or_else(|_| Some((StoreView::default(), ViewContext::new(store_id.into(), 0))));
let (mut view, view_context): (StoreView, ViewContext) = res.unwrap();
for event in events {
view.update(event);
}
self.update_view(view, view_context).await.unwrap();
}
}
// Our second query, this one will be handled with Postgres `GenericQuery`
// which will serialize and persist our view after it is updated. It also
// provides a `load` method to deserialize the view on request.
//pub type StoreQuery = GenericQuery<BillingDBPostgresAdapter, StoreView, Store>;
//pub type StoreQuery = Query<dyn BillingDBPostgresAdapter, StoreView, Store>;
//#[cfg(test)]
//mod tests {
// use super::*;
//
// use postgres_es::PostgresCqrs;
//
// use crate::{
// db::migrate::*,
// billing::{
// application::services::{
// add_category_service::tests::mock_add_category_service, add_customization_service::tests::mock_add_customization_service, add_line_item_service::tests::mock_add_line_item_service, add_product_service::tests::mock_add_product_service, add_store_service::AddStoreServiceBuilder, update_category_service::tests::mock_update_category_service, update_customization_service::tests::mock_update_customization_service, update_product_service::tests::mock_update_product_service, update_store_service::tests::mock_update_store_service, BillingServicesBuilder
// },
// domain::{
// add_category_command::AddCategoryCommand, add_customization_command,
// add_product_command::tests::get_command, add_store_command::AddStoreCommand,
// commands::BillingCommand,
// update_category_command::tests::get_update_category_command,
// update_customization_command::tests::get_update_customization_command,
// update_product_command, update_store_command::tests::get_update_store_cmd,
// },
// },
// tests::bdd::IS_NEVER_CALLED,
// utils::{random_string::GenerateRandomStringInterface, uuid::tests::UUID},
// };
// use std::sync::Arc;
//
// #[actix_rt::test]
// async fn pg_query() {
// let settings = crate::settings::tests::get_settings().await;
// //let settings = crate::settings::Settings::new().unwrap();
// settings.create_db().await;
//
// let db = crate::db::sqlx_postgres::Postgres::init(&settings.database.url).await;
// db.migrate().await;
// let db = BillingDBPostgresAdapter::new(db.pool.clone());
//
// let simple_query = SimpleLoggingQuery {};
//
// let queries: Vec<Box<dyn Query<Store>>> =
// vec![Box::new(simple_query), Box::new(db.clone())];
//
// let services = BillingServicesBuilder::default()
// .add_store(Arc::new(
// AddStoreServiceBuilder::default()
// .db_store_id_exists(Arc::new(db.clone()))
// .db_store_name_exists(Arc::new(db.clone()))
// .get_uuid(Arc::new(crate::utils::uuid::GenerateUUID {}))
// .build()
// .unwrap(),
// ))
// .add_category(mock_add_category_service(
// IS_NEVER_CALLED,
// AddCategoryCommand::new("foo".into(), None, UUID, UUID).unwrap(),
// ))
// .add_product(mock_add_product_service(IS_NEVER_CALLED, get_command()))
// .add_customization(mock_add_customization_service(
// IS_NEVER_CALLED,
// add_customization_command::tests::get_command(),
// ))
// .update_product(mock_update_product_service(
// IS_NEVER_CALLED,
// update_product_command::tests::get_command(),
// ))
// .update_customization(mock_update_customization_service(
// IS_NEVER_CALLED,
// get_update_customization_command(),
// ))
// .update_category(mock_update_category_service(
// IS_NEVER_CALLED,
// get_update_category_command(),
// ))
// .update_store(mock_update_store_service(
// IS_NEVER_CALLED,
// get_update_store_cmd(),
// ))
// .build()
// .unwrap();
//
// let (cqrs, _store_query): (
// Arc<PostgresCqrs<Store>>,
// Arc<dyn ViewRepository<StoreView, Store>>,
// ) = (
// Arc::new(postgres_es::postgres_cqrs(
// db.pool.clone(),
// queries,
// Arc::new(services),
// )),
// Arc::new(db.clone()),
// );
//
// let rand = crate::utils::random_string::GenerateRandomString {};
// let cmd = AddStoreCommand::new(rand.get_random(10), None, UUID).unwrap();
// cqrs.execute("", BillingCommand::AddStore(cmd.clone()))
// .await
// .unwrap();
//
// settings.drop_db().await;
// }
//}

View file

@ -1,3 +1,4 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net> // SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
// //
// SPDX-License-Identifier: AGPL-3.0-or-later // SPDX-License-Identifier: AGPL-3.0-or-later
mod db;

View file

@ -0,0 +1,55 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use mockall::predicate::*;
use mockall::*;
use uuid::Uuid;
use crate::billing::domain::store_aggregate::Store;
use super::errors::*;
#[cfg(test)]
#[allow(unused_imports)]
pub use tests::*;
#[automock]
#[async_trait::async_trait]
pub trait StoreIDExistsDBPort: Send + Sync {
async fn store_id_exists(&self, store_id: &Uuid) -> BillingDBResult<bool>;
}
pub type StoreIDExistsDBPortObj = std::sync::Arc<dyn StoreIDExistsDBPort>;
#[cfg(test)]
pub mod tests {
use super::*;
use std::sync::Arc;
pub fn mock_store_id_exists_db_port_false(times: Option<usize>) -> StoreIDExistsDBPortObj {
let mut m = MockStoreIDExistsDBPort::new();
if let Some(times) = times {
m.expect_store_id_exists()
.times(times)
.returning(|_| Ok(false));
} else {
m.expect_store_id_exists().returning(|_| Ok(false));
}
Arc::new(m)
}
pub fn mock_store_id_exists_db_port_true(times: Option<usize>) -> StoreIDExistsDBPortObj {
let mut m = MockStoreIDExistsDBPort::new();
if let Some(times) = times {
m.expect_store_id_exists()
.times(times)
.returning(|_| Ok(true));
} else {
m.expect_store_id_exists().returning(|_| Ok(true));
}
Arc::new(m)
}
}

View file

@ -0,0 +1,54 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use mockall::predicate::*;
use mockall::*;
use crate::billing::domain::store_aggregate::Store;
use super::errors::*;
#[cfg(test)]
#[allow(unused_imports)]
pub use tests::*;
#[automock]
#[async_trait::async_trait]
pub trait StoreNameExistsDBPort: Send + Sync {
async fn store_name_exists(&self, s: &Store) -> BillingDBResult<bool>;
}
pub type StoreNameExistsDBPortObj = std::sync::Arc<dyn StoreNameExistsDBPort>;
#[cfg(test)]
pub mod tests {
use super::*;
use std::sync::Arc;
pub fn mock_store_name_exists_db_port_false(times: Option<usize>) -> StoreNameExistsDBPortObj {
let mut m = MockStoreNameExistsDBPort::new();
if let Some(times) = times {
m.expect_store_name_exists()
.times(times)
.returning(|_| Ok(false));
} else {
m.expect_store_name_exists().returning(|_| Ok(false));
}
Arc::new(m)
}
pub fn mock_store_name_exists_db_port_true(times: Option<usize>) -> StoreNameExistsDBPortObj {
let mut m = MockStoreNameExistsDBPort::new();
if let Some(times) = times {
m.expect_store_name_exists()
.times(times)
.returning(|_| Ok(true));
} else {
m.expect_store_name_exists().returning(|_| Ok(true));
}
Arc::new(m)
}
}

View file

@ -0,0 +1,149 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use std::sync::Arc;
use derive_builder::Builder;
use mockall::predicate::*;
use mockall::*;
use super::errors::*;
use crate::billing::{
application::port::output::db::{store_id_exists::*, store_name_exists::*},
domain::{
add_store_command::AddStoreCommand,
store_added_event::{StoreAddedEvent, StoreAddedEventBuilder},
store_aggregate::*,
},
};
use crate::utils::uuid::*;
#[automock]
#[async_trait::async_trait]
pub trait AddStoreUseCase: Send + Sync {
async fn add_store(&self, cmd: AddStoreCommand) -> BillResult<StoreAddedEvent>;
}
pub type AddStoreServiceObj = Arc<dyn AddStoreUseCase>;
#[derive(Clone, Builder)]
pub struct AddStoreService {
db_store_id_exists: StoreIDExistsDBPortObj,
db_store_name_exists: StoreNameExistsDBPortObj,
get_uuid: GetUUIDInterfaceObj,
}
#[async_trait::async_trait]
impl AddStoreUseCase for AddStoreService {
async fn add_store(&self, cmd: AddStoreCommand) -> BillResult<StoreAddedEvent> {
let mut store_id = self.get_uuid.get_uuid();
loop {
if self.db_store_id_exists.store_id_exists(&store_id).await? {
store_id = self.get_uuid.get_uuid();
continue;
} else {
break;
}
}
let store = StoreBuilder::default()
.name(cmd.name().into())
.address(cmd.address().as_ref().map(|s| s.to_string()))
.owner(*cmd.owner())
.store_id(store_id)
.build()
.unwrap();
if self.db_store_name_exists.store_name_exists(&store).await? {
return Err(BillError::DuplicateStoreName);
}
Ok(StoreAddedEventBuilder::default()
.name(store.name().into())
.address(store.address().as_ref().map(|s| s.to_string()))
.owner(*cmd.owner())
.store_id(store_id)
.build()
.unwrap())
}
}
#[cfg(test)]
pub mod tests {
use super::*;
use crate::tests::bdd::*;
use crate::utils::uuid::tests::*;
pub fn mock_add_store_service(
times: Option<usize>,
cmd: AddStoreCommand,
) -> AddStoreServiceObj {
let mut m = MockAddStoreUseCase::new();
let res = StoreAddedEventBuilder::default()
.name(cmd.name().into())
.address(cmd.address().as_ref().map(|s| s.to_string()))
.owner(*cmd.owner())
.store_id(UUID)
.build()
.unwrap();
if let Some(times) = times {
m.expect_add_store()
.times(times)
.returning(move |_| Ok(res.clone()));
} else {
m.expect_add_store().returning(move |_| Ok(res.clone()));
}
Arc::new(m)
}
#[actix_rt::test]
async fn test_service_store_id_doesnt_exist() {
let name = "foo";
let address = "bar";
let owner = UUID;
// address = None
let cmd = AddStoreCommand::new(name.into(), Some(address.into()), owner).unwrap();
let s = AddStoreServiceBuilder::default()
.db_store_id_exists(mock_store_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
.db_store_name_exists(mock_store_name_exists_db_port_false(IS_CALLED_ONLY_ONCE))
.get_uuid(mock_get_uuid(IS_CALLED_ONLY_ONCE))
.build()
.unwrap();
let res = s.add_store(cmd.clone()).await.unwrap();
assert_eq!(res.name(), cmd.name());
assert_eq!(res.address(), cmd.address());
assert_eq!(res.owner(), cmd.owner());
assert_eq!(res.store_id(), &UUID);
}
#[actix_rt::test]
async fn test_service_store_name_exists() {
let name = "foo";
let address = "bar";
let owner = UUID;
// address = None
let cmd = AddStoreCommand::new(name.into(), Some(address.into()), owner).unwrap();
let s = AddStoreServiceBuilder::default()
.db_store_id_exists(mock_store_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
.db_store_name_exists(mock_store_name_exists_db_port_true(IS_CALLED_ONLY_ONCE))
.get_uuid(mock_get_uuid(IS_CALLED_ONLY_ONCE))
.build()
.unwrap();
assert_eq!(
s.add_store(cmd.clone()).await,
Err(BillError::DuplicateStoreName)
);
}
}

View file

@ -0,0 +1,146 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use std::sync::Arc;
use derive_builder::Builder;
use mockall::predicate::*;
use mockall::*;
use super::errors::*;
use crate::billing::{
application::port::output::db::{store_id_exists::*, store_name_exists::*},
domain::{
store_aggregate::*, store_updated_event::*, update_store_command::UpdateStoreCommand,
},
};
use crate::utils::uuid::*;
#[automock]
#[async_trait::async_trait]
pub trait UpdateStoreUseCase: Send + Sync {
async fn update_store(&self, cmd: UpdateStoreCommand) -> BillResult<StoreUpdatedEvent>;
}
pub type UpdateStoreServiceObj = Arc<dyn UpdateStoreUseCase>;
#[derive(Clone, Builder)]
pub struct UpdateStoreService {
db_store_id_exists: StoreIDExistsDBPortObj,
db_store_name_exists: StoreNameExistsDBPortObj,
}
#[async_trait::async_trait]
impl UpdateStoreUseCase for UpdateStoreService {
async fn update_store(&self, cmd: UpdateStoreCommand) -> BillResult<StoreUpdatedEvent> {
if !self
.db_store_id_exists
.store_id_exists(cmd.old_store().store_id())
.await?
{
return Err(BillError::StoreIDNotFound);
}
let store = StoreBuilder::default()
.name(cmd.name().into())
.address(cmd.address().as_ref().map(|s| s.to_string()))
.owner(*cmd.owner())
.store_id(*cmd.old_store().store_id())
.build()
.unwrap();
if cmd.name() != cmd.old_store().name() {
if self.db_store_name_exists.store_name_exists(&store).await? {
return Err(BillError::DuplicateStoreName);
}
}
Ok(StoreUpdatedEventBuilder::default()
.added_by_user(*cmd.adding_by())
.new_store(store)
.old_store(cmd.old_store().clone())
.build()
.unwrap())
}
}
#[cfg(test)]
pub mod tests {
use super::*;
use crate::billing::domain::store_updated_event::tests::get_store_updated_event_from_command;
use crate::billing::domain::update_store_command::tests::get_update_store_cmd;
use crate::tests::bdd::*;
use crate::utils::uuid::tests::*;
pub fn mock_update_store_service(
times: Option<usize>,
cmd: UpdateStoreCommand,
) -> UpdateStoreServiceObj {
let mut m = MockUpdateStoreUseCase::new();
let res = get_store_updated_event_from_command(&cmd);
if let Some(times) = times {
m.expect_update_store()
.times(times)
.returning(move |_| Ok(res.clone()));
} else {
m.expect_update_store().returning(move |_| Ok(res.clone()));
}
Arc::new(m)
}
#[actix_rt::test]
async fn test_service() {
let cmd = get_update_store_cmd();
let s = UpdateStoreServiceBuilder::default()
.db_store_id_exists(mock_store_id_exists_db_port_true(IS_CALLED_ONLY_ONCE))
.db_store_name_exists(mock_store_name_exists_db_port_false(IS_CALLED_ONLY_ONCE))
.build()
.unwrap();
let res = s.update_store(cmd.clone()).await.unwrap();
assert_eq!(res.new_store().name(), cmd.name());
assert_eq!(res.new_store().address(), cmd.address());
assert_eq!(res.new_store().owner(), cmd.owner());
assert_eq!(res.new_store().store_id(), cmd.old_store().store_id());
assert_eq!(res.old_store(), cmd.old_store());
assert_eq!(res.added_by_user(), cmd.adding_by());
}
#[actix_rt::test]
async fn test_service_store_name_exists() {
let cmd = get_update_store_cmd();
let s = UpdateStoreServiceBuilder::default()
.db_store_id_exists(mock_store_id_exists_db_port_true(IS_CALLED_ONLY_ONCE))
.db_store_name_exists(mock_store_name_exists_db_port_true(IS_CALLED_ONLY_ONCE))
.build()
.unwrap();
assert_eq!(
s.update_store(cmd.clone()).await,
Err(BillError::DuplicateStoreName)
);
}
#[actix_rt::test]
async fn test_service_store_id_doesnt_exist() {
let cmd = get_update_store_cmd();
let s = UpdateStoreServiceBuilder::default()
.db_store_id_exists(mock_store_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
.db_store_name_exists(mock_store_name_exists_db_port_false(IS_NEVER_CALLED))
.build()
.unwrap();
assert_eq!(
s.update_store(cmd.clone()).await,
Err(BillError::StoreIDNotFound)
);
}
}

View file

@ -0,0 +1,82 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use derive_getters::Getters;
use derive_more::{Display, Error};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
#[derive(Debug, Error, Display, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub enum AddStoreCommandError {
NameIsEmpty,
}
#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters)]
pub struct AddStoreCommand {
name: String,
address: Option<String>,
owner: Uuid,
}
impl AddStoreCommand {
pub fn new(
name: String,
address: Option<String>,
owner: Uuid,
) -> Result<Self, AddStoreCommandError> {
let address: Option<String> = if let Some(address) = address {
let address = address.trim();
if address.is_empty() {
None
} else {
Some(address.to_owned())
}
} else {
None
};
let name = name.trim().to_owned();
if name.is_empty() {
return Err(AddStoreCommandError::NameIsEmpty);
}
Ok(Self {
name,
address,
owner,
})
}
}
#[cfg(test)]
mod tests {
use crate::utils::uuid::tests::UUID;
use super::*;
#[test]
fn test_cmd() {
let name = "foo";
let address = "bar";
let owner = UUID;
// address = None
let cmd = AddStoreCommand::new(name.into(), None, owner).unwrap();
assert_eq!(cmd.name(), name);
assert_eq!(cmd.address(), &None);
assert_eq!(cmd.owner(), &owner);
// address = Some
let cmd = AddStoreCommand::new(name.into(), Some(address.into()), owner).unwrap();
assert_eq!(cmd.name(), name);
assert_eq!(cmd.address(), &Some(address.to_owned()));
assert_eq!(cmd.owner(), &owner);
// AddStoreCommandError::NameIsEmpty
assert_eq!(
AddStoreCommand::new("".into(), Some(address.into()), owner),
Err(AddStoreCommandError::NameIsEmpty)
)
}
}

View file

@ -0,0 +1,18 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use derive_builder::Builder;
use derive_getters::Getters;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
#[derive(
Clone, Debug, Builder, Serialize, Deserialize, Getters, Eq, PartialEq, Ord, PartialOrd,
)]
pub struct StoreAddedEvent {
name: String,
address: Option<String>,
owner: Uuid,
store_id: Uuid,
}

View file

@ -0,0 +1,146 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use async_trait::async_trait;
use cqrs_es::Aggregate;
use derive_builder::Builder;
use derive_getters::Getters;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::billing::application::services::errors::*;
use crate::billing::application::services::BillServicesInterface;
use super::{commands::BillingCommand, events::BillingEvent};
#[derive(
Clone, Default, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Builder, Getters,
)]
pub struct Store {
name: String,
address: Option<String>,
owner: Uuid,
store_id: Uuid,
#[builder(default = "false")]
deleted: bool,
}
#[async_trait]
impl Aggregate for Store {
type Command = BillingCommand;
type Event = BillingEvent;
type Error = BillError;
type Services = std::sync::Arc<dyn BillServicesInterface>;
// This identifier should be unique to the system.
fn aggregate_type() -> String {
"billing.store".to_string()
}
// The aggregate logic goes here. Note that this will be the _bulk_ of a CQRS system
// so expect to use helper functions elsewhere to keep the code clean.
async fn handle(
&self,
command: Self::Command,
services: &Self::Services,
) -> Result<Vec<Self::Event>, Self::Error> {
match command {
BillingCommand::AddStore(cmd) => {
let res = services.add_store().add_store(cmd).await?;
Ok(vec![BillingEvent::StoreAdded(res)])
}
BillingCommand::UpdateStore(cmd) => {
let res = services.update_store().update_store(cmd).await?;
Ok(vec![BillingEvent::StoreUpdated(res)])
}
_ => Ok(Vec::default()),
}
}
fn apply(&mut self, event: Self::Event) {
match event {
BillingEvent::StoreAdded(e) => {
self.name = e.name().into();
self.address = e.address().as_ref().map(|s| s.to_string());
self.owner = *e.owner();
self.store_id = *e.store_id();
self.deleted = false;
}
BillingEvent::StoreUpdated(e) => *self = e.new_store().clone(),
_ => (),
}
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use cqrs_es::test::TestFramework;
use update_store_service::tests::mock_update_store_service;
use super::*;
use crate::billing::{
application::services::{add_store_service::tests::*, *},
domain::{
add_store_command::*, commands::BillingCommand, events::BillingEvent,
store_added_event::*, store_updated_event::tests::get_store_updated_event_from_command,
update_store_command::tests::get_update_store_cmd,
},
};
use crate::tests::bdd::*;
use crate::utils::uuid::tests::*;
// A test framework that will apply our events and command
// and verify that the logic works as expected.
type StoreTestFramework = TestFramework<Store>;
#[test]
fn test_create_store() {
let name = "store_name";
let address = Some("store_address".to_string());
let owner = UUID;
let store_id = UUID;
let expected = StoreAddedEventBuilder::default()
.name(name.into())
.address(address.clone())
.store_id(store_id)
.owner(owner)
.build()
.unwrap();
let expected = BillingEvent::StoreAdded(expected);
let cmd = AddStoreCommand::new(name.into(), address.clone(), owner).unwrap();
let mut services = MockBillServicesInterface::new();
services
.expect_add_store()
.times(IS_CALLED_ONLY_ONCE.unwrap())
.return_const(mock_add_store_service(IS_CALLED_ONLY_ONCE, cmd.clone()));
StoreTestFramework::with(Arc::new(services))
.given_no_previous_events()
.when(BillingCommand::AddStore(cmd))
.then_expect_events(vec![expected]);
}
#[test]
fn test_update_store() {
let cmd = get_update_store_cmd();
let expected = BillingEvent::StoreUpdated(get_store_updated_event_from_command(&cmd));
let mut services = MockBillServicesInterface::new();
services
.expect_update_store()
.times(IS_CALLED_ONLY_ONCE.unwrap())
.return_const(mock_update_store_service(IS_CALLED_ONLY_ONCE, cmd.clone()));
StoreTestFramework::with(Arc::new(services))
.given_no_previous_events()
.when(BillingCommand::UpdateStore(cmd))
.then_expect_events(vec![expected]);
}
}

View file

@ -0,0 +1,43 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use derive_builder::Builder;
use derive_getters::Getters;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use super::store_aggregate::*;
#[derive(
Clone, Debug, Builder, Serialize, Deserialize, Getters, Eq, PartialEq, Ord, PartialOrd,
)]
pub struct StoreUpdatedEvent {
added_by_user: Uuid,
old_store: Store,
new_store: Store,
}
#[cfg(test)]
pub mod tests {
use crate::billing::domain::update_store_command::UpdateStoreCommand;
use super::*;
pub fn get_store_updated_event_from_command(cmd: &UpdateStoreCommand) -> StoreUpdatedEvent {
let new_store = StoreBuilder::default()
.name(cmd.name().into())
.address(cmd.address().as_ref().map(|s| s.to_string()))
.owner(*cmd.owner())
.store_id(*cmd.old_store().store_id())
.build()
.unwrap();
StoreUpdatedEventBuilder::default()
.new_store(new_store)
.old_store(cmd.old_store().clone())
.added_by_user(*cmd.adding_by())
.build()
.unwrap()
}
}

View file

@ -0,0 +1,127 @@
// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
use derive_getters::Getters;
use derive_more::{Display, Error};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use super::store_aggregate::*;
#[derive(Debug, Error, Display, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub enum UpdateStoreCommandError {
NameIsEmpty,
}
#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters)]
pub struct UpdateStoreCommand {
name: String,
address: Option<String>,
owner: Uuid,
old_store: Store,
adding_by: Uuid,
}
impl UpdateStoreCommand {
pub fn new(
name: String,
address: Option<String>,
owner: Uuid,
old_store: Store,
adding_by: Uuid,
) -> Result<Self, UpdateStoreCommandError> {
let address: Option<String> = if let Some(address) = address {
let address = address.trim();
if address.is_empty() {
None
} else {
Some(address.to_owned())
}
} else {
None
};
let name = name.trim().to_owned();
if name.is_empty() {
return Err(UpdateStoreCommandError::NameIsEmpty);
}
Ok(Self {
name,
address,
owner,
old_store,
adding_by,
})
}
}
#[cfg(test)]
pub mod tests {
use crate::utils::uuid::tests::UUID;
use super::*;
pub fn get_update_store_cmd() -> UpdateStoreCommand {
let name = "foo";
let address = "bar";
let owner = UUID;
let adding_by = UUID;
let old_store = Store::default();
UpdateStoreCommand::new(
name.into(),
Some(address.into()),
owner,
old_store.clone(),
adding_by,
)
.unwrap()
}
#[test]
fn test_cmd() {
let name = "foo";
let address = "bar";
let owner = UUID;
let old_store = Store::default();
let adding_by = Uuid::new_v4();
// address = None
let cmd = UpdateStoreCommand::new(name.into(), None, owner, old_store.clone(), adding_by)
.unwrap();
assert_eq!(cmd.name(), name);
assert_eq!(cmd.address(), &None);
assert_eq!(cmd.owner(), &owner);
assert_eq!(cmd.old_store(), &old_store);
assert_eq!(cmd.adding_by(), &adding_by);
// address = Some
let cmd = UpdateStoreCommand::new(
name.into(),
Some(address.into()),
owner,
old_store.clone(),
adding_by,
)
.unwrap();
assert_eq!(cmd.name(), name);
assert_eq!(cmd.address(), &Some(address.to_owned()));
assert_eq!(cmd.owner(), &owner);
assert_eq!(cmd.old_store(), &old_store);
assert_eq!(cmd.adding_by(), &adding_by);
// UpdateStoreCommandError::NameIsEmpty
assert_eq!(
UpdateStoreCommand::new(
"".into(),
Some(address.into()),
owner,
old_store.clone(),
adding_by
),
Err(UpdateStoreCommandError::NameIsEmpty)
)
}
}