Merge pull request #541 from AmbireTech/sentry-validator-messages-routes-to-axum

Sentry validator messages routes to axum

Author: elpiel

docker-compose.harness.yml

@@ -13,9 +13,10 @@ services:
       POSTGRES_USER: 'postgres'
       POSTGRES_PASSWORD: 'postgres'
       # harness_* databases are used by the `test_harness` crate for testing
-      # `sentry_leader` is the default database used by `sentry` for running tests
+      # `sentry_leader` is the default database used by `sentry` for running tests and the leader
+      # `sentry_leader` is for running the local follower when maunally testing
       # `primitives` is used for running tests in the `primitives` crate
-      POSTGRES_MULTIPLE_DATABASES: harness_leader,harness_follower,sentry_leader,primitives
+      POSTGRES_MULTIPLE_DATABASES: harness_leader,harness_follower,sentry_leader,sentry_follower,primitives
     networks:
       - adex-external
 

sentry/src/middleware/campaign.rs

@@ -10,6 +10,14 @@ use axum::{
 };
 use hyper::{Body, Request};
 use primitives::{campaign::Campaign, CampaignId, ChainOf};
+use serde::Deserialize;
+
+/// This struct is required because of routes that have more parameters
+/// apart from the `CampaignId`
+#[derive(Debug, Deserialize)]
+struct CampaignParam {
+    pub id: CampaignId,
+}
 
 #[derive(Debug)]
 pub struct CampaignLoad;
@@ -80,9 +88,10 @@ where
     let mut request_parts = RequestParts::new(request);
 
     let campaign_id = request_parts
-        .extract::<Path<CampaignId>>()
+        .extract::<Path<CampaignParam>>()
         .await
-        .map_err(|_| ResponseError::BadRequest("Bad Campaign Id".to_string()))?;
+        .map_err(|_| ResponseError::BadRequest("Bad Campaign Id".to_string()))?
+        .id;
 
     let campaign = fetch_campaign(pool.clone(), &campaign_id)
         .await?

sentry/src/middleware/channel.rs

@@ -1,9 +1,5 @@
 use std::sync::Arc;
 
-use crate::{
-    db::get_channel_by_id, middleware::Middleware, response::ResponseError,
-    routes::routers::RouteParams, Application, Auth,
-};
 use adapter::client::Locked;
 use axum::{
     extract::{Path, RequestParts},
@@ -15,6 +11,12 @@ use hyper::{Body, Request};
 use primitives::ChannelId;
 
 use async_trait::async_trait;
+use serde::Deserialize;
+
+use crate::{
+    db::get_channel_by_id, middleware::Middleware, response::ResponseError,
+    routes::routers::RouteParams, Application, Auth,
+};
 
 #[derive(Debug)]
 pub struct ChannelLoad;
@@ -70,6 +72,13 @@ fn channel_load_old<C: Locked>(
     .boxed()
 }
 
+/// This struct is required because of routes that have more parameters
+/// apart from the `ChannelId`
+#[derive(Debug, Deserialize)]
+struct ChannelParam {
+    pub id: ChannelId,
+}
+
 pub async fn channel_load<C: Locked + 'static, B>(
     request: axum::http::Request<B>,
     next: Next<B>,
@@ -86,12 +95,12 @@ where
     // running extractors requires a `RequestParts`
     let mut request_parts = RequestParts::new(request);
 
-    let channel_id = request_parts
-        .extract::<Path<ChannelId>>()
+    let channel_param = request_parts
+        .extract::<Path<ChannelParam>>()
         .await
         .map_err(|_| ResponseError::BadRequest("Bad Channel Id".to_string()))?;
 
-    let channel = get_channel_by_id(&app.pool, &channel_id)
+    let channel = get_channel_by_id(&app.pool, &channel_param.id)
         .await?
         .ok_or(ResponseError::NotFound)?;
 

sentry/src/routes/channel.rs

@@ -926,6 +926,13 @@ pub async fn channel_payout<C: Locked + 'static>(
     })?))
 }
 
+/// POST `/v5/channel/dummy-deposit` request
+///
+/// Full details about the route's API and intend can be found in the [`routes`](crate::routes#post-v5channeldummy-deposit-auth-required) module
+///
+/// Request body (json): [`ChannelDummyDeposit`]
+///
+/// Response: [`SuccessResponse`]
 pub async fn channel_dummy_deposit_axum<C: Locked + 'static>(
     Extension(app): Extension<Arc<Application<C>>>,
     Extension(auth): Extension<Auth>,
@@ -1016,7 +1023,10 @@ pub async fn channel_dummy_deposit<C: Locked + 'static>(
 /// starting with `/v5/channel/0xXXX.../validator-messages`
 ///
 pub mod validator_message {
+    use std::sync::Arc;
+
     use crate::{
+        application::Qs,
         db::validator_message::{get_validator_messages, insert_validator_message},
         Auth,
     };
@@ -1025,15 +1035,17 @@ pub mod validator_message {
         Application,
     };
     use adapter::client::Locked;
+    use axum::{extract::Path, Extension, Json};
     use futures::future::try_join_all;
     use hyper::{Body, Request, Response};
     use primitives::{
         sentry::{
             SuccessResponse, ValidatorMessagesCreateRequest, ValidatorMessagesListQuery,
             ValidatorMessagesListResponse,
         },
-        ChainOf, Channel, DomainError, ValidatorId,
+        ChainOf, Channel, ChannelId, DomainError, ValidatorId,
     };
+    use serde::Deserialize;
 
     pub fn extract_params(
         from_path: &str,
@@ -1065,6 +1077,108 @@ pub mod validator_message {
         Ok((validator_id, message_types.unwrap_or_default()))
     }
 
+    #[derive(Debug, Deserialize)]
+    pub struct MessagesListParams {
+        pub id: ChannelId,
+        // Optional filtering by ValidatorId
+        #[serde(default)]
+        pub address: Option<ValidatorId>,
+        /// List of validator message types separated by `+` (urlencoded):
+        /// e.g. `ApproveState+NewState`
+        #[serde(default)]
+        pub message_types: String,
+    }
+
+    /// GET `/v5/channel/0xXXX.../validator-messages`
+    ///
+    /// Full details about the route's API and intend can be found in the [`routes`](crate::routes#get-v5channelidvalidator-messages) module
+    ///
+    /// Request query parameters: [`ValidatorMessagesListQuery`]
+    ///
+    /// Response: [`ValidatorMessagesListResponse`]
+    ///
+    pub async fn list_validator_messages_axum<C: Locked + 'static>(
+        Extension(app): Extension<Arc<Application<C>>>,
+        Extension(channel_context): Extension<ChainOf<Channel>>,
+        Path(params): Path<MessagesListParams>,
+        Qs(query): Qs<ValidatorMessagesListQuery>,
+    ) -> Result<Json<ValidatorMessagesListResponse>, ResponseError> {
+        let message_types = {
+            // We need to strip the `/` prefix from axum
+            let stripped = params
+                .message_types
+                .strip_prefix('/')
+                .unwrap_or(&params.message_types);
+
+            if !stripped.is_empty() {
+                stripped
+                    .split('+')
+                    .map(|s| s.to_string())
+                    .collect::<Vec<_>>()
+            } else {
+                vec![]
+            }
+        };
+
+        let channel = channel_context.context;
+
+        let config_limit = app.config.msgs_find_limit as u64;
+        let limit = query
+            .limit
+            .filter(|n| *n >= 1)
+            .unwrap_or(config_limit)
+            .min(config_limit);
+
+        let validator_messages = get_validator_messages(
+            &app.pool,
+            &channel.id(),
+            &params.address,
+            &message_types,
+            limit,
+        )
+        .await?;
+
+        Ok(Json(ValidatorMessagesListResponse {
+            messages: validator_messages,
+        }))
+    }
+
+    /// POST `/v5/channel/0xXXX.../validator-messages`
+    ///
+    /// Full details about the route's API and intend can be found in the [`routes`](crate::routes#post-v5channelidvalidator-messages-auth-required) module
+    ///
+    /// Request body (json): [`ValidatorMessagesCreateRequest`]
+    ///
+    /// Response: [`SuccessResponse`]
+    ///
+    /// # Examples
+    ///
+    /// Request:
+    ///
+    /// ```
+    #[doc = include_str!("../../../primitives/examples/validator_messages_create_request.rs")]
+    /// ```
+    pub async fn create_validator_messages_axum<C: Locked + 'static>(
+        Extension(app): Extension<Arc<Application<C>>>,
+        Extension(auth): Extension<Auth>,
+        Extension(channel_context): Extension<ChainOf<Channel>>,
+        Json(create_request): Json<ValidatorMessagesCreateRequest>,
+    ) -> Result<Json<SuccessResponse>, ResponseError> {
+        let channel = channel_context.context;
+
+        match channel.find_validator(auth.uid) {
+            None => Err(ResponseError::Unauthorized),
+            _ => {
+                try_join_all(create_request.messages.iter().map(|message| {
+                    insert_validator_message(&app.pool, &channel, &auth.uid, message)
+                }))
+                .await?;
+
+                Ok(Json(SuccessResponse { success: true }))
+            }
+        }
+    }
+
     /// GET `/v5/channel/0xXXX.../validator-messages`
     ///
     /// Full details about the route's API and intend can be found in the [`routes`](crate::routes#get-v5channelidvalidator-messages) module

sentry/src/routes/routers.rs

@@ -58,7 +58,10 @@ use tower::ServiceBuilder;
 
 use super::{
     analytics::{analytics_axum, GET_ANALYTICS_ALLOWED_KEYS},
-    channel::{channel_dummy_deposit_axum, channel_list_axum, channel_payout_axum},
+    channel::{
+        channel_dummy_deposit_axum, channel_list_axum, channel_payout_axum,
+        validator_message::{create_validator_messages_axum, list_validator_messages_axum},
+    },
     units_for_slot::post_units_for_slot,
 };
 
@@ -162,6 +165,20 @@ pub fn channels_router_axum<C: Locked + 'static>() -> Router {
         .route("/accounting", get(get_accounting_for_channel_axum::<C>))
         .route("/last-approved", get(last_approved_axum::<C>))
         .nest("/spender", spender_routes)
+        .route(
+            "/validator-messages",
+            post(create_validator_messages_axum::<C>)
+                .route_layer(middleware::from_fn(authentication_required::<C, _>)),
+        )
+        .route(
+            "/validator-messages",
+            get(list_validator_messages_axum::<C>),
+        )
+        // We allow Message Type filtering only when filtering by a ValidatorId
+        .route(
+            "/validator-messages/:address/*message_types",
+            get(list_validator_messages_axum::<C>),
+        )
         .layer(
             // keeps the order from top to bottom!
             ServiceBuilder::new()