feat: impl View for Order aggregate

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

@@ -20,8 +20,11 @@ impl From<SqlxError> for OrderingDBError {
             if err.code() == Some(Cow::from("23505")) {
                 let msg = err.message();
 
-                if msg.contains("cqrs_ordering_store_query_product_id_key") {
+                if msg.contains("cqrs_ordering_store_query_line_item_id_key") {
                     return Self::DuplicateLineItemID;
+                } else if msg.contains("cqrs_ordering_store_query_order_id_key") {
+                    return Self::DuplicateOrderID;
+
                 } else {
                     println!("{msg}");
                 }
@@ -31,14 +34,14 @@ impl From<SqlxError> for OrderingDBError {
     }
 }
 
-///// 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()
-//    }
-//}
+/// 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 {

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

@@ -9,8 +9,10 @@ use sqlx::postgres::PgPool;
 use crate::db::{migrate::RunMigrations, sqlx_postgres::Postgres};
 
 mod errors;
-pub mod line_item_id_exists;
+mod line_item_id_exists;
+mod order_id_exists;
 mod line_item_view;
+mod order_view;
 
 #[derive(Clone)]
 pub struct OrderingDBPostgresAdapter {

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

@@ -0,0 +1,227 @@
+// 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::OrderingDBPostgresAdapter;
+use crate::ordering::domain::events::OrderingEvent;
+use crate::ordering::domain::order_aggregate::*;
+use crate::utils::parse_aggregate_id::parse_aggregate_id;
+
+pub const NEW_ORDER_NON_UUID: &str = "new_order_non_uuid-asdfa";
+
+// The view for a Order 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 OrderView {
+    customer_name: String,
+    order_id: Uuid,
+
+    created_time: OffsetDateTime,
+    deleted: bool,
+}
+
+impl Default for OrderView {
+    fn default() -> Self {
+        Self {
+            customer_name: Default::default(),
+            order_id: Default::default(),
+            created_time: OffsetDateTime::now_utc(),
+            deleted: false,
+        }
+    }
+}
+
+impl From<OrderView> for Order {
+    fn from(v: OrderView) -> Self {
+
+        OrderBuilder::default()
+            .customer_name(v.customer_name)
+            .order_id(v.order_id)
+            .created_time(v.created_time)
+            .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<Order> for OrderView {
+    fn update(&mut self, event: &EventEnvelope<Order>) {
+        match &event.payload {
+            OrderingEvent::OrderAdded(val) => {
+                self.customer_name = val.order().customer_name().into();
+                self.order_id = *val.order().order_id();
+                self.created_time = val.order().created_time().clone();
+
+                self.deleted = false;
+            }
+            _ => (),
+        }
+    }
+}
+
+#[async_trait]
+impl ViewRepository<OrderView, Order> for OrderingDBPostgresAdapter {
+    async fn load(&self, order_id: &str) -> Result<Option<OrderView>, PersistenceError> {
+        let order_id = match parse_aggregate_id(order_id, NEW_ORDER_NON_UUID)? {
+            Some((val, _)) => return Ok(Some(val)),
+            None => Uuid::parse_str(order_id).unwrap(),
+        };
+
+        let res = sqlx::query_as!(
+            OrderView,
+            "SELECT 
+                customer_name,
+                order_id,
+                created_time,
+                deleted
+             FROM
+                cqrs_ordering_order_query
+            WHERE
+                order_id = $1;",
+            order_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(PostgresAggregateError::from)?;
+
+        Ok(Some(res))
+    }
+
+    async fn load_with_context(
+        &self,
+        order_id: &str,
+    ) -> Result<Option<(OrderView, ViewContext)>, PersistenceError> {
+        let order_id = match parse_aggregate_id(order_id, NEW_ORDER_NON_UUID)? {
+            Some(val) => return Ok(Some(val)),
+            None => Uuid::parse_str(order_id).unwrap(),
+        };
+
+        let res = sqlx::query_as!(
+            OrderView,
+            "SELECT 
+                customer_name,
+                order_id,
+                created_time,
+                deleted
+             FROM
+                cqrs_ordering_order_query
+            WHERE
+                order_id = $1;",
+            order_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(PostgresAggregateError::from)?;
+
+        struct Context {
+            version: i64,
+            order_id: Uuid,
+        }
+
+        let ctx = sqlx::query_as!(
+            Context,
+            "SELECT 
+                order_id, version
+             FROM
+                cqrs_ordering_order_query
+            WHERE
+                order_id = $1;",
+            order_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(PostgresAggregateError::from)?;
+
+        let view_context = ViewContext::new(ctx.order_id.to_string(), ctx.version);
+        Ok(Some((res, view_context)))
+    }
+
+    async fn update_view(
+        &self,
+        view: OrderView,
+        context: ViewContext,
+    ) -> Result<(), PersistenceError> {
+        match context.version {
+            0 => {
+                let version = context.version + 1;
+                sqlx::query!(
+                    "INSERT INTO cqrs_ordering_order_query (
+                        version,
+                        customer_name,
+                        order_id,
+                        created_time,
+                        deleted
+
+                    ) VALUES (
+                        $1, $2, $3, $4, $5
+                    );",
+                    version,
+                    view.customer_name,
+                    view.order_id,
+                    view.created_time,
+                    view.deleted,
+                )
+                .execute(&self.pool)
+                .await
+                .map_err(PostgresAggregateError::from)?;
+            }
+
+            _ => {
+                let version = context.version + 1;
+                sqlx::query!(
+                    "UPDATE
+                        cqrs_ordering_order_query
+                    SET
+                        version = $1,
+                        customer_name = $2,
+                        order_id = $3,
+                        created_time = $4,
+                        deleted = $5;",
+                    version,
+                    view.customer_name,
+                    view.order_id,
+                    view.created_time,
+                    view.deleted,
+                )
+                .execute(&self.pool)
+                .await
+                .map_err(PostgresAggregateError::from)?;
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl Query<Order> for OrderingDBPostgresAdapter {
+    async fn dispatch(&self, order_id: &str, events: &[EventEnvelope<Order>]) {
+        let res = self
+            .load_with_context(order_id)
+            .await
+            .unwrap_or_else(|_| {
+                Some((
+                    OrderView::default(),
+                    ViewContext::new(order_id.into(), 0),
+                ))
+            });
+        let (mut view, view_context): (OrderView, ViewContext) = res.unwrap();
+        for event in events {
+            view.update(event);
+        }
+        self.update_view(view, view_context).await.unwrap();
+    }
+}