test(costs): get latest costs from api

fix: fmt

chore: add test, incl finalized only

fix: serde for query string

test: more shenanigans with query string

test: more shenanigans with query string

fix: option error handling

committer/src/api.rs

@@ -90,9 +90,17 @@ async fn metrics(registry: web::Data<Arc<Registry>>) -> impl Responder {
     std::result::Result::<_, InternalError<_>>::Ok(text)
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "lowercase")]
+enum HeightVariant {
+    Latest,
+    Specific,
+}
+
 #[derive(Deserialize)]
 struct CostQueryParams {
-    from_height: u32,
+    variant: HeightVariant,
+    value: Option<u32>,
     limit: Option<usize>,
 }
 
@@ -103,8 +111,18 @@ async fn costs(
 ) -> impl Responder {
     let limit = query.limit.unwrap_or(100);
 
-    match data.get_costs(query.from_height, limit).await {
-        Ok(bundle_costs) => HttpResponse::Ok().json(bundle_costs),
+    let response = match query.variant {
+        HeightVariant::Latest => data.get_latest_costs(limit).await,
+        HeightVariant::Specific => match query.value {
+            Some(height) => data.get_costs(height, limit).await,
+            None => Err(services::Error::Other(
+                "height value is required".to_string(),
+            )),
+        },
+    };
+
+    match response {
+        Ok(costs) => HttpResponse::Ok().json(costs),
         Err(services::Error::Other(e)) => {
             HttpResponse::from_error(InternalError::new(e, StatusCode::BAD_REQUEST))
         }

packages/adapters/storage/src/lib.rs

@@ -50,6 +50,10 @@ impl services::cost_reporter::port::Storage for Postgres {
             .await
             .map_err(Into::into)
     }
+
+    async fn get_latest_costs(&self, limit: usize) -> Result<Vec<BundleCost>> {
+        self._get_latest_costs(limit).await.map_err(Into::into)
+    }
 }
 
 impl services::status_reporter::port::Storage for Postgres {
@@ -1163,4 +1167,38 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn get_latest_finalized_costs() -> Result<()> {
+        use services::cost_reporter::port::Storage;
+
+        // given
+        let storage = start_db().await;
+
+        for i in 0..5 {
+            let start_height = i * 10 + 1;
+            let end_height = start_height + 9;
+            let block_range = start_height..=end_height;
+
+            ensure_finalized_fragments_exist_in_the_db(
+                storage.clone(),
+                block_range,
+                1000u128,
+                5000u64,
+            )
+            .await;
+        }
+
+        // when
+        let finalized_costs = storage.get_latest_costs(1).await?;
+
+        // then
+        assert_eq!(finalized_costs.len(), 1);
+        let finalized_cost = &finalized_costs[0];
+
+        assert_eq!(finalized_cost.start_height, 41);
+        assert_eq!(finalized_cost.end_height, 50);
+
+        Ok(())
+    }
 }

packages/adapters/storage/src/postgres.rs

@@ -861,6 +861,36 @@ impl Postgres {
         .collect::<Result<Vec<_>>>()
     }
 
+    pub(crate) async fn _get_latest_costs(&self, limit: usize) -> Result<Vec<BundleCost>> {
+        sqlx::query_as!(
+            tables::BundleCost,
+            r#"
+            SELECT
+                bc.bundle_id,
+                bc.cost,
+                bc.size,
+                bc.da_block_height,
+                bc.is_finalized,
+                b.start_height,
+                b.end_height
+            FROM
+                bundle_cost bc
+                JOIN bundles b ON bc.bundle_id = b.id
+            WHERE
+                bc.is_finalized = TRUE
+            ORDER BY
+                b.start_height DESC
+            LIMIT $1
+            "#,
+            limit as i64
+        )
+        .fetch_all(&self.connection_pool)
+        .await?
+        .into_iter()
+        .map(BundleCost::try_from)
+        .collect::<Result<Vec<_>>>()
+    }
+
     pub(crate) async fn _next_bundle_id(&self) -> Result<NonNegative<i32>> {
         let next_id = sqlx::query!("SELECT nextval(pg_get_serial_sequence('bundles', 'id'))")
             .fetch_one(&self.connection_pool)

packages/adapters/storage/src/test_instance.rs

@@ -1,9 +1,3 @@
-use std::{
-    borrow::Cow,
-    ops::RangeInclusive,
-    sync::{Arc, Weak},
-};
-
 use delegate::delegate;
 use services::{
     block_bundler, block_committer, block_importer,
@@ -14,6 +8,11 @@ use services::{
     },
 };
 use sqlx::Executor;
+use std::{
+    borrow::Cow,
+    ops::RangeInclusive,
+    sync::{Arc, Weak},
+};
 use testcontainers::{
     core::{ContainerPort, WaitFor},
     runners::AsyncRunner,
@@ -351,4 +350,8 @@ impl services::cost_reporter::port::Storage for DbWithProcess {
             .await
             .map_err(Into::into)
     }
+
+    async fn get_latest_costs(&self, limit: usize) -> services::Result<Vec<BundleCost>> {
+        self.db._get_latest_costs(limit).await.map_err(Into::into)
+    }
 }

packages/services/src/cost_reporter.rs

@@ -36,6 +36,17 @@ pub mod service {
                 .get_finalized_costs(from_block_height, limit)
                 .await
         }
+
+        pub async fn get_latest_costs(&self, limit: usize) -> Result<Vec<BundleCost>> {
+            if limit > self.request_limit {
+                return Err(Error::Other(format!(
+                    "requested: {} items, but limit is: {}",
+                    limit, self.request_limit
+                )));
+            }
+
+            self.storage.get_latest_costs(limit).await
+        }
     }
 }
 
@@ -50,5 +61,7 @@ pub mod port {
             from_block_height: u32,
             limit: usize,
         ) -> Result<Vec<BundleCost>>;
+
+        async fn get_latest_costs(&self, limit: usize) -> Result<Vec<BundleCost>>;
     }
 }