.sqlx/query-0268f0c43abe34a3147f0a43f0e11bf302a3750ecadf28b1d6af62da62934f87.json

@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE\n                        cqrs_billing_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                        created_time = $9,\n                        bill_id = $10,\n                        price_per_unit_minor = $11 ,\n                        price_per_unit_major = $12,\n                        price_per_unit_currency = $13,\n                            deleted = $14;",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Text",
+        "Uuid",
+        "Uuid",
+        "Text",
+        "Int4",
+        "Text",
+        "Int4",
+        "Timestamptz",
+        "Uuid",
+        "Int4",
+        "Int4",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0268f0c43abe34a3147f0a43f0e11bf302a3750ecadf28b1d6af62da62934f87"
+}

.sqlx/query-0d6ba3039f3419dd34a609ae207ac7ca9cf88099aa273972271040852794caa5.json

@@ -0,0 +1,94 @@
+{
+  "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                created_time,\n                bill_id,\n                price_per_unit_minor,\n                price_per_unit_major,\n                price_per_unit_currency,\n                deleted\n             FROM\n                cqrs_billing_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": "created_time",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 8,
+        "name": "bill_id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 9,
+        "name": "price_per_unit_minor",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 10,
+        "name": "price_per_unit_major",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 11,
+        "name": "price_per_unit_currency",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 12,
+        "name": "deleted",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "0d6ba3039f3419dd34a609ae207ac7ca9cf88099aa273972271040852794caa5"
+}

.sqlx/query-6ac0c3c64749d1eb4a93f8727b56d6a2c9e9b1e49258b38ed1dbebf2f601e99e.json

@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO cqrs_billing_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                        created_time,\n                        bill_id,\n                        price_per_unit_minor,\n                        price_per_unit_major,\n                        price_per_unit_currency,\n                        deleted\n                    ) VALUES (\n                        $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14\n                    );",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Text",
+        "Uuid",
+        "Uuid",
+        "Text",
+        "Int4",
+        "Text",
+        "Int4",
+        "Timestamptz",
+        "Uuid",
+        "Int4",
+        "Int4",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6ac0c3c64749d1eb4a93f8727b56d6a2c9e9b1e49258b38ed1dbebf2f601e99e"
+}

.sqlx/query-a80999e0bed9e7535187b20eb501258adf1c319543da07a8d79de67394b83eeb.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT \n                line_item_id, version\n             FROM\n                cqrs_billing_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": "a80999e0bed9e7535187b20eb501258adf1c319543da07a8d79de67394b83eeb"
+}

.sqlx/query-c1308bfe8d0d602e154a7d02df7e229e775326bf554e935b688585d0be253934.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS (\n                SELECT 1\n                    FROM cqrs_billing_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": "c1308bfe8d0d602e154a7d02df7e229e775326bf554e935b688585d0be253934"
+}

migrations/20240917082053_cqrs_billing_line_item_query.sql

@@ -0,0 +1,29 @@
+-- SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
+--
+-- SPDX-License-Identifier: AGPL-3.0-or-later
+
+CREATE TABLE IF NOT EXISTS cqrs_billing_line_item_query
+(
+    version           bigint CHECK (version >= 0) NOT NULL,
+
+    created_time timestamp with time zone DEFAULT (CURRENT_TIMESTAMP) NOT NULL,
+	line_item_id UUID NOT NULL UNIQUE,
+
+	bill_id UUID NOT NULL,
+
+    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,
+
+    price_per_unit_minor INTEGER NOT NULL,
+    price_per_unit_major INTEGER NOT NULL,
+    price_per_unit_currency TEXT NOT NULL,
+
+	deleted BOOLEAN NOT NULL DEFAULT FALSE,
+
+    PRIMARY KEY (line_item_id)
+);

src/billing/adapters/output/db/postgres/line_item_id_exists.rs

@@ -0,0 +1,109 @@
+// 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::*, line_item_id_exists::*};
+
+#[async_trait::async_trait]
+impl LineItemIDExistsDBPort for BillingDBPostgresAdapter {
+    async fn line_item_id_exists(&self, line_item_id: &Uuid) -> BillingDBResult<bool> {
+        let res = sqlx::query!(
+            "SELECT EXISTS (
+                SELECT 1
+                    FROM cqrs_billing_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::billing::domain::add_product_command::tests::get_customizations;
+    use crate::billing::domain::line_item_aggregate::*;
+
+    async fn create_dummy_line_item(line_item: &LineItem, db: &BillingDBPostgresAdapter) {
+        sqlx::query!(
+            "INSERT INTO cqrs_billing_line_item_query (
+                    version,
+                    product_name,
+                    product_id,
+                    line_item_id,
+                    quantity_minor_unit,
+                    quantity_minor_number,
+                    quantity_major_unit,
+                    quantity_major_number,
+
+                    bill_id,
+
+                price_per_unit_minor,
+                price_per_unit_major,
+                price_per_unit_currency,
+
+
+                    deleted
+                ) VALUES (
+                    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13
+                );",
+            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.bill_id(),
+            *line_item.price_per_unit().minor() as i32,
+            *line_item.price_per_unit().major() as i32,
+            line_item.price_per_unit().currency().to_string(),
+            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::BillingDBPostgresAdapter::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/billing/adapters/output/db/postgres/line_item_view.rs

@@ -0,0 +1,347 @@
+// 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 time::OffsetDateTime;
+use uuid::Uuid;
+
+use super::errors::*;
+use super::BillingDBPostgresAdapter;
+use crate::billing::domain::events::BillingEvent;
+use crate::billing::domain::line_item_aggregate::*;
+use crate::types::currency::*;
+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-billing";
+
+// 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, Serialize, Deserialize)]
+pub struct LineItemView {
+    product_name: String,
+    product_id: Uuid,
+    bill_id: Uuid,
+    created_time: OffsetDateTime,
+
+    line_item_id: Uuid,
+
+    quantity_major_number: i32,
+    quantity_minor_number: i32,
+    quantity_major_unit: String,
+    quantity_minor_unit: String,
+
+    price_per_unit_major: i32,
+    price_per_unit_minor: i32,
+    price_per_unit_currency: String,
+
+    deleted: bool,
+}
+
+impl Default for LineItemView {
+    fn default() -> Self {
+        Self {
+            created_time: OffsetDateTime::now_utc(),
+            product_name: Default::default(),
+            product_id: Default::default(),
+            bill_id: Default::default(),
+
+            line_item_id: Default::default(),
+
+            price_per_unit_minor: Default::default(),
+            price_per_unit_major: Default::default(),
+            price_per_unit_currency: Default::default(),
+
+            quantity_major_number: Default::default(),
+            quantity_minor_number: Default::default(),
+            quantity_major_unit: Default::default(),
+            quantity_minor_unit: Default::default(),
+            deleted: false,
+        }
+    }
+}
+
+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();
+
+        let price_per_unit = PriceBuilder::default()
+            .minor(v.price_per_unit_minor as usize)
+            .major(v.price_per_unit_major as usize)
+            .currency(Currency::from_str(&v.price_per_unit_currency).unwrap())
+            .build()
+            .unwrap();
+
+        LineItemBuilder::default()
+            .product_name(v.product_name)
+            .line_item_id(v.line_item_id)
+            .quantity(quantity)
+            .created_time(v.created_time)
+            .product_id(v.product_id)
+            .price_per_unit(price_per_unit)
+            .bill_id(v.bill_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 {
+            BillingEvent::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.price_per_unit_major = *val.line_item().price_per_unit().major() as i32;
+                self.price_per_unit_minor = *val.line_item().price_per_unit().minor() as i32;
+                self.price_per_unit_currency =
+                    val.line_item().price_per_unit().currency().to_string();
+
+                self.created_time = val.line_item().created_time().clone();
+                self.bill_id = *val.line_item().bill_id();
+
+                self.deleted = false;
+            }
+            _ => (),
+        }
+    }
+}
+
+#[async_trait]
+impl ViewRepository<LineItemView, LineItem> for BillingDBPostgresAdapter {
+    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,
+                created_time,
+                bill_id,
+                price_per_unit_minor,
+                price_per_unit_major,
+                price_per_unit_currency,
+                deleted
+             FROM
+                cqrs_billing_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,
+                created_time,
+                bill_id,
+                price_per_unit_minor,
+                price_per_unit_major,
+                price_per_unit_currency,
+                deleted
+             FROM
+                cqrs_billing_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_billing_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_billing_line_item_query (
+                        version,
+                        product_name,
+                        product_id,
+                        line_item_id,
+                        quantity_minor_unit,
+                        quantity_minor_number,
+                        quantity_major_unit,
+                        quantity_major_number,
+                        created_time,
+                        bill_id,
+                        price_per_unit_minor,
+                        price_per_unit_major,
+                        price_per_unit_currency,
+                        deleted
+                    ) VALUES (
+                        $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14
+                    );",
+                    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.created_time,
+                    view.bill_id,
+                    view.price_per_unit_minor,
+                    view.price_per_unit_major,
+                    view.price_per_unit_currency,
+                    view.deleted,
+                )
+                .execute(&self.pool)
+                .await
+                .map_err(PostgresAggregateError::from)?;
+            }
+
+            _ => {
+                let version = context.version + 1;
+                sqlx::query!(
+                    "UPDATE
+                        cqrs_billing_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,
+                        created_time = $9,
+                        bill_id = $10,
+                        price_per_unit_minor = $11 ,
+                        price_per_unit_major = $12,
+                        price_per_unit_currency = $13,
+                            deleted = $14;",
+                    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.created_time,
+                    view.bill_id,
+                    view.price_per_unit_minor,
+                    view.price_per_unit_major,
+                    view.price_per_unit_currency,
+                    view.deleted,
+                )
+                .execute(&self.pool)
+                .await
+                .map_err(PostgresAggregateError::from)?;
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl Query<LineItem> for BillingDBPostgresAdapter {
+    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/billing/adapters/output/db/postgres/mod.rs

@@ -8,6 +8,8 @@ use sqlx::postgres::PgPool;
 use crate::db::{migrate::RunMigrations, sqlx_postgres::Postgres};
 
 mod errors;
+mod line_item_id_exists;
+mod line_item_view;
 mod store_id_exists;
 mod store_name_exists;
 mod store_view;

src/billing/application/services/add_line_item_service.rs

@@ -0,0 +1,141 @@
+// 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::billing::{
+    application::port::output::db::bill_id_exists::*,
+    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) -> BillingResult<LineItemAddedEvent>;
+}
+
+pub type AddLineItemServiceObj = Arc<dyn AddLineItemUseCase>;
+
+#[derive(Clone, Builder)]
+pub struct AddLineItemService {
+    db_line_item_id_exists: LineItemIDExistsDBPortObj,
+    db_bill_id_exists: BillIDExistsDBPortObj,
+    get_uuid: GetUUIDInterfaceObj,
+}
+
+#[async_trait::async_trait]
+impl AddLineItemUseCase for AddLineItemService {
+    async fn add_line_item(&self, cmd: AddLineItemCommand) -> BillingResult<LineItemAddedEvent> {
+        if !self.db_bill_id_exists.bill_id_exists(cmd.bill_id()).await? {
+            return Err(BillingError::BillIDNotFound);
+        }
+
+        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()
+            .created_time(cmd.created_time().clone())
+            .product_name(cmd.product_name().into())
+            .product_id(*cmd.product_id())
+            .bill_id(*cmd.bill_id())
+            .line_item_id(line_item_id)
+            .quantity(cmd.quantity().clone())
+            .price_per_unit(cmd.price_per_unit().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::billing::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))
+            .db_bill_id_exists(mock_bill_id_exists_db_port_true(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().bill_id(), cmd.bill_id());
+        assert_eq!(res.line_item().quantity(), cmd.quantity());
+        assert_eq!(res.line_item().quantity(), cmd.quantity());
+        assert_eq!(res.line_item().created_time(), cmd.created_time());
+        assert!(!res.line_item().deleted());
+        assert_eq!(res.added_by_user(), cmd.adding_by());
+    }
+
+    #[actix_rt::test]
+    async fn test_service_bill_id_doesnt_exist() {
+        let cmd = AddLineItemCommand::get_cmd();
+
+        let s = AddLineItemServiceBuilder::default()
+            .db_line_item_id_exists(mock_line_item_id_exists_db_port_false(IS_NEVER_CALLED))
+            .db_bill_id_exists(mock_bill_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
+            .get_uuid(mock_get_uuid(IS_NEVER_CALLED))
+            .build()
+            .unwrap();
+
+        assert_eq!(
+            s.add_line_item(cmd.clone()).await,
+            Err(BillingError::BillIDNotFound)
+        );
+    }
+}

src/billing/application/services/delete_line_item_service.rs

@@ -0,0 +1,135 @@
+// 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::billing::{
+    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,
+    ) -> BillingResult<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,
+    ) -> BillingResult<LineItemDeletedEvent> {
+        if !self
+            .db_line_item_id_exists
+            .line_item_id_exists(cmd.line_item().line_item_id())
+            .await?
+        {
+            return Err(BillingError::LineItemIDNotFound);
+        }
+
+        let deleted_line_item = LineItemBuilder::default()
+            .created_time(cmd.line_item().created_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())
+            .bill_id(*cmd.line_item().bill_id())
+            .quantity(cmd.line_item().quantity().clone())
+            .price_per_unit(cmd.line_item().price_per_unit().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::billing::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_eq!(res.line_item().bill_id(), cmd.line_item().bill_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(BillingError::LineItemIDNotFound)
+        );
+    }
+}

src/billing/application/services/errors.rs

@@ -0,0 +1,43 @@
+// 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::billing::application::port::output::db::errors::BillingDBError;
+
+pub type BillingResult<V> = Result<V, BillingError>;
+
+#[derive(Debug, Error, Display, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub enum BillingError {
+    BillIDNotFound,
+    InternalError,
+    DuplicateStoreName,
+    StoreIDNotFound,
+    LineItemIDNotFound,
+}
+
+impl From<BillingDBError> for BillingError {
+    fn from(value: BillingDBError) -> Self {
+        match value {
+            BillingDBError::DuplicateBillID => {
+                error!("DuplicateBillID");
+                Self::InternalError
+            }
+            BillingDBError::DuplicateStoreName => Self::DuplicateStoreName,
+            BillingDBError::DuplicateStoreID => {
+                error!("DuplicateStoreID");
+                Self::InternalError
+            }
+            BillingDBError::StoreIDNotFound => BillingError::StoreIDNotFound,
+            BillingDBError::InternalError => BillingError::InternalError,
+            BillingDBError::DuplicateLineItemID => {
+                error!("DuplicateLineItemID");
+                Self::InternalError
+            }
+            BillingDBError::LineItemIDNotFound => BillingError::LineItemIDNotFound,
+        }
+    }
+}

src/billing/application/services/update_line_item_service.rs

@@ -0,0 +1,161 @@
+// 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::billing::{
+    application::port::output::db::bill_id_exists::*,
+    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,
+    ) -> BillingResult<LineItemUpdatedEvent>;
+}
+
+pub type UpdateLineItemServiceObj = Arc<dyn UpdateLineItemUseCase>;
+
+#[derive(Clone, Builder)]
+pub struct UpdateLineItemService {
+    db_line_item_id_exists: LineItemIDExistsDBPortObj,
+    db_bill_id_exists: BillIDExistsDBPortObj,
+}
+
+#[async_trait::async_trait]
+impl UpdateLineItemUseCase for UpdateLineItemService {
+    async fn update_line_item(
+        &self,
+        cmd: UpdateLineItemCommand,
+    ) -> BillingResult<LineItemUpdatedEvent> {
+        if !self
+            .db_line_item_id_exists
+            .line_item_id_exists(cmd.old_line_item().line_item_id())
+            .await?
+        {
+            return Err(BillingError::LineItemIDNotFound);
+        }
+
+        if !self.db_bill_id_exists.bill_id_exists(cmd.bill_id()).await? {
+            return Err(BillingError::BillIDNotFound);
+        }
+
+        let new_line_item = LineItemBuilder::default()
+            .created_time(cmd.created_time().clone())
+            .product_name(cmd.product_name().into())
+            .product_id(*cmd.product_id())
+            .line_item_id(*cmd.old_line_item().line_item_id())
+            .bill_id(*cmd.bill_id())
+            .quantity(cmd.quantity().clone())
+            .price_per_unit(cmd.price_per_unit().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::billing::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))
+            .db_bill_id_exists(mock_bill_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().line_item_id(),
+            cmd.old_line_item().line_item_id()
+        );
+        assert_eq!(res.new_line_item().bill_id(), cmd.old_line_item().bill_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))
+            .db_bill_id_exists(mock_bill_id_exists_db_port_true(IS_NEVER_CALLED))
+            .build()
+            .unwrap();
+
+        assert_eq!(
+            s.update_line_item(cmd.clone()).await,
+            Err(BillingError::LineItemIDNotFound)
+        );
+    }
+
+    #[actix_rt::test]
+    async fn test_service_bill_id_doesnt_exist() {
+        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))
+            .db_bill_id_exists(mock_bill_id_exists_db_port_false(IS_CALLED_ONLY_ONCE))
+            .build()
+            .unwrap();
+
+        assert_eq!(
+            s.update_line_item(cmd.clone()).await,
+            Err(BillingError::BillIDNotFound)
+        );
+    }
+}

src/billing/domain/add_line_item_command.rs

@@ -0,0 +1,174 @@
+// 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::{currency::*, 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()")]
+    created_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    bill_id: Uuid,
+    quantity: Quantity,
+    price_per_unit: Price,
+}
+
+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 {
+            created_time: self.created_time,
+            product_name,
+            product_id: self.product_id,
+            bill_id: self.bill_id,
+            quantity: self.quantity,
+            adding_by: self.adding_by,
+            price_per_unit: self.price_per_unit,
+        })
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters)]
+pub struct AddLineItemCommand {
+    created_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    bill_id: Uuid,
+    quantity: Quantity,
+    price_per_unit: Price,
+
+    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 bill_id = UUID;
+            let adding_by = UUID;
+            let quantity = Quantity::get_quantity();
+
+            UnvalidatedAddLineItemCommandBuilder::default()
+                .product_name(product_name.into())
+                .adding_by(adding_by)
+                .created_time(datetime!(1970-01-01 0:00 UTC))
+                .quantity(quantity.clone())
+                .price_per_unit(Price::default())
+                .product_id(product_id)
+                .bill_id(bill_id)
+                .build()
+                .unwrap()
+                .validate()
+                .unwrap()
+        }
+    }
+
+    #[test]
+    fn test_cmd() {
+        let product_name = "foo";
+        let product_id = UUID;
+        let bill_id = UUID;
+        let adding_by = UUID;
+        let quantity = Quantity::get_quantity();
+
+        let cmd = UnvalidatedAddLineItemCommandBuilder::default()
+            .product_name(product_name.into())
+            .adding_by(adding_by)
+            .price_per_unit(Price::default())
+            .quantity(quantity.clone())
+            .product_id(product_id)
+            .bill_id(bill_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 bill_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())
+                .price_per_unit(Price::default())
+                .product_id(product_id)
+                .bill_id(bill_id)
+                .build()
+                .unwrap()
+                .validate(),
+            Err(AddLineItemCommandError::ProductNameIsEmpty)
+        );
+    }
+
+    #[test]
+    fn test_cmd_quantity_empty() {
+        let product_name = "foo";
+        let product_id = UUID;
+        let bill_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)
+                .bill_id(bill_id)
+                .price_per_unit(Price::default())
+                .build()
+                .unwrap()
+                .validate(),
+            Err(AddLineItemCommandError::QuantityIsEmpty)
+        );
+    }
+}

src/billing/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/billing/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::billing::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/billing/domain/line_item_aggregate.rs

@@ -0,0 +1,219 @@
+// 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::billing::{
+    application::services::{errors::*, *},
+    domain::{commands::*, events::*},
+};
+use crate::types::{currency::Price, quantity::Quantity};
+
+#[derive(
+    Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Builder, Getters,
+)]
+pub struct LineItem {
+    #[builder(default = "OffsetDateTime::now_utc()")]
+    created_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    bill_id: Uuid,
+    line_item_id: Uuid,
+    quantity: Quantity,
+    price_per_unit: Price,
+    #[builder(default = "false")]
+    deleted: bool,
+}
+
+impl Default for LineItem {
+    fn default() -> Self {
+        Self {
+            created_time: OffsetDateTime::now_utc(),
+            product_name: String::default(),
+            product_id: Default::default(),
+            line_item_id: Default::default(),
+            bill_id: Default::default(),
+            quantity: Default::default(),
+            price_per_unit: Price::default(),
+            deleted: false,
+        }
+    }
+}
+
+impl LineItem {
+    pub fn total_price(&self) -> Price {
+        let total_price_as_minor = (self.quantity().major_as_minor().unwrap() // TODO: handle err
+             + self.quantity().minor().number())
+            * (self.price_per_unit().major_as_minor() + self.price_per_unit().minor());
+
+        Price::from_minor(
+            total_price_as_minor,
+            self.price_per_unit().currency().clone(),
+        )
+    }
+}
+
+#[cfg(test)]
+pub mod tests {
+    use crate::{
+        billing::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()
+                .created_time(cmd.created_time().clone())
+                .product_name("test_product".into())
+                .product_id(*cmd.product_id())
+                .quantity(cmd.quantity().clone())
+                .bill_id(*cmd.bill_id())
+                .price_per_unit(cmd.price_per_unit().clone())
+                .line_item_id(UUID)
+                .build()
+                .unwrap()
+        }
+    }
+}
+
+#[async_trait]
+impl Aggregate for LineItem {
+    type Command = BillingCommand;
+    type Event = BillingEvent;
+    type Error = BillingError;
+    type Services = std::sync::Arc<dyn BillingServicesInterface>;
+
+    // This identifier should be unique to the system.
+    fn aggregate_type() -> String {
+        "billing.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 {
+            BillingCommand::AddLineItem(cmd) => {
+                let res = services.add_line_item().add_line_item(cmd).await?;
+                Ok(vec![BillingEvent::LineItemAdded(res)])
+            }
+            BillingCommand::UpdateLineItem(cmd) => {
+                let res = services.update_line_item().update_line_item(cmd).await?;
+                Ok(vec![BillingEvent::LineItemUpdated(res)])
+            }
+            BillingCommand::DeleteLineItem(cmd) => {
+                let res = services.delete_line_item().delete_line_item(cmd).await?;
+                Ok(vec![BillingEvent::LineItemDeleted(res)])
+            }
+            _ => Ok(Vec::default()),
+        }
+    }
+
+    fn apply(&mut self, event: Self::Event) {
+        match event {
+            BillingEvent::LineItemAdded(e) => *self = e.line_item().clone(),
+            BillingEvent::LineItemUpdated(e) => *self = e.new_line_item().clone(),
+            BillingEvent::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::billing::application::services::*;
+    use crate::billing::domain::delete_line_item_command::DeleteLineItemCommand;
+    use crate::billing::domain::line_item_deleted_event::tests::get_deleted_line_item_event_from_command;
+    use crate::billing::domain::line_item_updated_event::tests::get_updated_line_item_event_from_command;
+    use crate::billing::domain::update_line_item_command::UpdateLineItemCommand;
+    use crate::tests::bdd::*;
+
+    use crate::billing::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 = BillingEvent::LineItemAdded(expected);
+
+        let mut services = MockBillingServicesInterface::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(BillingCommand::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 = BillingEvent::LineItemUpdated(expected);
+
+        let mut services = MockBillingServicesInterface::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(BillingCommand::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 = BillingEvent::LineItemDeleted(expected);
+
+        let mut services = MockBillingServicesInterface::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(BillingCommand::DeleteLineItem(cmd))
+            .then_expect_events(vec![expected]);
+    }
+}

src/billing/domain/line_item_deleted_event.rs

@@ -0,0 +1,54 @@
+// 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::billing::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()
+            .created_time(cmd.line_item().created_time().clone())
+            .product_name(cmd.line_item().product_name().into())
+            .product_id(*cmd.line_item().product_id())
+            .bill_id(*cmd.line_item().bill_id())
+            .line_item_id(*cmd.line_item().line_item_id())
+            .price_per_unit(cmd.line_item().price_per_unit().clone())
+            .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/billing/domain/line_item_updated_event.rs

@@ -0,0 +1,54 @@
+// 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::billing::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()
+            .created_time(cmd.created_time().clone())
+            .product_name(cmd.product_name().clone())
+            .product_id(*cmd.product_id())
+            .bill_id(*cmd.bill_id())
+            .price_per_unit(cmd.price_per_unit().clone())
+            .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/billing/domain/update_line_item_command.rs

@@ -0,0 +1,185 @@
+// 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::{currency::*, 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()")]
+    created_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    bill_id: Uuid,
+    quantity: Quantity,
+    price_per_unit: Price,
+
+    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 {
+            created_time: self.created_time,
+            product_name,
+            product_id: self.product_id,
+            bill_id: self.bill_id,
+            quantity: self.quantity,
+            adding_by: self.adding_by,
+            old_line_item: self.old_line_item,
+            price_per_unit: self.price_per_unit,
+        })
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Getters)]
+pub struct UpdateLineItemCommand {
+    created_time: OffsetDateTime,
+    product_name: String,
+    product_id: Uuid,
+    bill_id: Uuid,
+    quantity: Quantity,
+    price_per_unit: Price,
+
+    old_line_item: LineItem,
+
+    adding_by: Uuid,
+}
+//
+#[cfg(test)]
+mod tests {
+    use crate::{billing::domain::bill_aggregate::Bill, utils::uuid::tests::UUID};
+
+    use super::*;
+
+    impl UpdateLineItemCommand {
+        pub fn get_cmd() -> Self {
+            let product_name = "foo";
+            let product_id = UUID;
+            let bill_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)
+                .bill_id(product_id)
+                .old_line_item(LineItem::get_line_item())
+                .price_per_unit(Price::default())
+                .build()
+                .unwrap()
+                .validate()
+                .unwrap()
+        }
+    }
+
+    #[test]
+    fn test_cmd() {
+        let product_name = "foo";
+        let product_id = UUID;
+        let bill_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)
+            .bill_id(bill_id)
+            .old_line_item(old_line_item.clone())
+            .price_per_unit(Price::default())
+            .build()
+            .unwrap()
+            .validate()
+            .unwrap();
+
+        assert_eq!(cmd.quantity(), &quantity);
+        assert_eq!(*cmd.product_id(), product_id);
+        assert_eq!(*cmd.bill_id(), bill_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 bill_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)
+                .bill_id(bill_id)
+                .price_per_unit(Price::default())
+                .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 bill_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)
+                .bill_id(bill_id)
+                .old_line_item(LineItem::get_line_item())
+                .price_per_unit(Price::default())
+                .build()
+                .unwrap()
+                .validate(),
+            Err(UpdateLineItemCommandError::QuantityIsEmpty)
+        );
+    }
+}