Draft: feat: make average-rating optional, bump versions, and add basic OTLP metrics

Cargo.toml

@@ -1,22 +1,27 @@
 [package]
 name = "misarch-review"
-version = "0.1.0"
-edition = "2021"
+version = "0.3.0"
+edition = "2024"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-graphql = { version = "6.0.11", features = ["bson", "chrono", "uuid", "log"] }
-async-graphql-axum = "6.0.11"
-tokio = { version = "1.8", features = ["macros", "rt-multi-thread"] }
-axum = { version = "0.6.0", features = ["headers", "macros"] }
-mongodb = "2.8.0"
-serde = "1.0.193"
-bson = "2.8.1"
-clap = { version = "4.4.13", features = ["derive"] }
-uuid = { version = "1.6.1", features = ["v4", "serde"] }
+async-graphql = { version = "7.0.16", features = ["bson", "chrono", "uuid", "log"] }
+async-graphql-axum = "7.0.16"
+tokio = { version = "1.44.2", features = ["macros", "rt-multi-thread"] }
+axum = { version = "0.8.3", features = ["macros"] }
+mongodb = "2.8.2"
+serde = "1.0.219"
+bson = "2.14.0"
+clap = { version = "4.5.37", features = ["derive"] }
+uuid = { version = "1.16.0", features = ["v4", "serde"] }
 mongodb-cursor-pagination = "0.3.2"
 json = "0.12.4"
-log = "0.4.20"
-simple_logger = "4.3.3"
-serde_json = "1.0.113"
+log = "0.4.27"
+simple_logger = "5.0.0"
+serde_json = "1.0.140"
+opentelemetry = "0.30.0"
+opentelemetry_sdk = { version = "0.30.0", features = ["rt-tokio"]}
+opentelemetry-otlp = "0.30.0"
+axum-otel-metrics = { version = "0.12.0" }
+once_cell = "1.21.3"

src/graphql/model/product.rs

@@ -61,9 +61,11 @@ impl Product {
     }
 
     /// Retrieves average rating of product.
-    async fn average_rating<'a>(&self, ctx: &Context<'a>) -> Result<f32> {
-        let review_connection = self.reviews(&ctx, None, None, None).await?;
-        calculate_average_rating(review_connection).await
+    async fn average_rating<'a>(&self, ctx: &Context<'a>) -> Option<f32> {
+        match self.reviews(ctx, None, None, None).await {
+            Ok(review_connection) => calculate_average_rating(review_connection).await,
+            Err(_) => None,
+        }
     }
 }
 

src/graphql/model/product_variant.rs

@@ -28,6 +28,7 @@ pub struct ProductVariant {
 #[ComplexObject]
 impl ProductVariant {
     /// Retrieves reviews of product variant.
+    // TODO reviews should be optional
     async fn reviews<'a>(
         &self,
         ctx: &Context<'a>,
@@ -65,9 +66,11 @@ impl ProductVariant {
     }
 
     /// Retrieves average rating of product variant.
-    async fn average_rating<'a>(&self, ctx: &Context<'a>) -> Result<f32> {
-        let review_connection = self.reviews(&ctx, None, None, None).await?;
-        calculate_average_rating(review_connection).await
+    async fn average_rating<'a>(&self, ctx: &Context<'a>) -> Option<f32> {
+        match self.reviews(ctx, None, None, None).await {
+            Ok(review_connection) => calculate_average_rating(review_connection).await,
+            Err(_) => None,
+        }
     }
 }
 
@@ -91,7 +94,7 @@ impl From<ProductVariantEventData> for ProductVariant {
 /// Filters reviews with `is_visible == false` to exclude them from the average rating.
 ///
 /// `review_connection` - Connection of reviews to calculate average rating for.
-pub async fn calculate_average_rating<'a>(review_connection: ReviewConnection) -> Result<f32> {
+pub async fn calculate_average_rating<'a>(review_connection: ReviewConnection) -> Option<f32> {
     let reviews = review_connection.nodes.clone();
     let (accumulated_reviews, total_count) =
         reviews.iter().filter(|review| review.is_visible).fold(
@@ -104,13 +107,9 @@ pub async fn calculate_average_rating<'a>(review_connection: ReviewConnection) -
             },
         );
     if total_count == 0 {
-        let message = format!(
-            "Average rating can not be calculated, no review exists in review connection:`{:?}`",
-            review_connection
-        );
-        Err(Error::new(message))
+        None
     } else {
         let average_rating = accumulated_reviews as f32 / total_count as f32;
-        Ok(average_rating)
+        Some(average_rating)
     }
 }

src/graphql/query.rs

@@ -29,10 +29,10 @@ impl Query {
         &self,
         ctx: &Context<'a>,
         #[graphql(desc = "UUID of user to retrieve.")] id: Uuid,
-    ) -> Result<User> {
+    ) -> Result<Option<User>> {
         let db_client = ctx.data::<Database>()?;
         let collection: Collection<User> = db_client.collection::<User>("users");
-        query_object(&collection, id).await
+        query_object_optional(&collection, id).await
     }
 
     /// Entity resolver for product of specific UUID.
@@ -41,10 +41,10 @@ impl Query {
         &self,
         ctx: &Context<'a>,
         #[graphql(desc = "UUID of product to retrieve.")] id: Uuid,
-    ) -> Result<Product> {
+    ) -> Result<Option<Product>> {
         let db_client = ctx.data::<Database>()?;
         let collection: Collection<Product> = db_client.collection::<Product>("products");
-        query_object(&collection, id).await
+        query_object_optional(&collection, id).await
     }
 
     /// Entity resolver for product variant of specific UUID.
@@ -53,11 +53,11 @@ impl Query {
         &self,
         ctx: &Context<'a>,
         #[graphql(desc = "UUID of product variant to retrieve.")] id: Uuid,
-    ) -> Result<ProductVariant> {
+    ) -> Result<Option<ProductVariant>> {
         let db_client = ctx.data::<Database>()?;
         let collection: Collection<ProductVariant> =
             db_client.collection::<ProductVariant>("product_variants");
-        query_object(&collection, id).await
+        query_object_optional(&collection, id).await
     }
 
     /// Retrieves all reviews.
@@ -101,10 +101,10 @@ impl Query {
         &self,
         ctx: &Context<'a>,
         #[graphql(desc = "UUID of review to retrieve.")] id: Uuid,
-    ) -> Result<Review> {
+    ) -> Result<Option<Review>> {
         let db_client = ctx.data::<Database>()?;
         let collection: Collection<Review> = db_client.collection::<Review>("reviews");
-        query_object(&collection, id).await
+        query_object_optional(&collection, id).await
     }
 }
 
@@ -130,3 +130,21 @@ pub async fn query_object<T: for<'a> Deserialize<'a> + Unpin + Send + Sync>(
         }
     }
 }
+
+/// Shared function to query an optional object: `T` from a MongoDB collection of object: `T`.
+/// Used, since reviews can be null initially if no review exists for a new product variant.
+///
+/// * `collection` - Collection to query
+/// * `id` - UUID of object.
+pub async fn query_object_optional<T: for<'a> Deserialize<'a> + Unpin + Send + Sync>(
+    collection: &Collection<T>,
+    id: Uuid,
+) -> Result<Option<T>> {
+    match collection.find_one(doc! {"_id": id }, None).await {
+        Ok(maybe_object) => Ok(maybe_object),
+        Err(_) => {
+            let message = format!("{} with UUID: `{}` not found.", type_name::<T>(), id);
+            Err(Error::new(message))
+        }
+    }
+}

src/main.rs

@@ -11,7 +11,7 @@ use axum::{
     http::{header::HeaderMap, StatusCode},
     response::{self, IntoResponse},
     routing::{get, post},
-    Router, Server,
+    Router,
 };
 use clap::{arg, command, Parser};
 use event::http_event_service::{
@@ -25,6 +25,15 @@ use mongodb::{options::ClientOptions, Client, Database};
 
 use crate::graphql::{mutation::Mutation, query::Query};
 
+use once_cell::sync::Lazy;
+use axum_otel_metrics::HttpMetricsLayerBuilder;
+use axum_otel_metrics::HttpMetricsLayer;
+
+use opentelemetry::global;
+use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};
+use opentelemetry_sdk::Resource;
+use opentelemetry_otlp::WithExportConfig;
+
 mod authorization;
 mod event;
 mod graphql;
@@ -127,6 +136,37 @@ async fn graphql_handler(
     schema.execute(req).await.into()
 }
 
+static RESOURCE: Lazy<Resource> = Lazy::new(|| {
+    Resource::builder()
+        .with_service_name("review")
+        .build()
+});
+
+/// Initializes OpenTelemetry metrics exporter and sets the global meter provider.
+fn init_otlp() -> HttpMetricsLayer {
+    let exporter = opentelemetry_otlp::MetricExporter::builder()
+        .with_http()
+        .with_endpoint("http://otel-collector:4318/v1/metrics")
+        .with_temporality(Temporality::default())
+        .build()
+        .unwrap();
+
+    let reader = PeriodicReader::builder(exporter)
+        .with_interval(std::time::Duration::from_secs(5))
+        .build();
+
+    let provider = SdkMeterProvider::builder()
+        .with_reader(reader)
+        .with_resource(RESOURCE.clone())
+        .build();
+
+    global::set_meter_provider(provider.clone());
+
+    HttpMetricsLayerBuilder::new()
+        .with_provider(provider.clone())
+        .build()
+}
+
 /// Starts review service on port 8000.
 async fn start_service() {
     let client = db_connection().await;
@@ -143,11 +183,17 @@ async fn start_service() {
         .route("/health", get(StatusCode::OK))
         .with_state(schema);
     let dapr_router = build_dapr_router(db_client).await;
-    let app = Router::new().merge(graphiql).merge(dapr_router);
+    let metrics = init_otlp();
+
+    let app = Router::new()
+        .merge(graphiql)
+        .merge(dapr_router)
+        .layer(metrics);
 
     info!("GraphiQL IDE: http://0.0.0.0:8080");
-    Server::bind(&"0.0.0.0:8080".parse().unwrap())
-        .serve(app.into_make_service())
+
+    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await.unwrap();
+    axum::serve(listener, app)
         .await
         .unwrap();
 }