Merge pull request #207 from AdExNetwork/issue-9-validator-messages

Issue #9 Get Channel's Validator messages

Author: elpiel

primitives/src/config.rs

@@ -23,6 +23,7 @@ pub struct Config {
     pub heartbeat_time: u32, // in milliseconds
     pub channels_find_limit: u32,
     pub events_find_limit: u32,
+    pub msgs_find_limit: u32,
     pub health_threshold_promilles: u32,
     pub health_unsignable_promilles: u32,
     pub propagation_timeout: u32,

primitives/src/sentry.rs

@@ -1,5 +1,5 @@
 use crate::validator::{ApproveState, Heartbeat, MessageTypes, NewState};
-use crate::{BigNum, Channel, ChannelId};
+use crate::{BigNum, Channel, ChannelId, ValidatorId};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -99,7 +99,7 @@ pub struct SuccessResponse {
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ValidatorMessage {
-    pub from: String,
+    pub from: ValidatorId,
     pub received: DateTime<Utc>,
     pub msg: MessageTypes,
 }
@@ -114,3 +114,36 @@ pub struct ValidatorMessageResponse {
 pub struct EventAggregateResponse {
     pub events: Vec<EventAggregate>,
 }
+
+#[cfg(feature = "postgres")]
+mod postgres {
+    use super::ValidatorMessage;
+    use crate::validator::MessageTypes;
+    use bytes::BytesMut;
+    use postgres_types::{accepts, to_sql_checked, IsNull, Json, ToSql, Type};
+    use std::error::Error;
+    use tokio_postgres::Row;
+
+    impl From<&Row> for ValidatorMessage {
+        fn from(row: &Row) -> Self {
+            Self {
+                from: row.get("from"),
+                received: row.get("received"),
+                msg: row.get::<_, Json<MessageTypes>>("msg").0,
+            }
+        }
+    }
+
+    impl ToSql for MessageTypes {
+        fn to_sql(
+            &self,
+            ty: &Type,
+            w: &mut BytesMut,
+        ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
+            Json(self).to_sql(ty, w)
+        }
+
+        accepts!(JSONB);
+        to_sql_checked!();
+    }
+}

sentry/src/db.rs

@@ -8,7 +8,10 @@ use std::env;
 use lazy_static::lazy_static;
 
 mod channel;
+mod validator_message;
+
 pub use self::channel::*;
+pub use self::validator_message::*;
 
 pub type DbPool = Pool<PostgresConnectionManager<NoTls>>;
 

sentry/src/db/validator_message.rs

@@ -0,0 +1,56 @@
+use crate::db::DbPool;
+use bb8::RunError;
+use bb8_postgres::tokio_postgres::types::ToSql;
+use primitives::sentry::ValidatorMessage;
+use primitives::{ChannelId, ValidatorId};
+
+pub async fn get_validator_messages(
+    pool: &DbPool,
+    channel_id: &ChannelId,
+    validator_id: &Option<ValidatorId>,
+    message_types: &[String],
+    limit: u64,
+) -> Result<Vec<ValidatorMessage>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    let mut where_clauses: Vec<String> = vec!["channel_id = $1".to_string()];
+    let mut params: Vec<&(dyn ToSql + Sync)> = vec![&channel_id];
+
+    if let Some(validator_id) = validator_id {
+        where_clauses.push(format!(r#""from" = ${}"#, params.len() + 1));
+        params.push(validator_id);
+    }
+
+    add_message_types_params(&mut where_clauses, &mut params, message_types);
+
+    pool
+        .run(move |connection| {
+            async move {
+                let statement = format!(r#"SELECT "from", msg, received FROM validator_messages WHERE {} ORDER BY received DESC LIMIT {}"#, where_clauses.join(" AND "), limit);
+                match connection.prepare(&statement).await {
+                    Ok(select) => match connection.query(&select, params.as_slice()).await {
+                        Ok(results) => {
+                            let messages = results.iter().map(ValidatorMessage::from).collect();
+                            Ok((messages, connection))},
+                        Err(e) => Err((e, connection)),
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+}
+
+fn add_message_types_params<'a>(
+    where_clauses: &mut Vec<String>,
+    params: &mut Vec<&'a (dyn ToSql + Sync)>,
+    message_types: &'a [String],
+) {
+    let mut msg_prep = vec![];
+    for message_type in message_types.iter() {
+        msg_prep.push(format!("${}", params.len() + 1));
+        params.push(message_type);
+    }
+
+    if !msg_prep.is_empty() {
+        where_clauses.push(format!("msg->>'type' IN ({})", msg_prep.join(",")));
+    }
+}

sentry/src/lib.rs

@@ -7,6 +7,7 @@ use crate::middleware::auth;
 use crate::middleware::channel::channel_load;
 use crate::middleware::cors::{cors, Cors};
 use crate::routes::channel::channel_status;
+use crate::routes::validator_message::{extract_params, list_validator_messages};
 use hyper::{Body, Method, Request, Response, StatusCode};
 use lazy_static::lazy_static;
 use primitives::adapter::Adapter;
@@ -27,6 +28,7 @@ pub mod middleware {
 pub mod routes {
     pub mod cfg;
     pub mod channel;
+    pub mod validator_message;
 }
 
 pub mod access;
@@ -39,6 +41,8 @@ lazy_static! {
         Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
     static ref LAST_APPROVED_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/last-approved/?$").expect("The regex should be valid");
     static ref CHANNEL_STATUS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/status/?$").expect("The regex should be valid");
+    // Only the initial Regex to be matched.
+    static ref CHANNEL_VALIDATOR_MESSAGES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/validator-messages(/.*)?$").expect("The regex should be valid");
     // @TODO define other regex routes
 }
 
@@ -106,18 +110,19 @@ impl<A: Adapter + 'static> Application<A> {
             }
         };
 
-        let mut response = match (req.uri().path(), req.method()) {
+        let path = req.uri().path().to_string();
+        let mut response = match (path.as_ref(), req.method()) {
             ("/cfg", &Method::GET) => config(req, &self).await,
             ("/channel", &Method::POST) => create_channel(req, &self).await,
             ("/channel/list", &Method::GET) => channel_list(req, &self).await,
             // This is important becuase it prevents us from doing
             // expensive regex matching for routes without /channel
-            (route, method) if route.starts_with("/channel") => {
+            (path, method) if path.starts_with("/channel") => {
                 // example with
                 // @TODO remove later
                 // regex matching for routes with params
                 if let (Some(caps), &Method::GET) =
-                    (LAST_APPROVED_BY_CHANNEL_ID.captures(route), method)
+                    (LAST_APPROVED_BY_CHANNEL_ID.captures(path), method)
                 {
                     let param = RouteParams(vec![caps
                         .get(1)
@@ -135,7 +140,7 @@ impl<A: Adapter + 'static> Application<A> {
 
                     last_approved(req, &self).await
                 } else if let (Some(caps), &Method::GET) =
-                    (CHANNEL_STATUS_BY_CHANNEL_ID.captures(route), method)
+                    (CHANNEL_STATUS_BY_CHANNEL_ID.captures(path), method)
                 {
                     let param = RouteParams(vec![caps
                         .get(1)
@@ -150,6 +155,31 @@ impl<A: Adapter + 'static> Application<A> {
                     };
 
                     channel_status(req, &self).await
+                } else if let (Some(caps), &Method::GET) =
+                    (CHANNEL_VALIDATOR_MESSAGES.captures(path), method)
+                {
+                    let param = RouteParams(vec![caps
+                        .get(1)
+                        .map_or("".to_string(), |m| m.as_str().to_string())]);
+                    req.extensions_mut().insert(param);
+
+                    let req = match chain(req, &self, vec![channel_load]).await {
+                        Ok(req) => req,
+                        Err(error) => {
+                            return map_response_error(error);
+                        }
+                    };
+
+                    // @TODO: Move this to a middleware?!
+                    let extract_params =
+                        match extract_params(caps.get(2).map_or("", |m| m.as_str())) {
+                            Ok(params) => params,
+                            Err(error) => {
+                                return map_response_error(error.into());
+                            }
+                        };
+
+                    list_validator_messages(req, &self, &extract_params.0, &extract_params.1).await
                 } else {
                     Err(ResponseError::NotFound)
                 }

sentry/src/routes/validator_message.rs

@@ -0,0 +1,70 @@
+use crate::db::get_validator_messages;
+use crate::{success_response, Application, ResponseError};
+use hyper::{Body, Request, Response};
+use primitives::adapter::Adapter;
+use primitives::sentry::ValidatorMessageResponse;
+use primitives::{Channel, DomainError, ValidatorId};
+use serde::Deserialize;
+use std::convert::TryFrom;
+
+#[derive(Deserialize)]
+pub struct ValidatorMessagesListQuery {
+    limit: Option<u64>,
+}
+
+pub fn extract_params(from_path: &str) -> Result<(Option<ValidatorId>, Vec<String>), DomainError> {
+    // trim the `/` at the beginning & end if there is one or more
+    // and split the rest of the string at the `/`
+    let split: Vec<&str> = from_path.trim_matches('/').split('/').collect();
+
+    if split.len() > 2 {
+        return Err(DomainError::InvalidArgument(
+            "Too many parameters".to_string(),
+        ));
+    }
+
+    let validator_id = split
+        .get(0)
+        // filter an empty string
+        .filter(|string| !string.is_empty())
+        // then try to map it to ValidatorId
+        .map(|string| ValidatorId::try_from(*string))
+        // Transpose in order to check for an error from the conversion
+        .transpose()?;
+
+    let message_types = split
+        .get(1)
+        .filter(|string| !string.is_empty())
+        .map(|string| string.split('+').map(|s| s.to_string()).collect());
+
+    Ok((validator_id, message_types.unwrap_or_default()))
+}
+
+pub async fn list_validator_messages<A: Adapter>(
+    req: Request<Body>,
+    app: &Application<A>,
+    validator_id: &Option<ValidatorId>,
+    message_types: &[String],
+) -> Result<Response<Body>, ResponseError> {
+    let query =
+        serde_urlencoded::from_str::<ValidatorMessagesListQuery>(&req.uri().query().unwrap_or(""))?;
+
+    let channel = req
+        .extensions()
+        .get::<Channel>()
+        .expect("Request should have Channel");
+
+    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, validator_id, message_types, limit).await?;
+
+    let response = ValidatorMessageResponse { validator_messages };
+
+    Ok(success_response(serde_json::to_string(&response)?))
+}