fix: add /:id/events route

samparsky · samparsky · commit 65453790492d · 2020-01-21T11:48:00.000+01:00
diff --git a/sentry/src/access.rs b/sentry/src/access.rs
@@ -2,12 +2,12 @@ use chrono::Utc;
 use futures::future::try_join_all;
 use redis::aio::MultiplexedConnection;
 
+use crate::Session;
 use primitives::event_submission::{RateLimit, Rule};
 use primitives::sentry::Event;
 use primitives::Channel;
 use std::cmp::PartialEq;
 use std::error::Error;
-use crate::Session;
 use std::fmt;
 
 #[derive(Debug, PartialEq, Eq)]
diff --git a/sentry/src/db.rs b/sentry/src/db.rs
@@ -8,8 +8,8 @@ use std::env;
 use lazy_static::lazy_static;
 
 pub mod analytics;
-pub mod event_aggregate;
 mod channel;
+pub mod event_aggregate;
 mod validator_message;
 
 pub use self::channel::*;
diff --git a/sentry/src/db/event_aggregate.rs b/sentry/src/db/event_aggregate.rs
@@ -3,7 +3,7 @@ use bb8::RunError;
 use bb8_postgres::tokio_postgres::types::ToSql;
 use chrono::{DateTime, Utc};
 use primitives::sentry::EventAggregate;
-use primitives::{ValidatorId, ChannelId};
+use primitives::{ChannelId, ValidatorId};
 
 pub async fn list_event_aggregates(
     pool: &DbPool,
@@ -56,8 +56,8 @@ pub async fn list_event_aggregates(
 pub async fn insert_event_aggregate(
     pool: &DbPool,
     channel_id: &ChannelId,
-    event: &EventAggregate
-) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>>  {
+    event: &EventAggregate,
+) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>> {
     let mut values = Vec::new();
     let mut index = 0;
     let id = channel_id.to_string();
@@ -70,7 +70,13 @@ pub async fn insert_event_aggregate(
                 let event_count = value.to_string();
                 let event_payout = aggr.event_payouts[earner].to_string();
 
-                data.extend(vec![id.clone(), event_type.clone(), earner.clone(), event_count, event_payout]);
+                data.extend(vec![
+                    id.clone(),
+                    event_type.clone(),
+                    earner.clone(),
+                    event_count,
+                    event_payout,
+                ]);
                 //
                 // this is a work around for bulk inserts
                 // rust-postgres does not have native support for bulk inserts
@@ -79,35 +85,40 @@ pub async fn insert_event_aggregate(
                 // i.e.
                 // INSERT INTO event_aggregates (_, _) VALUES ($1, $2), ($3, $4), ($5, $6)
 
-                values.push(format!("(${}, ${}, ${}, ${}, ${})", index+1, index+2, index+3, index+4, index+5));
+                values.push(format!(
+                    "(${}, ${}, ${}, ${}, ${})",
+                    index + 1,
+                    index + 2,
+                    index + 3,
+                    index + 4,
+                    index + 5
+                ));
                 index += 5;
             }
         }
     }
 
     let inserts: Vec<&(dyn ToSql + Sync)> = data.iter().map(|x| x as &(dyn ToSql + Sync)).collect();
 
-
     //    the created field is supplied by postgres Default
     let query = format!("INSERT INTO event_aggregates (channel_id, event_type, earner, event_counts, event_payouts) values {}", values.join(" ,"));
 
     let result = pool
-        .run(
-            move |connection | {
-                async move {
-                    match connection.prepare(&query).await {
-                        Ok(stmt) => match connection.execute(&stmt, &inserts.as_slice()).await {
-                            Ok(row) => {
-                                let inserted = row == (index / 5);
-                                Ok((inserted, connection))
-                            },
-                            Err(e) => Err((e, connection)),
-                        },
+        .run(move |connection| {
+            async move {
+                match connection.prepare(&query).await {
+                    Ok(stmt) => match connection.execute(&stmt, &inserts.as_slice()).await {
+                        Ok(row) => {
+                            let inserted = row == (index / 5);
+                            Ok((inserted, connection))
+                        }
                         Err(e) => Err((e, connection)),
-                    }
+                    },
+                    Err(e) => Err((e, connection)),
                 }
             }
-        ).await?;
+        })
+        .await?;
 
     Ok(result)
 }
diff --git a/sentry/src/event_aggregator.rs b/sentry/src/event_aggregator.rs
@@ -1,25 +1,25 @@
+use crate::access::check_access;
+use crate::db::event_aggregate::insert_event_aggregate;
+use crate::db::DbPool;
 use crate::event_reducer;
-use primitives::{ChannelId, Channel};
+use crate::Application;
+use crate::ResponseError;
+use crate::Session;
+use chrono::{Duration, Utc};
 use primitives::adapter::Adapter;
-use primitives::sentry::{EventAggregate, Event};
+use primitives::sentry::{Event, EventAggregate};
+use primitives::{Channel, ChannelId};
 use std::collections::HashMap;
-use crate::db::event_aggregate::insert_event_aggregate;
-use crate::Session;
-use crate::access::check_access;
-use crate::Application;
-use chrono::{Utc, Duration};
-use crate::db::DbPool;
-use tokio::time::{delay_for};
 use std::time::Duration as TimeDuration;
-use crate::ResponseError;
+use tokio::time::delay_for;
 
 // use futures::
 use async_std::sync::RwLock;
-use std::sync::{Arc};
+use std::sync::Arc;
 
 #[derive(Default, Clone)]
 pub struct EventAggregator {
-    aggregate: Arc<RwLock<HashMap<String, EventAggregate>>>
+    aggregate: Arc<RwLock<HashMap<String, EventAggregate>>>,
 }
 
 pub fn new_aggr(channel_id: &ChannelId) -> EventAggregate {
@@ -30,7 +30,11 @@ pub fn new_aggr(channel_id: &ChannelId) -> EventAggregate {
     }
 }
 
-async fn store(db: &DbPool, channel_id: &ChannelId, aggr: Arc<RwLock<HashMap<String, EventAggregate>>>) {
+async fn store(
+    db: &DbPool,
+    channel_id: &ChannelId,
+    aggr: Arc<RwLock<HashMap<String, EventAggregate>>>,
+) {
     let mut recorder = aggr.write().await;
     let ev_aggr: Option<&EventAggregate> = recorder.get(&channel_id.to_string());
     if let Some(data) = ev_aggr {
@@ -44,58 +48,67 @@ async fn store(db: &DbPool, channel_id: &ChannelId, aggr: Arc<RwLock<HashMap<Str
 }
 
 impl EventAggregator {
-    pub async fn record<'a, A: Adapter + 'static>(
-        &'static self,
-        app: &'static Application<A>,
+    pub async fn record<'a, A: Adapter>(
+        &self,
+        app: &'a Application<A>,
         channel: Channel,
         session: Session,
         events: &'a [Event],
-        ) -> Result<(), ResponseError>
-    {
-        let has_access = check_access(&app.redis, &session, &app.config.ip_rate_limit, &channel, events).await;
+    ) -> Result<(), ResponseError> {
+        let has_access = check_access(
+            &app.redis,
+            &session,
+            &app.config.ip_rate_limit,
+            &channel,
+            events,
+        )
+        .await;
         if let Err(e) = has_access {
             return Err(ResponseError::BadRequest(e.to_string()));
         }
 
         let mut recorder = self.aggregate.write().await;
-        let mut aggr: &mut EventAggregate = if let Some(aggr) = recorder.get_mut(&channel.id.to_string()) {
-            aggr
-        } else {
-            // insert into 
-            recorder.insert(channel.id.to_string(), new_aggr(&channel.id));
-            recorder.get_mut(&channel.id.to_string()).expect("should have aggr, we just inserted")
-        };
+        let mut aggr: &mut EventAggregate =
+            if let Some(aggr) = recorder.get_mut(&channel.id.to_string()) {
+                aggr
+            } else {
+                // insert into
+                recorder.insert(channel.id.to_string(), new_aggr(&channel.id));
+                recorder
+                    .get_mut(&channel.id.to_string())
+                    .expect("should have aggr, we just inserted")
+            };
 
         // if aggr is none
-        events.iter().for_each( | ev| event_reducer::reduce(&channel, &mut aggr, ev));
+        events
+            .iter()
+            .for_each(|ev| event_reducer::reduce(&channel, &mut aggr, ev));
         let created = aggr.created;
+        let dbpool = app.pool.clone();
+        let aggr_throttle = app.config.aggr_throttle;
+        let aggregate = self.aggregate.clone();
 
         // drop write access to RwLock
         // this is required to prevent a deadlock in store
         drop(recorder);
 
         // Checks if aggr_throttle is set
         // and if current time is greater than aggr.created plus throttle seconds
-        // 
+        //
         // This approach spawns an async task every > AGGR_THROTTLE seconds
         // Each spawned task resolves after AGGR_THROTTLE seconds
-        // 
+        //
 
-        if app.config.aggr_throttle > 0
-            &&
-            Utc::now() > (created + Duration::seconds(app.config.aggr_throttle as i64))
-        {
+        if aggr_throttle > 0 && Utc::now() > (created + Duration::seconds(aggr_throttle as i64)) {
             // spawn a tokio task for saving to database
-            tokio::spawn(
-                async move {
-                    delay_for(TimeDuration::from_secs(app.config.aggr_throttle as u64)).await;
-                    store(&app.pool.clone(), &channel.id, self.aggregate.clone()).await;
-                }
-            );
+            tokio::spawn(async move {
+                delay_for(TimeDuration::from_secs(aggr_throttle as u64)).await;
+                store(&dbpool, &channel.id, aggregate).await;
+            });
         } else {
-            store(&app.pool, &channel.id, self.aggregate.clone()).await;
+            store(&app.pool, &channel.id, aggregate).await;
         }
 
         Ok(())
     }
-}
+}
diff --git a/sentry/src/event_reducer.rs b/sentry/src/event_reducer.rs
@@ -1,6 +1,5 @@
 use primitives::sentry::{AggregateEvents, Event, EventAggregate};
-use primitives::{Channel};
-
+use primitives::Channel;
 
 // @TODO: Remove attribute once we use this function!
 #[allow(dead_code)]
diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs
@@ -3,6 +3,7 @@
 
 use crate::chain::chain;
 use crate::db::DbPool;
+use crate::event_aggregator::EventAggregator;
 use crate::middleware::auth;
 use crate::middleware::channel::channel_load;
 use crate::middleware::cors::{cors, Cors};
@@ -18,11 +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, last_approved};
+use routes::channel::{channel_list, create_channel, insert_events, last_approved};
 use slog::{error, Logger};
 use std::collections::HashMap;
-use crate::event_aggregator::EventAggregator;
-
 
 pub mod middleware {
     pub mod auth;
@@ -41,9 +40,8 @@ pub mod routes {
 pub mod access;
 mod chain;
 pub mod db;
-pub mod event_reducer;
 pub mod event_aggregator;
-
+pub mod event_reducer;
 
 lazy_static! {
     static ref CHANNEL_GET_BY_ID: Regex =
@@ -269,9 +267,23 @@ async fn channels_router<A: Adapter + 'static>(
         let req = chain(req, app, vec![Box::new(channel_load)]).await?;
 
         list_channel_event_aggregates(req, app).await
+    } else if let (Some(caps), &Method::POST) =
+        (CREATE_EVENTS_BY_CHANNEL_ID.captures(&path), method)
+    {
+        if req.extensions().get::<Session>().is_none() {
+            return Err(ResponseError::Unauthorized);
+        }
+
+        let param = RouteParams(vec![caps
+            .get(1)
+            .map_or("".to_string(), |m| m.as_str().to_string())]);
+
+        req.extensions_mut().insert(param);
+
+        let req = chain(req, app, vec![Box::new(channel_load)]).await?;
+
+        insert_events(req, app).await
     } else {
-    // else if let (Some(caps), &Method::POST) = (CREATE_EVENTS_BY_CHANNEL_ID.captures(&path) ,method) {
-    //} else {
         Err(ResponseError::NotFound)
     }
 }
diff --git a/sentry/src/routes/channel.rs b/sentry/src/routes/channel.rs
@@ -8,7 +8,7 @@ use crate::Session;
 use hex::FromHex;
 use hyper::{Body, Request, Response};
 use primitives::adapter::Adapter;
-use primitives::sentry::{SuccessResponse, Event};
+use primitives::sentry::{Event, SuccessResponse};
 use primitives::{Channel, ChannelId};
 use slog::error;
 
@@ -103,9 +103,9 @@ pub async fn last_approved<A: Adapter>(
         .unwrap())
 }
 
-pub async fn events<A: Adapter>(
+pub async fn insert_events<A: Adapter + 'static>(
     req: Request<Body>,
-    app: &'static Application<A>,
+    app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
     let session = req
         .extensions()
@@ -122,14 +122,14 @@ pub async fn events<A: Adapter>(
     let body = hyper::body::to_bytes(into_body).await?;
     let events = serde_json::from_slice::<Vec<Event>>(&body)?;
 
-    
-    app.event_aggregator.record(app, channel, session, &events.as_slice()).await?;
+    app.event_aggregator
+        .record(app, channel, session, &events.as_slice())
+        .await?;
 
     Ok(Response::builder()
         .header("Content-type", "application/json")
-        .body(serde_json::to_string(&SuccessResponse { success: true})?.into())
+        .body(serde_json::to_string(&SuccessResponse { success: true })?.into())
         .unwrap())
-
 }
 
 mod channel_list {
