src/ordering/adapters/output/db/errors.rs

@@ -0,0 +1,84 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+// 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::ordering::application::port::output::db::errors::OrderingDBError;
+
+impl From<SqlxError> for OrderingDBError {
+    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_ordering_store_query_product_id_key") {
+                    return Self::DuplicateLineItemID;
+                } 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: OrderingDBError) -> OrderingDBError {
+//    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),
+        }
+    }
+}

src/ordering/adapters/output/db/line_item_id_exists.rs

@@ -0,0 +1,97 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use uuid::Uuid;
+
+use super::OrderingDBPostgresAdapter;
+use crate::ordering::application::port::output::db::{errors::*, line_item_id_exists::*};
+
+#[async_trait::async_trait]
+impl LineItemIDExistsDBPort for OrderingDBPostgresAdapter {
+    async fn line_item_id_exists(&self, line_item_id: &Uuid) -> OrderingDBResult<bool> {
+        let res = sqlx::query!(
+            "SELECT EXISTS (
+                SELECT 1
+                    FROM cqrs_ordering_line_item_query
+                WHERE
+                    line_item_id = $1
+            );",
+            line_item_id
+        )
+        .fetch_one(&self.pool)
+        .await?;
+        if let Some(x) = res.exists {
+            Ok(x)
+        } else {
+            Ok(false)
+        }
+    }
+}
+
+#[cfg(test)]
+pub mod tests {
+
+    use super::*;
+    //    use crate::ordering::domain::add_product_command::tests::get_customizations;
+    use crate::ordering::domain::line_item_aggregate::*;
+
+    async fn create_dummy_line_item(line_item: &LineItem, db: &OrderingDBPostgresAdapter) {
+        sqlx::query!(
+            "INSERT INTO cqrs_ordering_line_item_query (
+                    version,
+                    product_name,
+                    product_id,
+                    line_item_id,
+                    quantity_minor_unit,
+                    quantity_minor_number,
+                    quantity_major_unit,
+                    quantity_major_number,
+                    deleted
+                ) VALUES (
+                    $1, $2, $3, $4, $5, $6, $7, $8, $9
+                );",
+            1,
+            line_item.product_name(),
+            line_item.product_id(),
+            line_item.line_item_id(),
+            line_item.quantity().major().unit().to_string(),
+            line_item.quantity().major().number().clone() as i32,
+            line_item.quantity().minor().unit().to_string(),
+            line_item.quantity().minor().number().clone() as i32,
+            line_item.deleted().clone(),
+        )
+        .execute(&db.pool)
+        .await
+        .unwrap();
+    }
+
+    #[actix_rt::test]
+    async fn test_postgres_product_exists() {
+        let settings = crate::settings::tests::get_settings().await;
+        settings.create_db().await;
+        let db = super::OrderingDBPostgresAdapter::new(
+            sqlx::postgres::PgPool::connect(&settings.database.url)
+                .await
+                .unwrap(),
+        );
+
+        let line_item = LineItem::default();
+
+        // state doesn't exist
+        assert!(!db
+            .line_item_id_exists(line_item.line_item_id())
+            .await
+            .unwrap());
+
+        create_dummy_line_item(&line_item, &db).await;
+
+        // state exists
+        assert!(db
+            .line_item_id_exists(line_item.product_id())
+            .await
+            .unwrap());
+
+        settings.drop_db().await;
+    }
+}

src/ordering/adapters/output/db/mod.rs

@@ -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;
+pub mod line_item_id_exists;
+mod line_item_view;
+
+#[derive(Clone)]
+pub struct OrderingDBPostgresAdapter {
+    pool: PgPool,
+}
+
+impl OrderingDBPostgresAdapter {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+
+    pub fn migratable(&self) -> Arc<dyn RunMigrations> {
+        Arc::new(Postgres::new(self.pool.clone()))
+    }
+}

src/ordering/adapters/output/mod.rs

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

src/ordering/application/mod.rs

@@ -2,5 +2,5 @@
 //
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-mod port;
+pub mod port;
-mod services;
+pub mod services;

src/ordering/application/port/mod.rs

@@ -3,4 +3,4 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 mod input;
-mod output;
+pub mod output;

src/ordering/application/port/output/db/errors.rs

@@ -0,0 +1,15 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use derive_more::Display;
+use serde::{Deserialize, Serialize};
+
+pub type OrderingDBResult<V> = Result<V, OrderingDBError>;
+
+#[derive(Debug, Display, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub enum OrderingDBError {
+    DuplicateLineItemID,
+    LineItemIDNotFound,
+    InternalError,
+}

src/ordering/application/port/output/db/line_item_id_exists.rs

@@ -0,0 +1,57 @@
+// 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 super::errors::*;
+#[cfg(test)]
+#[allow(unused_imports)]
+pub use tests::*;
+
+#[automock]
+#[async_trait::async_trait]
+pub trait LineItemIDExistsDBPort: Send + Sync {
+    async fn line_item_id_exists(&self, line_item_id: &Uuid) -> OrderingDBResult<bool>;
+}
+
+pub type LineItemIDExistsDBPortObj = std::sync::Arc<dyn LineItemIDExistsDBPort>;
+
+#[cfg(test)]
+pub mod tests {
+    use super::*;
+
+    use std::sync::Arc;
+
+    pub fn mock_line_item_id_exists_db_port_false(
+        times: Option<usize>,
+    ) -> LineItemIDExistsDBPortObj {
+        let mut m = MockLineItemIDExistsDBPort::new();
+        if let Some(times) = times {
+            m.expect_line_item_id_exists()
+                .times(times)
+                .returning(|_| Ok(false));
+        } else {
+            m.expect_line_item_id_exists().returning(|_| Ok(false));
+        }
+
+        Arc::new(m)
+    }
+
+    pub fn mock_line_item_id_exists_db_port_true(
+        times: Option<usize>,
+    ) -> LineItemIDExistsDBPortObj {
+        let mut m = MockLineItemIDExistsDBPort::new();
+        if let Some(times) = times {
+            m.expect_line_item_id_exists()
+                .times(times)
+                .returning(|_| Ok(true));
+        } else {
+            m.expect_line_item_id_exists().returning(|_| Ok(true));
+        }
+
+        Arc::new(m)
+    }
+}

src/ordering/application/port/output/db/mod.rs

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

src/ordering/application/port/output/mod.rs

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