Merge pull request #246 from samparsky/validator-messages

elpiel · web-flow · commit c29fdf685b88 · 2020-02-05T10:32:23.000+02:00
Validator messages
diff --git a/sentry/src/chain.rs b/sentry/src/chain.rs
@@ -7,19 +7,21 @@ use primitives::adapter::Adapter;
 //
 // function signature
 // fn middleware(mut req: Request) -> Result<Request, ResponseError>
-
-pub async fn chain<'a, A: Adapter + 'static, M>(
+#[allow(clippy::type_complexity)]
+pub async fn chain<'a, A: Adapter + 'static>(
     req: Request<Body>,
     app: &'a Application<A>,
-    middlewares: Vec<M>,
-) -> Result<Request<Body>, ResponseError>
-where
-    M: FnMut(
-            Request<Body>,
-            &'a Application<A>,
-        ) -> BoxFuture<'a, Result<Request<Body>, ResponseError>>
-        + 'static,
-{
+    middlewares: Vec<
+        Box<
+            dyn FnMut(
+                    Request<Body>,
+                    &'a Application<A>,
+                ) -> BoxFuture<'a, Result<Request<Body>, ResponseError>>
+                + 'static
+                + Send,
+        >,
+    >,
+) -> Result<Request<Body>, ResponseError> {
     let mut req = Ok(req);
 
     for mut mw in middlewares.into_iter() {
diff --git a/sentry/src/db/channel.rs b/sentry/src/db/channel.rs
@@ -1,5 +1,7 @@
 use crate::db::DbPool;
 use bb8::RunError;
+use chrono::Utc;
+use primitives::validator::MessageTypes;
 use primitives::{Channel, ChannelId, ValidatorId};
 use std::str::FromStr;
 
@@ -70,6 +72,30 @@ pub async fn insert_channel(
         .await
 }
 
+pub async fn insert_validator_messages(
+    pool: &DbPool,
+    channel: &Channel,
+    from: &ValidatorId,
+    validator_message: &MessageTypes,
+) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool
+        .run(move | connection| {
+            async move {
+                match connection.prepare("INSERT INTO validator_messages (channel_id, \"from\", msg, received) values ($1, $2, $3, $4)").await {
+                    Ok(stmt) => match connection.execute(&stmt, &[&channel.id, &from, &validator_message, &Utc::now()]).await {
+                        Ok(row) => {
+                            let inserted = row == 1;
+                            Ok((inserted, connection))
+                        },
+                        Err(e) => Err((e, connection)),
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+}
+
 mod list_channels {
     use crate::db::DbPool;
     use bb8::RunError;
diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs
@@ -19,7 +19,9 @@ use redis::aio::MultiplexedConnection;
 use regex::Regex;
 use routes::analytics::{advertiser_analytics, analytics, publisher_analytics};
 use routes::cfg::config;
-use routes::channel::{channel_list, create_channel, insert_events, last_approved};
+use routes::channel::{
+    channel_list, create_channel, create_validator_messages, insert_events, last_approved,
+};
 use slog::{error, Logger};
 use std::collections::HashMap;
 
@@ -54,6 +56,7 @@ lazy_static! {
     static ref ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
     static ref ADVERTISER_ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/for-advertiser/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
     static ref CREATE_EVENTS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events(/.*)?$").expect("The regex should be valid");
+
 }
 
 fn auth_required_middleware<'a, A: Adapter>(
@@ -251,6 +254,28 @@ async fn channels_router<A: Adapter + 'static>(
         };
 
         list_validator_messages(req, &app, &extract_params.0, &extract_params.1).await
+    } else if let (Some(caps), &Method::POST) = (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,
+            app,
+            vec![Box::new(auth_required_middleware), Box::new(channel_load)],
+        )
+        .await
+        {
+            Ok(req) => req,
+            Err(error) => {
+                return Err(error);
+            }
+        };
+
+        create_validator_messages(req, &app).await
     } else if let (Some(caps), &Method::GET) = (CHANNEL_EVENTS_AGGREGATES.captures(&path), method) {
         if req.extensions().get::<Session>().is_none() {
             return Err(ResponseError::Unauthorized);
diff --git a/sentry/src/routes/channel.rs b/sentry/src/routes/channel.rs
@@ -1,14 +1,17 @@
 use self::channel_list::ChannelListQuery;
-use crate::db::{get_channel_by_id, insert_channel, list_channels};
+use crate::db::{get_channel_by_id, insert_channel, insert_validator_messages, list_channels};
 use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
 use crate::RouteParams;
 use crate::Session;
+use futures::future::try_join_all;
 use hex::FromHex;
 use hyper::{Body, Request, Response};
 use primitives::adapter::Adapter;
+use primitives::channel::SpecValidator;
 use primitives::sentry::{Event, SuccessResponse};
+use primitives::validator::MessageTypes;
 use primitives::{Channel, ChannelId};
 use slog::error;
 use std::collections::HashMap;
@@ -138,6 +141,44 @@ pub async fn insert_events<A: Adapter + 'static>(
         .unwrap())
 }
 
+pub async fn create_validator_messages<A: Adapter + 'static>(
+    req: Request<Body>,
+    app: &Application<A>,
+) -> Result<Response<Body>, ResponseError> {
+    let session = req
+        .extensions()
+        .get::<Session>()
+        .expect("request session")
+        .to_owned();
+
+    let channel = req
+        .extensions()
+        .get::<Channel>()
+        .expect("Request should have Channel")
+        .to_owned();
+
+    let into_body = req.into_body();
+    let body = hyper::body::to_bytes(into_body).await?;
+    let request_body = serde_json::from_slice::<HashMap<String, Vec<MessageTypes>>>(&body)?;
+    let messages = request_body
+        .get("messages")
+        .ok_or_else(|| ResponseError::BadRequest("missing messages body".to_string()))?;
+
+    match channel.spec.validators.find(&session.uid) {
+        SpecValidator::None => Err(ResponseError::Unauthorized),
+        _ => {
+            try_join_all(messages.iter().map(|message| {
+                insert_validator_messages(&app.pool, &channel, &session.uid, &message)
+            }))
+            .await?;
+
+            Ok(success_response(serde_json::to_string(&SuccessResponse {
+                success: true,
+            })?))
+        }
+    }
+}
+
 mod channel_list {
     use chrono::serde::ts_seconds::deserialize as ts_seconds;
     use chrono::{DateTime, Utc};
