Restaurant materialized view added / query side / trigger

Author: idugalic

Cargo.toml

@@ -22,6 +22,7 @@ serde = { version = "1.0.203", features = ["derive"] }
 fmodel-rust = "0.7.0"
 serde_json = "1.0.117"
 uuid = { version = "1.8.0", features = ["serde", "v4"] }
+thiserror = "1.0.61"
 
 [dev-dependencies]
 pgrx-tests = "=0.11.4"

src/application/mod.rs

@@ -1 +1,2 @@
 pub mod order_restaurant_aggregate;
+pub mod restaurant_materialized_view;

src/application/restaurant_materialized_view.rs

@@ -0,0 +1,12 @@
+use crate::domain::api::RestaurantEvent;
+use crate::domain::restaurant_view::{RestaurantView, RestaurantViewState};
+use crate::framework::application::materialized_view::MaterializedView;
+use crate::infrastructure::restaurant_view_state_repository::RestaurantViewStateRepository;
+
+/// A convenient type alias for the restaurant materialized view.
+pub type RestaurantMeterializedView<'a> = MaterializedView<
+    Option<RestaurantViewState>,
+    RestaurantEvent,
+    RestaurantViewStateRepository,
+    RestaurantView<'a>,
+>;

src/domain/api.rs

@@ -1,5 +1,7 @@
+use crate::framework::domain::api::Identifier;
 use pgrx::{PostgresEnum, PostgresType};
 use serde::{Deserialize, Serialize};
+use std::fmt;
 use uuid::Uuid;
 
 // ########################################################
@@ -10,6 +12,12 @@ use uuid::Uuid;
 // Similarly, in Rust, the 'newtype' idiom brings compile-time guarantees that the correct value type is supplied. The 'newtype' is a struct that wraps a single value and provides a new type for that value. A 'newtype' is the same as the underlying type at runtime, so it will not introduce any performance overhead.
 #[derive(PostgresType, Serialize, Deserialize, Clone, PartialEq, Eq, Debug)]
 pub struct RestaurantId(pub Uuid);
+impl fmt::Display for RestaurantId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // Delegate the formatting to the inner Uuid
+        write!(f, "{}", self.0)
+    }
+}
 
 #[derive(PostgresType, Serialize, Deserialize, Clone, PartialEq, Eq, Debug)]
 pub struct RestaurantName(pub String);
@@ -170,6 +178,16 @@ pub enum RestaurantEvent {
     OrderPlaced(OrderPlaced),
 }
 
+impl Identifier for RestaurantEvent {
+    fn identifier(&self) -> Uuid {
+        match self {
+            RestaurantEvent::Created(e) => e.identifier.0,
+            RestaurantEvent::MenuChanged(e) => e.identifier.0,
+            RestaurantEvent::OrderPlaced(e) => e.identifier.0,
+        }
+    }
+}
+
 /// Fact/Event that a restaurant was created
 #[derive(PostgresType, Serialize, Deserialize, Debug, PartialEq, Clone, Eq)]
 pub struct RestaurantCreated {
@@ -206,6 +224,15 @@ pub enum OrderEvent {
     Prepared(OrderPrepared),
 }
 
+impl Identifier for OrderEvent {
+    fn identifier(&self) -> Uuid {
+        match self {
+            OrderEvent::Created(e) => e.identifier.0,
+            OrderEvent::Prepared(e) => e.identifier.0,
+        }
+    }
+}
+
 /// Fact/Event that an order was created
 #[derive(PostgresType, Serialize, Deserialize, Debug, PartialEq, Clone, Eq)]
 pub struct OrderCreated {

src/domain/mod.rs

@@ -200,3 +200,13 @@ pub fn sum_to_event(event: &Sum<RestaurantEvent, OrderEvent>) -> Event {
         },
     }
 }
+
+pub fn event_to_restaurant_event(event: &Event) -> Option<RestaurantEvent> {
+    match event {
+        Event::RestaurantCreated(e) => Some(RestaurantEvent::Created(e.to_owned())),
+        Event::RestaurantMenuChanged(e) => Some(RestaurantEvent::MenuChanged(e.to_owned())),
+        Event::OrderPlaced(e) => Some(RestaurantEvent::OrderPlaced(e.to_owned())),
+        Event::OrderCreated(_e) => None,
+        Event::OrderPrepared(_e) => None,
+    }
+}

src/domain/order_decider.rs

@@ -3,7 +3,7 @@ use pgrx::error;
 
 use crate::domain::api::{
     OrderCommand, OrderCreated, OrderEvent, OrderId, OrderLineItem, OrderPrepared, OrderStatus,
-    Reason, RestaurantId,
+    RestaurantId,
 };
 
 /// The state of the Order is represented by this struct. It belongs to the Domain layer.

src/domain/restaurant_view.rs

@@ -1,18 +1,19 @@
 use fmodel_rust::view::View;
+use pgrx::PostgresType;
 use serde::{Deserialize, Serialize};
 
 use crate::domain::api::{RestaurantEvent, RestaurantId, RestaurantMenu, RestaurantName};
 
 /// The state of the Restaurant View is represented by this struct. It belongs to the Domain layer.
-#[derive(Clone, PartialEq, Debug, Serialize, Deserialize)]
+#[derive(PostgresType, Clone, PartialEq, Debug, Serialize, Deserialize)]
 pub struct RestaurantViewState {
     pub identifier: RestaurantId,
     pub name: RestaurantName,
     pub menu: RestaurantMenu,
 }
 
 /// A convenient type alias for the Restaurant view
-type RestaurantView<'a> = View<'a, Option<RestaurantViewState>, RestaurantEvent>;
+pub type RestaurantView<'a> = View<'a, Option<RestaurantViewState>, RestaurantEvent>;
 
 /// View represents the event handling algorithm. It belongs to the Domain layer.
 pub fn restaurant_view<'a>() -> RestaurantView<'a> {

src/framework/application/event_sourced_aggregate.rs

@@ -71,7 +71,7 @@ where
             version = Some(ver);
             current_events.push(event);
         }
-        let new_events = self.decider.compute_new_events(&current_events, command);
+        let new_events = self.compute_new_events(&current_events, command);
         self.repository.save(&new_events, &version)
     }
 }

src/framework/application/materialized_view.rs

@@ -0,0 +1,59 @@
+use crate::framework::infrastructure::errors::ErrorMessage;
+use crate::framework::infrastructure::view_state_repository::ViewStateRepository;
+use fmodel_rust::view::ViewStateComputation;
+use std::marker::PhantomData;
+
+/// Materialized View.
+///
+/// It is using a `View` / [ViewStateComputation] to compute new state based on the current state and the event.
+/// It is using a [ViewStateRepository] to fetch the current state and to save the new state.
+///
+/// Generic parameters:
+///
+/// - `S` - State
+/// - `E` - Event
+/// - `Repository` - View State repository
+/// - `View` - View
+pub struct MaterializedView<S, E, Repository, View>
+where
+    Repository: ViewStateRepository<E, S>,
+    View: ViewStateComputation<E, S>,
+{
+    repository: Repository,
+    view: View,
+    _marker: PhantomData<(S, E)>,
+}
+
+/// Implementation of the view state computation for the materialized view.
+impl<S, E, Repository, View> ViewStateComputation<E, S> for MaterializedView<S, E, Repository, View>
+where
+    Repository: ViewStateRepository<E, S>,
+    View: ViewStateComputation<E, S>,
+{
+    /// Computes new state based on the current state and the events.
+    fn compute_new_state(&self, current_state: Option<S>, events: &[&E]) -> S {
+        self.view.compute_new_state(current_state, events)
+    }
+}
+
+/// Implementation of the `handle` method for the materialized view.
+impl<S, E, Repository, View> MaterializedView<S, E, Repository, View>
+where
+    Repository: ViewStateRepository<E, S>,
+    View: ViewStateComputation<E, S>,
+{
+    /// Creates a new instance of [MaterializedView].
+    pub fn new(repository: Repository, view: View) -> Self {
+        MaterializedView {
+            repository,
+            view,
+            _marker: PhantomData,
+        }
+    }
+    /// Handles the event by fetching the state from the repository, computing new state based on the current state and the event, and saving the new state to the repository.
+    pub fn handle(&self, event: &E) -> Result<S, ErrorMessage> {
+        let state = self.repository.fetch_state(event)?;
+        let new_state = self.compute_new_state(state, &[event]);
+        self.repository.save(&new_state)
+    }
+}

src/framework/application/mod.rs

@@ -1 +1,2 @@
 pub mod event_sourced_aggregate;
+pub mod materialized_view;

src/framework/infrastructure/errors.rs

@@ -1,6 +1,9 @@
+use pgrx::prelude::*;
+use pgrx::TryFromDatumError;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::fmt;
+use std::num::TryFromIntError;
 
 /// Error message to be returned to the client
 #[derive(Serialize, Deserialize)]
@@ -24,3 +27,17 @@ impl fmt::Debug for ErrorMessage {
 
 /// Implement Error for ErrorMessage
 impl Error for ErrorMessage {}
+
+#[derive(thiserror::Error, Debug)]
+pub enum TriggerError {
+    #[error("Null Trigger Tuple found")]
+    NullTriggerTuple,
+    #[error("PgHeapTuple error: {0}")]
+    PgHeapTuple(#[from] PgHeapTupleError),
+    #[error("TryFromDatumError error: {0}")]
+    TryFromDatum(#[from] TryFromDatumError),
+    #[error("TryFromInt error: {0}")]
+    TryFromInt(#[from] TryFromIntError),
+    #[error("Event Handling Error: {0}")]
+    EventHandlingError(String),
+}

src/framework/infrastructure/event_repository.rs

@@ -1,19 +1,14 @@
 use crate::framework::domain::api::{DeciderType, EventType, Identifier, IsFinal};
 use crate::framework::infrastructure::errors::ErrorMessage;
+use crate::framework::infrastructure::to_payload;
 use pgrx::{IntoDatum, JsonB, PgBuiltInOids, Spi, Uuid};
 use serde::de::DeserializeOwned;
 use serde::Serialize;
 use std::fmt::Debug;
 use uuid::Uuid as UUID;
 
-/// Converts a `JsonB` to an event payload type.
-fn to_event<E: DeserializeOwned>(jsonb: JsonB) -> Result<E, ErrorMessage> {
-    let value = jsonb.0.clone();
-    serde_json::from_value(value).map_err(|err| ErrorMessage {
-        message: "Failed to deserialize event: ".to_string() + &err.to_string(),
-    })
-}
-/// A trait for event repositories.
+/// A trait for event repositories / the command side of the CQRS pattern.
+/// Default implementation includes fetching and saving events.
 pub trait EventRepository<C, E>
 where
     C: Identifier,
@@ -55,7 +50,7 @@ where
                                 .to_string(),
                     })?;
 
-                results.push((to_event(data)?, UUID::from_bytes(*event_id.as_bytes())));
+                results.push((to_payload(data)?, UUID::from_bytes(*event_id.as_bytes())));
             }
             Ok(results)
         })
@@ -139,7 +134,7 @@ where
                                     .to_string(),
                         })?;
 
-                    results.push((to_event(data)?, UUID::from_bytes(*event_id.as_bytes())));
+                    results.push((to_payload(data)?, UUID::from_bytes(*event_id.as_bytes())));
                 }
                 version = Some(event_id);
             }
@@ -149,6 +144,7 @@ where
 }
 
 /// A trait for event orchestrating repositories.
+/// Default implementation includes fetching events, fetching latest version and saving events.
 pub trait EventOrchestratingRepository<C, E>
 where
     C: Identifier,
@@ -196,7 +192,7 @@ where
                             "Failed to fetch event id (map `data` to `JsonB`): No event id found"
                                 .to_string(),
                     })?;
-                results.push((to_event(data)?, UUID::from_bytes(*event_id.as_bytes())));
+                results.push((to_payload(data)?, UUID::from_bytes(*event_id.as_bytes())));
             }
             Ok(results)
         })
@@ -313,7 +309,7 @@ where
                                 "Failed to save event id (map `data` to `JsonB`): No event id found"
                                     .to_string(),
                         })?;
-                    results.push((to_event(data)?, UUID::from_bytes(*event_id.as_bytes())));
+                    results.push((to_payload(data)?, UUID::from_bytes(*event_id.as_bytes())));
                 }
             }
             Ok(results)

src/framework/infrastructure/mod.rs

@@ -1,2 +1,15 @@
+use crate::framework::infrastructure::errors::ErrorMessage;
+use pgrx::JsonB;
+use serde::de::DeserializeOwned;
+
 pub mod errors;
 pub mod event_repository;
+pub mod view_state_repository;
+
+/// Converts a `JsonB` to the payload type.
+pub fn to_payload<E: DeserializeOwned>(jsonb: JsonB) -> Result<E, ErrorMessage> {
+    let value = jsonb.0.clone();
+    serde_json::from_value(value).map_err(|err| ErrorMessage {
+        message: "Failed to deserialize payload: ".to_string() + &err.to_string(),
+    })
+}

src/framework/infrastructure/view_state_repository.rs

@@ -0,0 +1,9 @@
+use crate::framework::infrastructure::errors::ErrorMessage;
+
+/// A trait for a view state repository / the query side of the CQRS pattern.
+pub trait ViewStateRepository<E, S> {
+    /// Fetches current state, based on the event.
+    fn fetch_state(&self, event: &E) -> Result<Option<S>, ErrorMessage>;
+    /// Saves the new state.
+    fn save(&self, state: &S) -> Result<S, ErrorMessage>;
+}

src/infrastructure/mod.rs

@@ -1 +1,2 @@
 pub mod order_restaurant_event_repository;
+pub mod restaurant_view_state_repository;