fix: /:id/events POST route

Author: samparsky

sentry/src/access.rs

@@ -6,25 +6,39 @@ 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)]
-pub enum Error {
+pub enum AccessError {
     OnlyCreatorCanCloseChannel,
     ChannelIsExpired,
     ChannelIsInWithdrawPeriod,
     RulesError(String),
 }
 
+impl Error for AccessError {}
+
+impl fmt::Display for AccessError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            AccessError::OnlyCreatorCanCloseChannel => write!(f, "only creator can create channel"),
+            AccessError::ChannelIsExpired => write!(f, "channel has expired"),
+            AccessError::ChannelIsInWithdrawPeriod => write!(f, "channel is in withdraw period"),
+            AccessError::RulesError(error) => write!(f, "{}", error),
+        }
+    }
+}
+
 // @TODO: Make pub(crate)
 pub async fn check_access(
     redis: &MultiplexedConnection,
     session: &Session,
     rate_limit: &RateLimit,
     channel: &Channel,
     events: &[Event],
-) -> Result<(), Error> {
+) -> Result<(), AccessError> {
     let is_close_event = |e: &Event| match e {
         Event::Close => true,
         _ => false,
@@ -33,7 +47,7 @@ pub async fn check_access(
     let is_in_withdraw_period = current_time > channel.spec.withdraw_period_start;
 
     if current_time > channel.valid_until {
-        return Err(Error::ChannelIsExpired);
+        return Err(AccessError::ChannelIsExpired);
     }
 
     // We're only sending a CLOSE
@@ -46,11 +60,11 @@ pub async fn check_access(
 
     // Only the creator can send a CLOSE
     if session.uid != channel.creator && events.iter().any(is_close_event) {
-        return Err(Error::OnlyCreatorCanCloseChannel);
+        return Err(AccessError::OnlyCreatorCanCloseChannel);
     }
 
     if is_in_withdraw_period {
-        return Err(Error::ChannelIsInWithdrawPeriod);
+        return Err(AccessError::ChannelIsInWithdrawPeriod);
     }
 
     let default_rules = [
@@ -92,7 +106,7 @@ pub async fn check_access(
     );
 
     if let Err(rule_error) = apply_all_rules.await {
-        Err(Error::RulesError(rule_error))
+        Err(AccessError::RulesError(rule_error))
     } else {
         Ok(())
     }
@@ -225,7 +239,7 @@ mod test {
         let err_response =
             check_access(&redis, &session, &config.ip_rate_limit, &channel, &events).await;
         assert_eq!(
-            Err(Error::RulesError(
+            Err(AccessError::RulesError(
                 "rateLimit: too many requests".to_string()
             )),
             err_response
@@ -261,7 +275,7 @@ mod test {
         .await;
 
         assert_eq!(
-            Err(Error::RulesError(
+            Err(AccessError::RulesError(
                 "rateLimit: only allows 1 event".to_string()
             )),
             err_response

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;
-mod event_aggregate;
 mod validator_message;
 
 pub use self::channel::*;

sentry/src/db/event_aggregate.rs

@@ -1,7 +1,7 @@
 use crate::db::DbPool;
 use bb8::RunError;
 use bb8_postgres::tokio_postgres::types::ToSql;
-use chrono::{DateTime, Utc, Date};
+use chrono::{DateTime, Utc};
 use primitives::sentry::EventAggregate;
 use primitives::{ValidatorId, ChannelId};
 
@@ -58,35 +58,44 @@ pub async fn insert_event_aggregate(
     channel_id: &ChannelId,
     event: &EventAggregate
 ) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>>  {
-
-    let mut inserts : Vec<String> = Vec::new();
     let mut values = Vec::new();
     let mut index = 0;
     let id = channel_id.to_string();
 
-    for (event_type, aggr) in event.events {
-        if let Some(event_counts) = aggr.event_counts {
+    let mut data: Vec<String> = Vec::new();
+
+    for (event_type, aggr) in &event.events {
+        if let Some(event_counts) = &aggr.event_counts {
             for (earner, value) in event_counts {
-                let data =  vec![id.clone(), event_type.clone(), earner, value.to_string(), aggr.event_payouts[&earner].to_string()];
-                inserts.extend(data);
+                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]);
                 //
                 // this is a work around for bulk inserts
                 // rust-postgres does not have native support for bulk inserts
                 // so we have to manually build up a query string dynamically based on
                 // how many things we want to insert
-                //
+                // 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));
                 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
-     pool
+    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(&format!("INSERT INTO event_aggregates (channel_id, event_type, earner, event_counts, event_payouts) values {}", values.join(" ,"))).await {
+                    match connection.prepare(&query).await {
                         Ok(stmt) => match connection.execute(&stmt, &inserts.as_slice()).await {
                             Ok(row) => {
                                 let inserted = row == (index / 5);
@@ -98,5 +107,7 @@ pub async fn insert_event_aggregate(
                     }
                 }
             }
-        ).await
+        ).await?;
+
+    Ok(result)
 }

sentry/src/event_aggregator.rs

@@ -3,67 +3,99 @@ use primitives::{ChannelId, Channel};
 use primitives::adapter::Adapter;
 use primitives::sentry::{EventAggregate, Event};
 use std::collections::HashMap;
-use futures::future::BoxFuture;
+use crate::db::event_aggregate::insert_event_aggregate;
 use crate::Session;
 use crate::access::check_access;
 use crate::Application;
 use chrono::{Utc, Duration};
-use async_std::stream;
+use crate::db::DbPool;
+use tokio::time::{delay_for};
+use std::time::Duration as TimeDuration;
+use crate::ResponseError;
 
 // use futures::
 use async_std::sync::RwLock;
 use std::sync::{Arc};
 
+#[derive(Default, Clone)]
 pub struct EventAggregator {
     // recorder: HashMap<ChannelId, FnMut(&Session, &str) -> BoxFuture>,
-    aggregate: Arc<RwLock<HashMap<ChannelId, EventAggregate>>>
+    aggregate: Arc<RwLock<HashMap<String, EventAggregate>>>
 }
 
-fn persist(aggr_throttle: i64, channel_id: ChannelId, aggr: Arc<RwLock<HashMap<ChannelId, EventAggregate>>>) {
-    let mut interval = stream::interval(Duration::from_secs(aggr_throttle));
-    while let Some(_) = interval.next().await {
-//        loop through the keys and persist them in the
-//        database
+pub fn new_aggr(channel_id: &ChannelId) -> EventAggregate {
+    EventAggregate {
+        channel_id: channel_id.to_owned(),
+        created: Utc::now(),
+        events: HashMap::new(),
+    }
+}
+
+async fn store(db: &DbPool, channel_id: &ChannelId, aggr: Arc<RwLock<HashMap<String, EventAggregate>>>) {
+    let recorder = aggr.write().await;
+    let ev_aggr: Option<&EventAggregate> = recorder.get(&channel_id.to_string());
+    if let Some(data) = ev_aggr {
+        if let Err(e) = insert_event_aggregate(&db, &channel_id, data).await {
+            eprintln!("{}", e);
+        };
     }
 }
 
 impl EventAggregator {
-    pub async fn record<A: Adapter>(
-        &self,
-        app: &Application<A>,
-        channel: &Channel,
-        session: &Session,
-        events: &[Event]
-        )
+    pub async fn record<'a, A: Adapter + 'static>(
+        &'static self,
+        app: &'static Application<A>,
+        channel: Channel,
+        session: Session,
+        events: &'a [Event],
+        ) -> Result<(), ResponseError>
     {
         // eventAggrCol
         // channelsCol
         // try getting aggr if none create and store new aggr
-    //     redis: &MultiplexedConnection,
-    // session: &Session,
-    // rate_limit: &RateLimit,
-    // channel: &Channel,
-    // events: &[Event]
-        let has_access = check_access(&app.redis, session, &app.config.ip_rate_limit, channel, events).await;
-        if has_access.is_err() {
-            // return the error
+        // redis: &MultiplexedConnection,
+        // session: &Session,
+        // rate_limit: &RateLimit,
+        // channel: &Channel,
+        // events: &[Event]
+        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 recorder = self.aggregate.write().await.expect("should acquire lock");
-        let mut aggr: EventAggregate = *recorder.get_mut(&channel.id);
+
+        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")
+        };
+
         // if aggr is none
         // spawn a tokio task for saving to database
-        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;
+
+        // drop write access to RwLock and mut access to aggr
+        // we don't need it anymore
+        drop(recorder);
 
         if app.config.aggr_throttle > 0
             &&
-            Utc::now() < (aggr.created + Duration::seconds(app.config.aggr_throttle as i64))
+            Utc::now() > (created + Duration::seconds(app.config.aggr_throttle as i64))
         {
 
+            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;
+                }
+            );
+        } else {
+            store(&app.pool, &channel.id, self.aggregate.clone()).await;
         }
 
-
-
-
-
+        Ok(())
     }
 }

sentry/src/event_reducer.rs

@@ -1,5 +1,6 @@
 use primitives::sentry::{AggregateEvents, Event, EventAggregate};
-use primitives::{Channel, ValidatorId};
+use primitives::{Channel};
+
 
 // @TODO: Remove attribute once we use this function!
 #[allow(dead_code)]

sentry/src/lib.rs

@@ -21,6 +21,9 @@ use routes::cfg::config;
 use routes::channel::{channel_list, create_channel, last_approved};
 use slog::{error, Logger};
 use std::collections::HashMap;
+use crate::event_aggregator::EventAggregator;
+
+
 pub mod middleware {
     pub mod auth;
     pub mod channel;
@@ -89,6 +92,7 @@ pub struct Application<A: Adapter> {
     pub redis: MultiplexedConnection,
     pub pool: DbPool,
     pub config: Config,
+    pub event_aggregator: EventAggregator,
     __secret: (),
 }
 
@@ -106,6 +110,7 @@ impl<A: Adapter + 'static> Application<A> {
             logger,
             redis,
             pool,
+            event_aggregator: Default::default(),
             __secret: (),
         }
     }

sentry/src/middleware/auth.rs

@@ -75,6 +75,7 @@ fn get_request_ip(req: &Request<Body>) -> Option<String> {
         .get("true-client-ip")
         .or_else(|| req.headers().get("x-forwarded-for"))
         .and_then(|hv| hv.to_str().map(ToString::to_string).ok())
+        .map(|token| token.split(',').collect::<Vec<&str>>()[0].to_string())
 }
 
 #[cfg(test)]