Merge pull request #4 from AdExNetwork/aip-6-additions

Separate Session and Auth, clean iterators for payouts

Author: samparsky

sentry/src/access.rs

@@ -2,7 +2,7 @@ use chrono::Utc;
 use futures::future::try_join_all;
 use redis::aio::MultiplexedConnection;
 
-use crate::{AuthSession, Session};
+use crate::{Auth, Session};
 use primitives::event_submission::{RateLimit, Rule};
 use primitives::sentry::Event;
 use primitives::Channel;
@@ -38,8 +38,8 @@ impl fmt::Display for Error {
 // @TODO: Make pub(crate)
 pub async fn check_access(
     redis: &MultiplexedConnection,
-    auth_session: Option<&AuthSession>,
     session: &Session,
+    auth: Option<&Auth>,
     rate_limit: &RateLimit,
     channel: &Channel,
     events: &[Event],
@@ -61,7 +61,7 @@ pub async fn check_access(
         return Err(Error::ChannelIsExpired);
     }
 
-    let (is_creator, auth_uid) = match auth_session.as_ref() {
+    let (is_creator, auth_uid) = match auth {
         Some(auth) => (auth.uid == channel.creator, auth.uid.to_string()),
         None => (false, Default::default()),
     };
@@ -258,19 +258,18 @@ mod test {
     async fn session_uid_rate_limit() {
         let (config, redis) = setup().await;
 
+        let auth = Auth {
+            era: 0,
+            uid: IDS["follower"].clone(),
+        };
+
         let session = Session {
             ip: Default::default(),
             referrer_header: None,
             country: None,
             os: None,
         };
 
-        let auth = AuthSession {
-            era: 0,
-            uid: IDS["follower"].clone(),
-            session: session.clone(),
-        };
-
         let rule = Rule {
             uids: None,
             rate_limit: Some(RateLimit {
@@ -283,8 +282,8 @@ mod test {
 
         let response = check_access(
             &redis,
-            Some(&auth),
             &session,
+            Some(&auth),
             &config.ip_rate_limit,
             &channel,
             &events,
@@ -294,8 +293,8 @@ mod test {
 
         let err_response = check_access(
             &&redis,
-            Some(&auth),
             &session,
+            Some(&auth),
             &config.ip_rate_limit,
             &channel,
             &events,
@@ -313,19 +312,18 @@ mod test {
     async fn ip_rate_limit() {
         let (config, redis) = setup().await;
 
+        let auth = Auth {
+            era: 0,
+            uid: IDS["follower"].clone(),
+        };
+
         let session = Session {
             ip: Default::default(),
             referrer_header: None,
             country: None,
             os: None,
         };
 
-        let auth = AuthSession {
-            era: 0,
-            uid: IDS["follower"].clone(),
-            session: session.clone(),
-        };
-
         let rule = Rule {
             uids: None,
             rate_limit: Some(RateLimit {
@@ -337,8 +335,8 @@ mod test {
 
         let err_response = check_access(
             &redis,
-            Some(&auth),
             &session,
+            Some(&auth),
             &config.ip_rate_limit,
             &channel,
             &get_impression_events(2),
@@ -354,8 +352,8 @@ mod test {
 
         let response = check_access(
             &redis,
-            Some(&auth),
             &session,
+            Some(&auth),
             &config.ip_rate_limit,
             &channel,
             &get_impression_events(1),

sentry/src/db/analytics.rs

@@ -1,6 +1,6 @@
 use crate::db::DbPool;
 use crate::epoch;
-use crate::AuthSession;
+use crate::Auth;
 use bb8::RunError;
 use bb8_postgres::tokio_postgres::types::ToSql;
 use chrono::Utc;
@@ -13,9 +13,9 @@ use std::collections::HashMap;
 use std::error::Error;
 
 pub enum AnalyticsType {
-    Advertiser { auth: AuthSession },
+    Advertiser { auth: Auth },
     Global,
-    Publisher { auth: AuthSession },
+    Publisher { auth: Auth },
 }
 
 pub async fn advertiser_channel_ids(

sentry/src/event_aggregator.rs

@@ -1,13 +1,13 @@
 use crate::access::check_access;
 use crate::access::Error as AccessError;
-use crate::analytics_recorder;
 use crate::db::event_aggregate::insert_event_aggregate;
 use crate::db::get_channel_by_id;
 use crate::db::DbPool;
 use crate::event_reducer;
 use crate::Application;
 use crate::ResponseError;
-use crate::{AuthSession, Session};
+use crate::Session;
+use crate::{analytics_recorder, Auth};
 use async_std::sync::RwLock;
 use chrono::Utc;
 use lazy_static::lazy_static;
@@ -68,8 +68,8 @@ impl EventAggregator {
         &self,
         app: &'a Application<A>,
         channel_id: &ChannelId,
-        auth_session: Option<&AuthSession>,
         session: &Session,
+        auth: Option<&Auth>,
         events: &'a [Event],
     ) -> Result<(), ResponseError> {
         let recorder = self.recorder.clone();
@@ -127,8 +127,8 @@ impl EventAggregator {
 
         check_access(
             &app.redis,
-            auth_session,
             session,
+            auth,
             &app.config.ip_rate_limit,
             &record.channel,
             events,

sentry/src/lib.rs

@@ -453,14 +453,7 @@ pub struct Session {
 }
 
 #[derive(Debug, Clone)]
-pub struct AuthSession {
+pub struct Auth {
     pub era: i64,
     pub uid: ValidatorId,
-    pub session: Session,
-}
-
-impl AsRef<Session> for AuthSession {
-    fn as_ref(&self) -> &Session {
-        &self.session
-    }
 }

sentry/src/middleware/auth.rs

@@ -6,7 +6,7 @@ use redis::aio::MultiplexedConnection;
 
 use primitives::adapter::{Adapter, Session as AdapterSession};
 
-use crate::{AuthSession, Session};
+use crate::{Auth, Session};
 
 /// Check `Authorization` header for `Bearer` scheme with `Adapter::session_from_token`.
 /// If the `Adapter` fails to create an `AdapterSession`, `ResponseError::BadRequest` will be returned.
@@ -27,8 +27,7 @@ pub(crate) async fn for_request(
         referrer_header: referrer,
         os: None,
     };
-
-    req.extensions_mut().insert(session.clone());
+    req.extensions_mut().insert(session);
 
     let authorization = req.headers().get(AUTHORIZATION);
 
@@ -73,10 +72,9 @@ pub(crate) async fn for_request(
             }
         };
 
-        let auth = AuthSession {
+        let auth = Auth {
             era: adapter_session.era,
             uid: adapter_session.uid,
-            session,
         };
 
         req.extensions_mut().insert(auth);
@@ -134,7 +132,7 @@ mod test {
             .expect("Handling the Request shouldn't have failed");
 
         assert!(
-            no_auth.extensions().get::<AuthSession>().is_none(),
+            no_auth.extensions().get::<Auth>().is_none(),
             "There shouldn't be a Session in the extensions"
         );
 
@@ -147,7 +145,7 @@ mod test {
             .await
             .expect("Handling the Request shouldn't have failed");
         assert!(
-            incorrect_auth.extensions().get::<AuthSession>().is_none(),
+            incorrect_auth.extensions().get::<Auth>().is_none(),
             "There shouldn't be a Session in the extensions"
         );
 
@@ -178,12 +176,18 @@ mod test {
         let altered_request = for_request(req, &dummy_adapter, redis)
             .await
             .expect("Valid requests should succeed");
-        let session = altered_request
+
+        let auth = altered_request
             .extensions()
-            .get::<AuthSession>()
+            .get::<Auth>()
             .expect("There should be a Session set inside the request");
 
-        assert_eq!(IDS["leader"], session.uid);
-        assert!(session.session.ip.is_none());
+        assert_eq!(IDS["leader"], auth.uid);
+
+        let session = altered_request
+            .extensions()
+            .get::<Session>()
+            .expect("There should be a Session set inside the request");
+        assert!(session.ip.is_none());
     }
 }

sentry/src/payout.rs

@@ -34,24 +34,19 @@ fn payout(
     min_price: BigNum,
     publisher: &ValidatorId,
 ) -> BigNum {
-    let matching_rules: Vec<&PriceMultiplicationRules> = rules
-        .iter()
-        .filter(|&rule| match_rule(rule, &event, &session, &publisher))
-        .collect();
-    let fixed_amount_rule = matching_rules
-        .iter()
-        .find(|&rule| rule.amount.is_some())
-        .map(|&rule| rule.amount.as_ref().expect("should have value"));
+    let fixed_amount_rule = rules.iter().find_map(|rule| {
+        match (match_rule(rule, &event, &session, &publisher), &rule.amount) {
+            (true, Some(amount)) => Some(amount.clone()),
+            _ => None,
+        }
+    });
 
     let price_by_rules = match fixed_amount_rule {
-        Some(amount) => amount.clone(),
+        Some(amount) => amount,
         None => {
             let exponent: f64 = 10.0;
-            let multiplier = rules
-                .iter()
-                .filter(|&rule| rule.multiplier.is_some())
-                .map(|rule| rule.multiplier.expect("should have value"))
-                .fold(1.0, |result, i| result * i);
+            // @TODO: Know-issue - `multiplier` and `value` can overflow, should return Error instead
+            let multiplier: f64 = rules.iter().filter_map(|rule| rule.multiplier).product();
             let value: u64 = (multiplier * exponent.powi(18)) as u64;
             let result = min_price * BigNum::from(value);
 
@@ -170,7 +165,6 @@ mod tests {
         cases.iter().for_each(|case| {
             let (event, expected_result, message) = case;
             let payout = get_payout(&channel, &event, &session);
-            // println!("payout {:?}", payout.to_f64());
             assert!(&payout == expected_result, message.clone());
         })
     }
@@ -224,8 +218,7 @@ mod tests {
             os: None,
         };
 
-        cases.iter().for_each(|case| {
-            let (event, expected_result, message) = case;
+        cases.iter().for_each(|(event, expected_result, message)| {
             let payout = get_payout(&channel, &event, &session);
             assert!(&payout == expected_result, message.clone());
         })

sentry/src/routes/analytics.rs

@@ -3,7 +3,7 @@ use crate::db::analytics::{
 };
 use crate::success_response;
 use crate::Application;
-use crate::AuthSession;
+use crate::Auth;
 use crate::ResponseError;
 use crate::RouteParams;
 use hyper::{Body, Request, Response};
@@ -17,11 +17,13 @@ pub async fn publisher_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
-    let sess = req.extensions().get::<AuthSession>();
+    let auth = req
+        .extensions()
+        .get::<Auth>()
+        .ok_or(ResponseError::Unauthorized)?
+        .clone();
 
-    let analytics_type = AnalyticsType::Publisher {
-        auth: sess.cloned().ok_or(ResponseError::Unauthorized)?,
-    };
+    let analytics_type = AnalyticsType::Publisher { auth };
 
     process_analytics(req, app, analytics_type)
         .await
@@ -65,7 +67,7 @@ pub async fn advertiser_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
-    let sess = req.extensions().get::<AuthSession>();
+    let sess = req.extensions().get::<Auth>();
     let analytics_type = AnalyticsType::Advertiser {
         auth: sess.ok_or(ResponseError::Unauthorized)?.to_owned(),
     };
@@ -110,10 +112,7 @@ pub async fn advanced_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
-    let auth = req
-        .extensions()
-        .get::<AuthSession>()
-        .expect("auth is required");
+    let auth = req.extensions().get::<Auth>().expect("auth is required");
     let advertiser_channels = advertiser_channel_ids(&app.pool, &auth.uid).await?;
 
     let query = serde_urlencoded::from_str::<AnalyticsQuery>(&req.uri().query().unwrap_or(""))?;

sentry/src/routes/channel.rs

@@ -5,7 +5,7 @@ use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
 use crate::RouteParams;
-use crate::{AuthSession, Session};
+use crate::{Auth, Session};
 use bb8::RunError;
 use bb8_postgres::tokio_postgres::error;
 use futures::future::try_join_all;
@@ -184,7 +184,7 @@ pub async fn insert_events<A: Adapter + 'static>(
 ) -> Result<Response<Body>, ResponseError> {
     let (req_head, req_body) = req.into_parts();
 
-    let auth_session = req_head.extensions.get::<AuthSession>();
+    let auth = req_head.extensions.get::<Auth>();
     let session = req_head
         .extensions
         .get::<Session>()
@@ -205,7 +205,7 @@ pub async fn insert_events<A: Adapter + 'static>(
         .ok_or_else(|| ResponseError::BadRequest("invalid request".to_string()))?;
 
     app.event_aggregator
-        .record(app, &channel_id, auth_session, session, &events)
+        .record(app, &channel_id, session, auth, &events)
         .await?;
 
     Ok(Response::builder()
@@ -220,7 +220,7 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
 ) -> Result<Response<Body>, ResponseError> {
     let session = req
         .extensions()
-        .get::<AuthSession>()
+        .get::<Auth>()
         .expect("auth request session")
         .to_owned();