add: /:id/validator-messages route impl

samparsky · samparsky · commit 8a93cc377dd7 · 2020-01-28T13:06:04.000+01:00
diff --git a/sentry/src/db/channel.rs b/sentry/src/db/channel.rs
@@ -1,7 +1,9 @@
 use crate::db::DbPool;
 use bb8::RunError;
-use primitives::{Channel, ChannelId, ValidatorId};
+use primitives::{Channel, ChannelId, ValidatorId,};
+use primitives::validator::MessageTypes;
 use std::str::FromStr;
+use chrono::Utc;
 
 pub use list_channels::list_channels;
 
@@ -70,6 +72,31 @@ 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/routes/channel.rs b/sentry/src/routes/channel.rs
@@ -1,5 +1,5 @@
 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, list_channels, insert_validator_messages};
 use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
@@ -12,6 +12,10 @@ use primitives::sentry::{Event, SuccessResponse};
 use primitives::{Channel, ChannelId};
 use slog::error;
 use std::collections::HashMap;
+use primitives::channel::SpecValidator;
+use primitives::validator::MessageTypes;
+use futures::future::try_join_all;
+
 
 pub async fn channel_status<A: Adapter>(
     req: Request<Body>,
@@ -138,6 +142,39 @@ pub async fn insert_events<A: Adapter + 'static>(
         .unwrap())
 }
 
+pub async fn 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 messages = serde_json::from_slice::<Vec<MessageTypes>>(&body)?;
+    
+    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};
