Merge pull request 'feat: order: LineItem add and update cmd,events&service' (#59) from ordering-line-item into master

Reviewed-on: #59

.sqlx/query-1e7df92c508fac4c32c00621b099c673d8745b8d145b603807c771906a7af756.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO cqrs_ordering_line_item_query (\n                        version,\n                        product_name,\n                        product_id,\n                        line_item_id,\n                        quantity_minor_unit,\n                        quantity_minor_number,\n                        quantity_major_unit,\n                        quantity_major_number,\n\n                        deleted\n                    ) VALUES (\n                        $1, $2, $3, $4, $5, $6, $7, $8, $9\n                    );",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Text",
+        "Uuid",
+        "Uuid",
+        "Text",
+        "Int4",
+        "Text",
+        "Int4",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1e7df92c508fac4c32c00621b099c673d8745b8d145b603807c771906a7af756"
+}

.sqlx/query-3f2b3da434c433067e0a68fa98ad0d7b1a00a836682d9ef20fd58c72bd5115f5.json

@@ -0,0 +1,64 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT \n                product_name,\n                product_id,\n                line_item_id,\n                quantity_minor_unit,\n                quantity_minor_number,\n                quantity_major_unit,\n                quantity_major_number,\n                deleted\n             FROM\n                cqrs_ordering_line_item_query\n            WHERE\n                line_item_id = $1;",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "product_name",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "product_id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 2,
+        "name": "line_item_id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 3,
+        "name": "quantity_minor_unit",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
+        "name": "quantity_minor_number",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 5,
+        "name": "quantity_major_unit",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 6,
+        "name": "quantity_major_number",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 7,
+        "name": "deleted",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "3f2b3da434c433067e0a68fa98ad0d7b1a00a836682d9ef20fd58c72bd5115f5"
+}

.sqlx/query-43c79c431bea6029ee2d2dc5997ab00ee6db214f683a1c877b08e27381148a91.json

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

.sqlx/query-46fddc14a06f84a15fbcc04cfff9d3f41e03c73e2db7d7ee39a0f3e86cc38fe9.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE\n                        cqrs_ordering_line_item_query\n                    SET\n                        version = $1,\n                        product_name = $2,\n                        product_id = $3,\n                        line_item_id = $4,\n                        quantity_minor_unit = $5,\n                        quantity_minor_number = $6,\n                        quantity_major_unit = $7,\n                        quantity_major_number = $8,\n                        deleted = $9;",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Text",
+        "Uuid",
+        "Uuid",
+        "Text",
+        "Int4",
+        "Text",
+        "Int4",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "46fddc14a06f84a15fbcc04cfff9d3f41e03c73e2db7d7ee39a0f3e86cc38fe9"
+}

.sqlx/query-da317770b9f4874f805c9edea8a7fed98c2186a30ac370864a4db2d64cc14b75.json

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

.sqlx/query-ea58498982a42d17c51828387df4b446f9cdc89986fd49c8f9d736eeeda12f48.json

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

migrations/20240723095912_cqrs_ordering_line_item_query.sql

@@ -0,0 +1,23 @@
+-- SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+--
+-- SPDX-License-Identifier: AGPL-3.0-or-later
+
+CREATE TABLE IF NOT EXISTS cqrs_ordering_line_item_query
+(
+    version           bigint CHECK (version >= 0) NOT NULL,
+
+    sale_time timestamp with time zone DEFAULT (CURRENT_TIMESTAMP),
+	line_item_id UUID NOT NULL UNIQUE,
+
+    product_name TEXT NOT NULL,
+	product_id UUID NOT NULL,
+
+    quantity_major_number INTEGER NOT NULL,
+    quantity_minor_number INTEGER NOT NULL,
+    quantity_major_unit TEXT NOT NULL,
+    quantity_minor_unit TEXT NOT NULL,
+
+	deleted BOOLEAN NOT NULL DEFAULT FALSE,
+
+    PRIMARY KEY (line_item_id)
+);

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/line_item_view.rs

@@ -0,0 +1,269 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use std::str::FromStr;
+
+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::OrderingDBPostgresAdapter;
+use crate::ordering::domain::events::OrderingEvent;
+use crate::ordering::domain::line_item_aggregate::*;
+use crate::types::quantity::*;
+use crate::utils::parse_aggregate_id::parse_aggregate_id;
+
+pub const NEW_LINE_ITEM_NON_UUID: &str = "new_line_item_non_uuid-asdfa";
+
+// The view for a LineItem 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 LineItemView {
+    product_name: String,
+    product_id: Uuid,
+
+    line_item_id: Uuid,
+
+    quantity_major_number: i32,
+    quantity_minor_number: i32,
+    quantity_major_unit: String,
+    quantity_minor_unit: String,
+
+    deleted: bool,
+}
+
+impl From<LineItemView> for LineItem {
+    fn from(v: LineItemView) -> Self {
+        let quantity = QuantityBuilder::default()
+            .minor(
+                QuantityPartBuilder::default()
+                    .number(v.quantity_minor_number as usize)
+                    .unit(QuantityUnit::from_str(&v.quantity_minor_unit).unwrap())
+                    .build()
+                    .unwrap(),
+            )
+            .major(
+                QuantityPartBuilder::default()
+                    .number(v.quantity_major_number as usize)
+                    .unit(QuantityUnit::from_str(&v.quantity_major_unit).unwrap())
+                    .build()
+                    .unwrap(),
+            )
+            .build()
+            .unwrap();
+
+        LineItemBuilder::default()
+            .product_name(v.product_name)
+            .line_item_id(v.line_item_id)
+            .quantity(quantity)
+            .product_id(v.product_id)
+            .deleted(v.deleted)
+            .build()
+            .unwrap()
+    }
+}
+
+// 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<LineItem> for LineItemView {
+    fn update(&mut self, event: &EventEnvelope<LineItem>) {
+        match &event.payload {
+            OrderingEvent::LineItemAdded(val) => {
+                self.product_name = val.line_item().product_name().into();
+                self.product_id = *val.line_item().product_id();
+                self.line_item_id = *val.line_item().line_item_id();
+
+                self.quantity_major_number = *val.line_item().quantity().major().number() as i32;
+                self.quantity_minor_number = *val.line_item().quantity().minor().number() as i32;
+                self.quantity_major_unit = val.line_item().quantity().major().unit().to_string();
+                self.quantity_minor_unit = val.line_item().quantity().minor().unit().to_string();
+
+                self.deleted = false;
+            }
+            _ => (),
+        }
+    }
+}
+
+#[async_trait]
+impl ViewRepository<LineItemView, LineItem> for OrderingDBPostgresAdapter {
+    async fn load(&self, line_item_id: &str) -> Result<Option<LineItemView>, PersistenceError> {
+        let line_item_id = match parse_aggregate_id(line_item_id, NEW_LINE_ITEM_NON_UUID)? {
+            Some((val, _)) => return Ok(Some(val)),
+            None => Uuid::parse_str(line_item_id).unwrap(),
+        };
+
+        let res = sqlx::query_as!(
+            LineItemView,
+            "SELECT 
+                product_name,
+                product_id,
+                line_item_id,
+                quantity_minor_unit,
+                quantity_minor_number,
+                quantity_major_unit,
+                quantity_major_number,
+                deleted
+             FROM
+                cqrs_ordering_line_item_query
+            WHERE
+                line_item_id = $1;",
+            line_item_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(PostgresAggregateError::from)?;
+
+        Ok(Some(res))
+    }
+
+    async fn load_with_context(
+        &self,
+        line_item_id: &str,
+    ) -> Result<Option<(LineItemView, ViewContext)>, PersistenceError> {
+        let line_item_id = match parse_aggregate_id(line_item_id, NEW_LINE_ITEM_NON_UUID)? {
+            Some(val) => return Ok(Some(val)),
+            None => Uuid::parse_str(line_item_id).unwrap(),
+        };
+
+        let res = sqlx::query_as!(
+            LineItemView,
+            "SELECT 
+                product_name,
+                product_id,
+                line_item_id,
+                quantity_minor_unit,
+                quantity_minor_number,
+                quantity_major_unit,
+                quantity_major_number,
+                deleted
+             FROM
+                cqrs_ordering_line_item_query
+            WHERE
+                line_item_id = $1;",
+            line_item_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(PostgresAggregateError::from)?;
+
+        struct Context {
+            version: i64,
+            line_item_id: Uuid,
+        }
+
+        let ctx = sqlx::query_as!(
+            Context,
+            "SELECT 
+                line_item_id, version
+             FROM
+                cqrs_ordering_line_item_query
+            WHERE
+                line_item_id = $1;",
+            line_item_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(PostgresAggregateError::from)?;
+
+        let view_context = ViewContext::new(ctx.line_item_id.to_string(), ctx.version);
+        Ok(Some((res, view_context)))
+    }
+
+    async fn update_view(
+        &self,
+        view: LineItemView,
+        context: ViewContext,
+    ) -> Result<(), PersistenceError> {
+        match context.version {
+            0 => {
+                let version = context.version + 1;
+                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
+                    );",
+                    version,
+                    view.product_name,
+                    view.product_id,
+                    view.line_item_id,
+                    view.quantity_minor_unit,
+                    view.quantity_minor_number,
+                    view.quantity_major_unit,
+                    view.quantity_major_number,
+                    view.deleted,
+                )
+                .execute(&self.pool)
+                .await
+                .map_err(PostgresAggregateError::from)?;
+            }
+
+            _ => {
+                let version = context.version + 1;
+                sqlx::query!(
+                    "UPDATE
+                        cqrs_ordering_line_item_query
+                    SET
+                        version = $1,
+                        product_name = $2,
+                        product_id = $3,
+                        line_item_id = $4,
+                        quantity_minor_unit = $5,
+                        quantity_minor_number = $6,
+                        quantity_major_unit = $7,
+                        quantity_major_number = $8,
+                        deleted = $9;",
+                    version,
+                    view.product_name,
+                    view.product_id,
+                    view.line_item_id,
+                    view.quantity_minor_unit,
+                    view.quantity_minor_number,
+                    view.quantity_major_unit,
+                    view.quantity_major_number,
+                    view.deleted,
+                )
+                .execute(&self.pool)
+                .await
+                .map_err(PostgresAggregateError::from)?;
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl Query<LineItem> for OrderingDBPostgresAdapter {
+    async fn dispatch(&self, line_item_id: &str, events: &[EventEnvelope<LineItem>]) {
+        let res = self
+            .load_with_context(line_item_id)
+            .await
+            .unwrap_or_else(|_| {
+                Some((
+                    LineItemView::default(),
+                    ViewContext::new(line_item_id.into(), 0),
+                ))
+            });
+        let (mut view, view_context): (LineItemView, ViewContext) = res.unwrap();
+        for event in events {
+            view.update(event);
+        }
+        self.update_view(view, view_context).await.unwrap();
+    }
+}

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;
-mod services;
+pub mod port;
+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;

src/ordering/application/services/add_line_item_service.rs

@@ -0,0 +1,113 @@
+// 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 time::OffsetDateTime;
+
+use super::errors::*;
+use crate::ordering::{
+    application::port::output::db::line_item_id_exists::*,
+    domain::{add_line_item_command::*, line_item_added_event::*, line_item_aggregate::*},
+};
+use crate::utils::uuid::*;
+
+#[automock]
+#[async_trait::async_trait]
+pub trait AddLineItemUseCase: Send + Sync {
+    async fn add_line_item(&self, cmd: AddLineItemCommand) -> OrderingResult<LineItemAddedEvent>;
+}
+
+pub type AddLineItemServiceObj = Arc<dyn AddLineItemUseCase>;
+
+#[derive(Clone, Builder)]
+pub struct AddLineItemService {
+    db_line_item_id_exists: LineItemIDExistsDBPortObj,
+    get_uuid: GetUUIDInterfaceObj,
+}
+
+#[async_trait::async_trait]
+impl AddLineItemUseCase for AddLineItemService {
+    async fn add_line_item(&self, cmd: AddLineItemCommand) -> OrderingResult<LineItemAddedEvent> {
+        let mut line_item_id = self.get_uuid.get_uuid();
+
+        loop {
+            if self
+                .db_line_item_id_exists
+                .line_item_id_exists(&line_item_id)
+                .await?
+            {
+                line_item_id = self.get_uuid.get_uuid();
+                continue;
+            } else {
+                break;
+            }
+        }
+
+        let line_item = LineItemBuilder::default()
+            .sale_time(OffsetDateTime::now_utc())
+            .product_name(cmd.product_name().into())
+            .product_id(*cmd.product_id())
+            .line_item_id(line_item_id)
+            .quantity(cmd.quantity().clone())
+            .deleted(false)
+            .build()
+            .unwrap();
+
+        Ok(LineItemAddedEventBuilder::default()
+            .added_by_user(*cmd.adding_by())
+            .line_item(line_item)
+            .build()
+            .unwrap())
+    }
+}
+
+#[cfg(test)]
+pub mod tests {
+    use super::*;
+
+    use crate::ordering::domain::line_item_added_event::tests::get_added_line_item_event_from_command;
+    use crate::utils::uuid::tests::UUID;
+    use crate::{tests::bdd::*, utils::uuid::tests::mock_get_uuid};
+
+    pub fn mock_add_line_item_service(
+        times: Option<usize>,
+        cmd: AddLineItemCommand,
+    ) -> AddLineItemServiceObj {
+        let mut m = MockAddLineItemUseCase::new();
+
+        let res = get_added_line_item_event_from_command(&cmd);
+        if let Some(times) = times {
+            m.expect_add_line_item()
+                .times(times)
+                .returning(move |_| Ok(res.clone()));
+        } else {
+            m.expect_add_line_item().returning(move |_| Ok(res.clone()));
+        }
+
+        Arc::new(m)
+    }
+
+    #[actix_rt::test]
+    async fn test_service() {
+        let cmd = AddLineItemCommand::get_cmd();
+
+        let s = AddLineItemServiceBuilder::default()
+            .db_line_item_id_exists(mock_line_item_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
+            .get_uuid(mock_get_uuid(IS_CALLED_ONLY_ONCE))
+            .build()
+            .unwrap();
+
+        let res = s.add_line_item(cmd.clone()).await.unwrap();
+        assert_eq!(res.line_item().product_name(), cmd.product_name());
+        assert_eq!(res.line_item().product_id(), cmd.product_id());
+        assert_eq!(res.line_item().quantity(), cmd.quantity());
+        assert_eq!(res.line_item().quantity(), cmd.quantity());
+        assert!(!res.line_item().deleted());
+        assert_eq!(res.added_by_user(), cmd.adding_by());
+    }
+}

src/ordering/application/services/delete_line_item_service.rs

@@ -0,0 +1,132 @@
+// 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 time::OffsetDateTime;
+
+use super::errors::*;
+use crate::ordering::{
+    application::port::output::db::line_item_id_exists::*,
+    domain::{delete_line_item_command::*, line_item_aggregate::*, line_item_deleted_event::*},
+};
+use crate::utils::uuid::*;
+
+#[automock]
+#[async_trait::async_trait]
+pub trait DeleteLineItemUseCase: Send + Sync {
+    async fn delete_line_item(
+        &self,
+        cmd: DeleteLineItemCommand,
+    ) -> OrderingResult<LineItemDeletedEvent>;
+}
+
+pub type DeleteLineItemServiceObj = Arc<dyn DeleteLineItemUseCase>;
+
+#[derive(Clone, Builder)]
+pub struct DeleteLineItemService {
+    db_line_item_id_exists: LineItemIDExistsDBPortObj,
+}
+
+#[async_trait::async_trait]
+impl DeleteLineItemUseCase for DeleteLineItemService {
+    async fn delete_line_item(
+        &self,
+        cmd: DeleteLineItemCommand,
+    ) -> OrderingResult<LineItemDeletedEvent> {
+        if !self
+            .db_line_item_id_exists
+            .line_item_id_exists(cmd.line_item().line_item_id())
+            .await?
+        {
+            return Err(OrderingError::LineItemIDNotFound);
+        }
+
+        let deleted_line_item = LineItemBuilder::default()
+            .sale_time(cmd.line_item().sale_time().clone())
+            .product_name(cmd.line_item().product_name().into())
+            .product_id(*cmd.line_item().product_id())
+            .line_item_id(*cmd.line_item().line_item_id())
+            .quantity(cmd.line_item().quantity().clone())
+            .deleted(true)
+            .build()
+            .unwrap();
+
+        Ok(LineItemDeletedEventBuilder::default()
+            .added_by_user(*cmd.adding_by())
+            .line_item(deleted_line_item)
+            .build()
+            .unwrap())
+    }
+}
+
+#[cfg(test)]
+pub mod tests {
+    use super::*;
+
+    use crate::ordering::domain::line_item_deleted_event::tests::get_deleted_line_item_event_from_command;
+    use crate::tests::bdd::*;
+
+    pub fn mock_delete_line_item_service(
+        times: Option<usize>,
+        cmd: DeleteLineItemCommand,
+    ) -> DeleteLineItemServiceObj {
+        let mut m = MockDeleteLineItemUseCase::new();
+
+        let res = get_deleted_line_item_event_from_command(&cmd);
+        if let Some(times) = times {
+            m.expect_delete_line_item()
+                .times(times)
+                .returning(move |_| Ok(res.clone()));
+        } else {
+            m.expect_delete_line_item()
+                .returning(move |_| Ok(res.clone()));
+        }
+
+        Arc::new(m)
+    }
+
+    #[actix_rt::test]
+    async fn test_service() {
+        let cmd = DeleteLineItemCommand::get_cmd();
+
+        let s = DeleteLineItemServiceBuilder::default()
+            .db_line_item_id_exists(mock_line_item_id_exists_db_port_true(IS_CALLED_ONLY_ONCE))
+            .build()
+            .unwrap();
+
+        let res = s.delete_line_item(cmd.clone()).await.unwrap();
+        assert_eq!(
+            res.line_item().product_name(),
+            cmd.line_item().product_name()
+        );
+        assert_eq!(res.line_item().product_id(), cmd.line_item().product_id());
+        assert_eq!(res.line_item().quantity(), cmd.line_item().quantity());
+        assert_eq!(
+            res.line_item().line_item_id(),
+            cmd.line_item().line_item_id()
+        );
+        assert!(res.line_item().deleted());
+
+        assert_eq!(res.added_by_user(), cmd.adding_by());
+    }
+
+    #[actix_rt::test]
+    async fn test_service_line_item_id_doesnt_exist() {
+        let cmd = DeleteLineItemCommand::get_cmd();
+
+        let s = DeleteLineItemServiceBuilder::default()
+            .db_line_item_id_exists(mock_line_item_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
+            .build()
+            .unwrap();
+
+        assert_eq!(
+            s.delete_line_item(cmd.clone()).await,
+            Err(OrderingError::LineItemIDNotFound)
+        );
+    }
+}

src/ordering/application/services/errors.rs

@@ -0,0 +1,30 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use derive_more::{Display, Error};
+use log::error;
+use serde::{Deserialize, Serialize};
+
+use crate::ordering::application::port::output::db::errors::OrderingDBError;
+
+pub type OrderingResult<V> = Result<V, OrderingError>;
+
+#[derive(Debug, Error, Display, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub enum OrderingError {
+    LineItemIDNotFound,
+    InternalError,
+}
+//
+impl From<OrderingDBError> for OrderingError {
+    fn from(value: OrderingDBError) -> Self {
+        match value {
+            OrderingDBError::DuplicateLineItemID => {
+                error!("DuplicateLineItemID");
+                Self::InternalError
+            }
+            OrderingDBError::LineItemIDNotFound => OrderingError::LineItemIDNotFound,
+            OrderingDBError::InternalError => Self::InternalError,
+        }
+    }
+}

src/ordering/application/services/mod.rs

@@ -1,3 +1,41 @@
 // SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
 //
 // SPDX-License-Identifier: AGPL-3.0-or-later
+
+use derive_builder::Builder;
+use mockall::predicate::*;
+use mockall::*;
+
+pub mod errors;
+
+//services
+pub mod add_line_item_service;
+pub mod delete_line_item_service;
+pub mod update_line_item_service;
+
+#[automock]
+pub trait OrderingServicesInterface: Send + Sync {
+    fn add_line_item(&self) -> add_line_item_service::AddLineItemServiceObj;
+    fn update_line_item(&self) -> update_line_item_service::UpdateLineItemServiceObj;
+    fn delete_line_item(&self) -> delete_line_item_service::DeleteLineItemServiceObj;
+}
+
+#[derive(Clone, Builder)]
+pub struct OrderingServices {
+    add_line_item: add_line_item_service::AddLineItemServiceObj,
+    update_line_item: update_line_item_service::UpdateLineItemServiceObj,
+    delete_line_item: delete_line_item_service::DeleteLineItemServiceObj,
+}
+
+impl OrderingServicesInterface for OrderingServices {
+    fn add_line_item(&self) -> add_line_item_service::AddLineItemServiceObj {
+        self.add_line_item.clone()
+    }
+    fn update_line_item(&self) -> update_line_item_service::UpdateLineItemServiceObj {
+        self.update_line_item.clone()
+    }
+
+    fn delete_line_item(&self) -> delete_line_item_service::DeleteLineItemServiceObj {
+        self.delete_line_item.clone()
+    }
+}

src/ordering/application/services/update_line_item_service.rs

@@ -0,0 +1,134 @@
+// 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 time::OffsetDateTime;
+
+use super::errors::*;
+use crate::ordering::{
+    application::port::output::db::line_item_id_exists::*,
+    domain::{line_item_aggregate::*, line_item_updated_event::*, update_line_item_command::*},
+};
+use crate::utils::uuid::*;
+
+#[automock]
+#[async_trait::async_trait]
+pub trait UpdateLineItemUseCase: Send + Sync {
+    async fn update_line_item(
+        &self,
+        cmd: UpdateLineItemCommand,
+    ) -> OrderingResult<LineItemUpdatedEvent>;
+}
+
+pub type UpdateLineItemServiceObj = Arc<dyn UpdateLineItemUseCase>;
+
+#[derive(Clone, Builder)]
+pub struct UpdateLineItemService {
+    db_line_item_id_exists: LineItemIDExistsDBPortObj,
+}
+
+#[async_trait::async_trait]
+impl UpdateLineItemUseCase for UpdateLineItemService {
+    async fn update_line_item(
+        &self,
+        cmd: UpdateLineItemCommand,
+    ) -> OrderingResult<LineItemUpdatedEvent> {
+        if !self
+            .db_line_item_id_exists
+            .line_item_id_exists(cmd.old_line_item().line_item_id())
+            .await?
+        {
+            return Err(OrderingError::LineItemIDNotFound);
+        }
+
+        let new_line_item = LineItemBuilder::default()
+            .sale_time(cmd.sale_time().clone())
+            .product_name(cmd.product_name().into())
+            .product_id(*cmd.product_id())
+            .line_item_id(*cmd.old_line_item().line_item_id())
+            .quantity(cmd.quantity().clone())
+            .deleted(false)
+            .build()
+            .unwrap();
+
+        Ok(LineItemUpdatedEventBuilder::default()
+            .added_by_user(*cmd.adding_by())
+            .new_line_item(new_line_item)
+            .old_line_item(cmd.old_line_item().clone())
+            .build()
+            .unwrap())
+    }
+}
+
+#[cfg(test)]
+pub mod tests {
+    use super::*;
+
+    use crate::ordering::domain::line_item_updated_event::tests::get_updated_line_item_event_from_command;
+    use crate::utils::uuid::tests::UUID;
+    use crate::{tests::bdd::*, utils::uuid::tests::mock_get_uuid};
+
+    pub fn mock_update_line_item_service(
+        times: Option<usize>,
+        cmd: UpdateLineItemCommand,
+    ) -> UpdateLineItemServiceObj {
+        let mut m = MockUpdateLineItemUseCase::new();
+
+        let res = get_updated_line_item_event_from_command(&cmd);
+        if let Some(times) = times {
+            m.expect_update_line_item()
+                .times(times)
+                .returning(move |_| Ok(res.clone()));
+        } else {
+            m.expect_update_line_item()
+                .returning(move |_| Ok(res.clone()));
+        }
+
+        Arc::new(m)
+    }
+
+    #[actix_rt::test]
+    async fn test_service() {
+        let cmd = UpdateLineItemCommand::get_cmd();
+
+        let s = UpdateLineItemServiceBuilder::default()
+            .db_line_item_id_exists(mock_line_item_id_exists_db_port_true(IS_CALLED_ONLY_ONCE))
+            .build()
+            .unwrap();
+
+        let res = s.update_line_item(cmd.clone()).await.unwrap();
+        assert_eq!(res.new_line_item().product_name(), cmd.product_name());
+        assert_eq!(res.new_line_item().product_id(), cmd.product_id());
+        assert_eq!(res.new_line_item().quantity(), cmd.quantity());
+        assert_eq!(res.new_line_item().quantity(), cmd.quantity());
+        assert_eq!(
+            res.new_line_item().line_item_id(),
+            cmd.old_line_item().line_item_id()
+        );
+        assert!(!res.new_line_item().deleted());
+
+        assert_eq!(res.old_line_item(), cmd.old_line_item());
+
+        assert_eq!(res.added_by_user(), cmd.adding_by());
+    }
+
+    #[actix_rt::test]
+    async fn test_service_line_item_id_doesnt_exist() {
+        let cmd = UpdateLineItemCommand::get_cmd();
+
+        let s = UpdateLineItemServiceBuilder::default()
+            .db_line_item_id_exists(mock_line_item_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
+            .build()
+            .unwrap();
+
+        assert_eq!(
+            s.update_line_item(cmd.clone()).await,
+            Err(OrderingError::LineItemIDNotFound)
+        );
+    }
+}

src/ordering/domain/add_line_item_command.rs

@@ -0,0 +1,156 @@
+// 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 derive_more::{Display, Error};
+use serde::{Deserialize, Serialize};
+use time::OffsetDateTime;
+use uuid::Uuid;
+
+use crate::types::quantity::*;
+use crate::utils::string::empty_string_err;
+
+#[derive(Debug, Error, Display, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub enum AddLineItemCommandError {
+    QuantityIsEmpty,
+    ProductNameIsEmpty,
+}
+
+#[derive(
+    Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters, Builder,
+)]
+pub struct UnvalidatedAddLineItemCommand {
+    adding_by: Uuid,
+
+    #[builder(default = "OffsetDateTime::now_utc()")]
+    sale_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    quantity: Quantity,
+}
+
+impl UnvalidatedAddLineItemCommand {
+    pub fn validate(self) -> Result<AddLineItemCommand, AddLineItemCommandError> {
+        let product_name = empty_string_err(
+            self.product_name,
+            AddLineItemCommandError::ProductNameIsEmpty,
+        )?;
+
+        if self.quantity.is_empty() {
+            return Err(AddLineItemCommandError::QuantityIsEmpty);
+        }
+
+        Ok(AddLineItemCommand {
+            sale_time: self.sale_time,
+            product_name,
+            product_id: self.product_id,
+            quantity: self.quantity,
+            adding_by: self.adding_by,
+        })
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters)]
+pub struct AddLineItemCommand {
+    sale_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    quantity: Quantity,
+
+    adding_by: Uuid,
+}
+
+#[cfg(test)]
+mod tests {
+    use time::macros::datetime;
+
+    use crate::utils::uuid::tests::UUID;
+
+    use super::*;
+
+    impl AddLineItemCommand {
+        pub fn get_cmd() -> Self {
+            let product_name = "foo";
+            let product_id = UUID;
+            let adding_by = UUID;
+            let quantity = Quantity::get_quantity();
+
+            UnvalidatedAddLineItemCommandBuilder::default()
+                .product_name(product_name.into())
+                .adding_by(adding_by)
+                .sale_time(datetime!(1970-01-01 0:00 UTC))
+                .quantity(quantity.clone())
+                .product_id(product_id)
+                .build()
+                .unwrap()
+                .validate()
+                .unwrap()
+        }
+    }
+
+    #[test]
+    fn test_cmd() {
+        let product_name = "foo";
+        let product_id = UUID;
+        let adding_by = UUID;
+        let quantity = Quantity::get_quantity();
+
+        let cmd = UnvalidatedAddLineItemCommandBuilder::default()
+            .product_name(product_name.into())
+            .adding_by(adding_by)
+            .quantity(quantity.clone())
+            .product_id(product_id)
+            .build()
+            .unwrap()
+            .validate()
+            .unwrap();
+
+        assert_eq!(cmd.quantity(), &quantity);
+        assert_eq!(*cmd.product_id(), product_id);
+        assert_eq!(*cmd.adding_by(), adding_by);
+        assert_eq!(cmd.product_name(), product_name);
+    }
+
+    #[test]
+    fn test_cmd_product_name_empty() {
+        let product_name = "";
+        let product_id = UUID;
+        let adding_by = UUID;
+        let quantity = Quantity::get_quantity();
+
+        assert_eq!(
+            UnvalidatedAddLineItemCommandBuilder::default()
+                .product_name(product_name.into())
+                .adding_by(adding_by)
+                .quantity(quantity.clone())
+                .product_id(product_id)
+                .build()
+                .unwrap()
+                .validate(),
+            Err(AddLineItemCommandError::ProductNameIsEmpty)
+        );
+    }
+
+    #[test]
+    fn test_cmd_quantity_empty() {
+        let product_name = "foo";
+        let product_id = UUID;
+        let adding_by = UUID;
+        // minor = 0; major = 0;
+        let quantity = Quantity::default();
+
+        assert_eq!(
+            UnvalidatedAddLineItemCommandBuilder::default()
+                .product_name(product_name.into())
+                .adding_by(adding_by)
+                .quantity(quantity.clone())
+                .product_id(product_id)
+                .build()
+                .unwrap()
+                .validate(),
+            Err(AddLineItemCommandError::QuantityIsEmpty)
+        );
+    }
+}

src/ordering/domain/commands.rs

@@ -0,0 +1,18 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use mockall::predicate::*;
+use serde::{Deserialize, Serialize};
+
+use super::{
+    add_line_item_command::AddLineItemCommand, delete_line_item_command::DeleteLineItemCommand,
+    update_line_item_command::UpdateLineItemCommand,
+};
+
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd)]
+pub enum OrderingCommand {
+    AddLineItem(AddLineItemCommand),
+    UpdateLineItem(UpdateLineItemCommand),
+    DeleteLineItem(DeleteLineItemCommand),
+}

src/ordering/domain/delete_line_item_command.rs

@@ -0,0 +1,37 @@
+// 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::line_item_aggregate::LineItem;
+
+#[derive(
+    Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Builder, Getters,
+)]
+pub struct DeleteLineItemCommand {
+    adding_by: Uuid,
+    line_item: LineItem,
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::utils::uuid::tests::UUID;
+
+    use super::*;
+
+    impl DeleteLineItemCommand {
+        pub fn get_cmd() -> Self {
+            let adding_by = UUID;
+
+            DeleteLineItemCommandBuilder::default()
+                .adding_by(adding_by)
+                .line_item(LineItem::get_line_item())
+                .build()
+                .unwrap()
+        }
+    }
+}

src/ordering/domain/events.rs

@@ -0,0 +1,34 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use cqrs_es::DomainEvent;
+use serde::{Deserialize, Serialize};
+
+use super::{
+    line_item_added_event::LineItemAddedEvent, line_item_deleted_event::LineItemDeletedEvent,
+    line_item_updated_event::LineItemUpdatedEvent,
+};
+
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd)]
+pub enum OrderingEvent {
+    LineItemAdded(LineItemAddedEvent),
+    LineItemUpdated(LineItemUpdatedEvent),
+    LineItemDeleted(LineItemDeletedEvent),
+}
+
+impl DomainEvent for OrderingEvent {
+    fn event_version(&self) -> String {
+        "1.0".to_string()
+    }
+
+    fn event_type(&self) -> String {
+        let e: &str = match self {
+            OrderingEvent::LineItemAdded { .. } => "OrderingLineItemAdded",
+            OrderingEvent::LineItemUpdated { .. } => "OrderingLineItemUpdated",
+            OrderingEvent::LineItemDeleted { .. } => "OrderingLineItemDeleted",
+        };
+
+        e.to_string()
+    }
+}

src/ordering/domain/kot_aggregate.rs

@@ -0,0 +1,26 @@
+// 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 time::OffsetDateTime;
+use uuid::Uuid;
+
+use super::line_item_aggregate::*;
+
+#[derive(
+    Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Builder, Getters,
+)]
+pub struct Kot {
+    #[builder(default = "OffsetDateTime::now_utc()")]
+    created_time: OffsetDateTime,
+    line_items: Vec<LineItem>,
+    kot_id: Uuid,
+    order_id: Uuid,
+    #[builder(default = "false")]
+    deleted: bool,
+}

src/ordering/domain/line_item_added_event.rs

@@ -0,0 +1,41 @@
+// 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::line_item_aggregate::LineItem;
+
+#[derive(
+    Clone, Debug, Builder, Serialize, Deserialize, Getters, Eq, PartialEq, Ord, PartialOrd,
+)]
+pub struct LineItemAddedEvent {
+    added_by_user: Uuid,
+
+    line_item: LineItem,
+}
+
+#[cfg(test)]
+pub mod tests {
+    use crate::ordering::domain::add_line_item_command::AddLineItemCommand;
+
+    use super::*;
+
+    pub fn get_added_line_item_event_from_command(cmd: &AddLineItemCommand) -> LineItemAddedEvent {
+        let line_item = LineItem::get_line_item();
+
+        LineItemAddedEventBuilder::default()
+            .added_by_user(cmd.adding_by().clone())
+            .line_item(line_item)
+            .build()
+            .unwrap()
+    }
+
+    #[test]
+    fn test_event() {
+        get_added_line_item_event_from_command(&AddLineItemCommand::get_cmd());
+    }
+}

src/ordering/domain/line_item_aggregate.rs

@@ -0,0 +1,198 @@
+// 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 time::OffsetDateTime;
+use uuid::Uuid;
+
+use crate::ordering::{
+    application::services::{errors::*, *},
+    domain::{commands::*, events::*},
+};
+use crate::types::quantity::Quantity;
+
+#[derive(
+    Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Builder, Getters,
+)]
+pub struct LineItem {
+    #[builder(default = "OffsetDateTime::now_utc()")]
+    sale_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    line_item_id: Uuid,
+    quantity: Quantity,
+    #[builder(default = "false")]
+    deleted: bool,
+}
+
+impl Default for LineItem {
+    fn default() -> Self {
+        Self {
+            sale_time: OffsetDateTime::now_utc(),
+            product_name: String::default(),
+            product_id: Default::default(),
+            line_item_id: Default::default(),
+            quantity: Default::default(),
+            deleted: false,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::{
+        ordering::domain::add_line_item_command::AddLineItemCommand, utils::uuid::tests::UUID,
+    };
+
+    use super::*;
+
+    impl LineItem {
+        pub fn get_line_item() -> Self {
+            let cmd = AddLineItemCommand::get_cmd();
+
+            LineItemBuilder::default()
+                .sale_time(cmd.sale_time().clone())
+                .product_name("test_product".into())
+                .product_id(*cmd.product_id())
+                .quantity(cmd.quantity().clone())
+                .line_item_id(UUID)
+                .build()
+                .unwrap()
+        }
+    }
+}
+
+#[async_trait]
+impl Aggregate for LineItem {
+    type Command = OrderingCommand;
+    type Event = OrderingEvent;
+    type Error = OrderingError;
+    type Services = std::sync::Arc<dyn OrderingServicesInterface>;
+
+    // This identifier should be unique to the system.
+    fn aggregate_type() -> String {
+        "ordering.line_item".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 {
+            OrderingCommand::AddLineItem(cmd) => {
+                let res = services.add_line_item().add_line_item(cmd).await?;
+                Ok(vec![OrderingEvent::LineItemAdded(res)])
+            }
+            OrderingCommand::UpdateLineItem(cmd) => {
+                let res = services.update_line_item().update_line_item(cmd).await?;
+                Ok(vec![OrderingEvent::LineItemUpdated(res)])
+            }
+            OrderingCommand::DeleteLineItem(cmd) => {
+                let res = services.delete_line_item().delete_line_item(cmd).await?;
+                Ok(vec![OrderingEvent::LineItemDeleted(res)])
+            } //    _ => Ok(Vec::default()),
+        }
+    }
+
+    fn apply(&mut self, event: Self::Event) {
+        match event {
+            OrderingEvent::LineItemAdded(e) => *self = e.line_item().clone(),
+            OrderingEvent::LineItemUpdated(e) => *self = e.new_line_item().clone(),
+            OrderingEvent::LineItemDeleted(e) => *self = e.line_item().clone(),
+            //            _ => (),
+        }
+    }
+}
+
+#[cfg(test)]
+mod aggregate_tests {
+    use std::sync::Arc;
+
+    use add_line_item_service::tests::mock_add_line_item_service;
+    use cqrs_es::test::TestFramework;
+    use delete_line_item_service::tests::mock_delete_line_item_service;
+    use update_line_item_service::tests::mock_update_line_item_service;
+
+    use super::*;
+
+    use crate::ordering::domain::delete_line_item_command::DeleteLineItemCommand;
+    use crate::ordering::domain::line_item_deleted_event::tests::get_deleted_line_item_event_from_command;
+    use crate::ordering::domain::line_item_updated_event::tests::get_updated_line_item_event_from_command;
+    use crate::ordering::domain::update_line_item_command::UpdateLineItemCommand;
+    use crate::tests::bdd::*;
+
+    use crate::ordering::domain::{
+        add_line_item_command::*,
+        line_item_added_event::tests::get_added_line_item_event_from_command,
+    };
+
+    type LineItemTestFramework = TestFramework<LineItem>;
+
+    #[test]
+    fn test_add_line_item() {
+        let cmd = AddLineItemCommand::get_cmd();
+        let expected = get_added_line_item_event_from_command(&cmd);
+        let expected = OrderingEvent::LineItemAdded(expected);
+
+        let mut services = MockOrderingServicesInterface::new();
+        services
+            .expect_add_line_item()
+            .times(IS_CALLED_ONLY_ONCE.unwrap())
+            .return_const(mock_add_line_item_service(IS_CALLED_ONLY_ONCE, cmd.clone()));
+
+        LineItemTestFramework::with(Arc::new(services))
+            .given_no_previous_events()
+            .when(OrderingCommand::AddLineItem(cmd))
+            .then_expect_events(vec![expected]);
+    }
+
+    #[test]
+    fn test_update_line_item() {
+        let cmd = UpdateLineItemCommand::get_cmd();
+        let expected = get_updated_line_item_event_from_command(&cmd);
+        let expected = OrderingEvent::LineItemUpdated(expected);
+
+        let mut services = MockOrderingServicesInterface::new();
+        services
+            .expect_update_line_item()
+            .times(IS_CALLED_ONLY_ONCE.unwrap())
+            .return_const(mock_update_line_item_service(
+                IS_CALLED_ONLY_ONCE,
+                cmd.clone(),
+            ));
+
+        LineItemTestFramework::with(Arc::new(services))
+            .given_no_previous_events()
+            .when(OrderingCommand::UpdateLineItem(cmd))
+            .then_expect_events(vec![expected]);
+    }
+
+    #[test]
+    fn test_delete_line_item() {
+        let cmd = DeleteLineItemCommand::get_cmd();
+        let expected = get_deleted_line_item_event_from_command(&cmd);
+        let expected = OrderingEvent::LineItemDeleted(expected);
+
+        let mut services = MockOrderingServicesInterface::new();
+        services
+            .expect_delete_line_item()
+            .times(IS_CALLED_ONLY_ONCE.unwrap())
+            .return_const(mock_delete_line_item_service(
+                IS_CALLED_ONLY_ONCE,
+                cmd.clone(),
+            ));
+
+        LineItemTestFramework::with(Arc::new(services))
+            .given_no_previous_events()
+            .when(OrderingCommand::DeleteLineItem(cmd))
+            .then_expect_events(vec![expected]);
+    }
+}

src/ordering/domain/line_item_deleted_event.rs

@@ -0,0 +1,52 @@
+// 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::line_item_aggregate::*;
+
+#[derive(
+    Clone, Debug, Builder, Serialize, Deserialize, Getters, Eq, PartialEq, Ord, PartialOrd,
+)]
+pub struct LineItemDeletedEvent {
+    added_by_user: Uuid,
+
+    line_item: LineItem,
+}
+
+#[cfg(test)]
+pub mod tests {
+    use crate::ordering::domain::delete_line_item_command::DeleteLineItemCommand;
+
+    use super::*;
+
+    pub fn get_deleted_line_item_event_from_command(
+        cmd: &DeleteLineItemCommand,
+    ) -> LineItemDeletedEvent {
+        let deleted_line_item = LineItemBuilder::default()
+            .sale_time(cmd.line_item().sale_time().clone())
+            .product_name(cmd.line_item().product_name().into())
+            .product_id(*cmd.line_item().product_id())
+            .line_item_id(*cmd.line_item().line_item_id())
+            .quantity(cmd.line_item().quantity().clone())
+            .deleted(true)
+            .build()
+            .unwrap();
+
+        LineItemDeletedEventBuilder::default()
+            .added_by_user(cmd.adding_by().clone())
+            .line_item(deleted_line_item)
+            .build()
+            .unwrap()
+    }
+
+    #[test]
+    fn test_event() {
+        let event = get_deleted_line_item_event_from_command(&DeleteLineItemCommand::get_cmd());
+        assert!(event.line_item().deleted());
+    }
+}

src/ordering/domain/line_item_updated_event.rs

@@ -0,0 +1,52 @@
+// 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::line_item_aggregate::*;
+
+#[derive(
+    Clone, Debug, Builder, Serialize, Deserialize, Getters, Eq, PartialEq, Ord, PartialOrd,
+)]
+pub struct LineItemUpdatedEvent {
+    added_by_user: Uuid,
+
+    new_line_item: LineItem,
+    old_line_item: LineItem,
+}
+
+#[cfg(test)]
+pub mod tests {
+    use crate::ordering::domain::update_line_item_command::UpdateLineItemCommand;
+
+    use super::*;
+
+    pub fn get_updated_line_item_event_from_command(
+        cmd: &UpdateLineItemCommand,
+    ) -> LineItemUpdatedEvent {
+        let new_line_item = LineItemBuilder::default()
+            .sale_time(cmd.sale_time().clone())
+            .product_name(cmd.product_name().clone())
+            .product_id(*cmd.product_id())
+            .quantity(cmd.quantity().clone())
+            .line_item_id(*cmd.old_line_item().line_item_id())
+            .build()
+            .unwrap();
+
+        LineItemUpdatedEventBuilder::default()
+            .added_by_user(cmd.adding_by().clone())
+            .old_line_item(cmd.old_line_item().clone())
+            .new_line_item(new_line_item)
+            .build()
+            .unwrap()
+    }
+
+    #[test]
+    fn test_event() {
+        get_updated_line_item_event_from_command(&UpdateLineItemCommand::get_cmd());
+    }
+}

src/ordering/domain/mod.rs

@@ -1,3 +1,19 @@
 // SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
 //
 // SPDX-License-Identifier: AGPL-3.0-or-later
+
+// aggregates
+pub mod kot_aggregate;
+pub mod line_item_aggregate;
+
+// commands
+pub mod add_line_item_command;
+pub mod commands;
+pub mod delete_line_item_command;
+pub mod update_line_item_command;
+
+// events
+pub mod events;
+pub mod line_item_added_event;
+pub mod line_item_deleted_event;
+pub mod line_item_updated_event;

src/ordering/domain/update_line_item_command.rs

@@ -0,0 +1,166 @@
+// 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 derive_more::{Display, Error};
+use serde::{Deserialize, Serialize};
+use time::OffsetDateTime;
+use uuid::Uuid;
+
+use crate::types::quantity::*;
+use crate::utils::string::empty_string_err;
+
+use super::line_item_aggregate::LineItem;
+
+#[derive(Debug, Error, Display, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub enum UpdateLineItemCommandError {
+    QuantityIsEmpty,
+    ProductNameIsEmpty,
+}
+
+#[derive(
+    Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters, Builder,
+)]
+pub struct UnvalidatedUpdateLineItemCommand {
+    adding_by: Uuid,
+
+    #[builder(default = "OffsetDateTime::now_utc()")]
+    sale_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    quantity: Quantity,
+
+    old_line_item: LineItem,
+}
+
+impl UnvalidatedUpdateLineItemCommand {
+    pub fn validate(self) -> Result<UpdateLineItemCommand, UpdateLineItemCommandError> {
+        let product_name = empty_string_err(
+            self.product_name,
+            UpdateLineItemCommandError::ProductNameIsEmpty,
+        )?;
+
+        if self.quantity.is_empty() {
+            return Err(UpdateLineItemCommandError::QuantityIsEmpty);
+        }
+
+        Ok(UpdateLineItemCommand {
+            sale_time: self.sale_time,
+            product_name,
+            product_id: self.product_id,
+            quantity: self.quantity,
+            adding_by: self.adding_by,
+            old_line_item: self.old_line_item,
+        })
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters)]
+pub struct UpdateLineItemCommand {
+    sale_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    quantity: Quantity,
+
+    old_line_item: LineItem,
+
+    adding_by: Uuid,
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::utils::uuid::tests::UUID;
+
+    use super::*;
+
+    impl UpdateLineItemCommand {
+        pub fn get_cmd() -> Self {
+            let product_name = "foo";
+            let product_id = UUID;
+            let adding_by = UUID;
+            let quantity = Quantity::get_quantity();
+
+            UnvalidatedUpdateLineItemCommandBuilder::default()
+                .product_name(product_name.into())
+                .adding_by(adding_by)
+                .quantity(quantity.clone())
+                .product_id(product_id)
+                .old_line_item(LineItem::get_line_item())
+                .build()
+                .unwrap()
+                .validate()
+                .unwrap()
+        }
+    }
+
+    #[test]
+    fn test_cmd() {
+        let product_name = "foo";
+        let product_id = UUID;
+        let adding_by = UUID;
+        let quantity = Quantity::get_quantity();
+        let old_line_item = LineItem::get_line_item();
+
+        let cmd = UnvalidatedUpdateLineItemCommandBuilder::default()
+            .product_name(product_name.into())
+            .adding_by(adding_by)
+            .quantity(quantity.clone())
+            .product_id(product_id)
+            .old_line_item(old_line_item.clone())
+            .build()
+            .unwrap()
+            .validate()
+            .unwrap();
+
+        assert_eq!(cmd.quantity(), &quantity);
+        assert_eq!(*cmd.product_id(), product_id);
+        assert_eq!(*cmd.adding_by(), adding_by);
+        assert_eq!(cmd.product_name(), product_name);
+        assert_eq!(cmd.old_line_item(), &old_line_item);
+    }
+
+    #[test]
+    fn test_cmd_product_name_empty() {
+        let product_name = "";
+        let product_id = UUID;
+        let adding_by = UUID;
+        let quantity = Quantity::get_quantity();
+
+        assert_eq!(
+            UnvalidatedUpdateLineItemCommandBuilder::default()
+                .product_name(product_name.into())
+                .adding_by(adding_by)
+                .quantity(quantity.clone())
+                .product_id(product_id)
+                .old_line_item(LineItem::get_line_item())
+                .build()
+                .unwrap()
+                .validate(),
+            Err(UpdateLineItemCommandError::ProductNameIsEmpty)
+        );
+    }
+
+    #[test]
+    fn test_cmd_quantity_empty() {
+        let product_name = "foo";
+        let product_id = UUID;
+        let adding_by = UUID;
+        // minor = 0; major = 0;
+        let quantity = Quantity::default();
+
+        assert_eq!(
+            UnvalidatedUpdateLineItemCommandBuilder::default()
+                .product_name(product_name.into())
+                .adding_by(adding_by)
+                .quantity(quantity.clone())
+                .product_id(product_id)
+                .old_line_item(LineItem::get_line_item())
+                .build()
+                .unwrap()
+                .validate(),
+            Err(UpdateLineItemCommandError::QuantityIsEmpty)
+        );
+    }
+}

src/utils/mod.rs

@@ -4,4 +4,5 @@
 
 pub mod parse_aggregate_id;
 pub mod random_string;
+pub mod string;
 pub mod uuid;

src/utils/string.rs

@@ -0,0 +1,61 @@
+// SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+// if let s: Option<String> = Some("".into()); transform to None;
+pub fn clean_option_empty_string(s: Option<String>) -> Option<String> {
+    if let Some(s) = s {
+        let s = s.trim();
+        if s.is_empty() {
+            None
+        } else {
+            Some(s.to_owned())
+        }
+    } else {
+        None
+    }
+}
+
+pub fn empty_string_err<E: std::error::Error>(s: String, e: E) -> Result<String, E> {
+    let s = s.trim().to_owned();
+    if s.is_empty() {
+        return Err(e);
+    }
+    Ok(s)
+}
+
+#[cfg(test)]
+mod tests {
+    use derive_more::{Display, Error};
+
+    use super::*;
+
+    #[test]
+    fn test_clean_option_empty_string() {
+        assert!(clean_option_empty_string(Some("".into())).is_none());
+        assert!(clean_option_empty_string(Some("foo".into())).is_some());
+        assert_eq!(
+            clean_option_empty_string(Some("foo".into())).unwrap(),
+            "foo"
+        );
+    }
+
+    #[test]
+    fn test_empty_string_err() {
+        let s: String = "foo".into();
+
+        #[derive(Display, Debug, Eq, PartialEq, Error)]
+        enum TestError {
+            EmptyString,
+        }
+
+        assert_eq!(
+            empty_string_err(format!("  {s}  "), TestError::EmptyString).unwrap(),
+            s
+        );
+        assert_eq!(
+            empty_string_err(format!("    "), TestError::EmptyString),
+            Err(TestError::EmptyString)
+        );
+    }
+}