sentry - Session now includes the Auth, instead of the opposite + clean iterators for payouts

elpiel · elpiel · commit 3e786e24f213 · 2020-04-03T14:40:24.000+03:00
diff --git a/sentry/src/access.rs b/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::Session;
 use primitives::event_submission::{RateLimit, Rule};
 use primitives::sentry::Event;
 use primitives::Channel;
@@ -38,7 +38,6 @@ impl fmt::Display for Error {
 // @TODO: Make pub(crate)
 pub async fn check_access(
     redis: &MultiplexedConnection,
-    auth_session: Option<&AuthSession>,
     session: &Session,
     rate_limit: &RateLimit,
     channel: &Channel,
@@ -61,7 +60,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 &session.auth {
         Some(auth) => (auth.uid == channel.creator, auth.uid.to_string()),
         None => (false, Default::default()),
     };
@@ -258,19 +257,19 @@ 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 {
+            auth: Some(auth),
             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 {
@@ -281,26 +280,12 @@ mod test {
         let events = get_impression_events(2);
         let channel = get_channel(rule);
 
-        let response = check_access(
-            &redis,
-            Some(&auth),
-            &session,
-            &config.ip_rate_limit,
-            &channel,
-            &events,
-        )
-        .await;
+        let response =
+            check_access(&redis, &session, &config.ip_rate_limit, &channel, &events).await;
         assert_eq!(Ok(()), response);
 
-        let err_response = check_access(
-            &&redis,
-            Some(&auth),
-            &session,
-            &config.ip_rate_limit,
-            &channel,
-            &events,
-        )
-        .await;
+        let err_response =
+            check_access(&&redis, &session, &config.ip_rate_limit, &channel, &events).await;
         assert_eq!(
             Err(Error::RulesError(
                 "rateLimit: too many requests".to_string()
@@ -313,17 +298,17 @@ 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(),
+            auth: Some(auth),
         };
 
         let rule = Rule {
@@ -337,7 +322,6 @@ mod test {
 
         let err_response = check_access(
             &redis,
-            Some(&auth),
             &session,
             &config.ip_rate_limit,
             &channel,
@@ -354,7 +338,6 @@ mod test {
 
         let response = check_access(
             &redis,
-            Some(&auth),
             &session,
             &config.ip_rate_limit,
             &channel,
diff --git a/sentry/src/db/analytics.rs b/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(
diff --git a/sentry/src/event_aggregator.rs b/sentry/src/event_aggregator.rs
@@ -7,7 +7,7 @@ use crate::db::DbPool;
 use crate::event_reducer;
 use crate::Application;
 use crate::ResponseError;
-use crate::{AuthSession, Session};
+use crate::Session;
 use async_std::sync::RwLock;
 use chrono::Utc;
 use lazy_static::lazy_static;
@@ -68,7 +68,6 @@ impl EventAggregator {
         &self,
         app: &'a Application<A>,
         channel_id: &ChannelId,
-        auth_session: Option<&AuthSession>,
         session: &Session,
         events: &'a [Event],
     ) -> Result<(), ResponseError> {
@@ -127,7 +126,6 @@ impl EventAggregator {
 
         check_access(
             &app.redis,
-            auth_session,
             session,
             &app.config.ip_rate_limit,
             &record.channel,
diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs
@@ -446,21 +446,15 @@ pub fn epoch() -> f64 {
 // @TODO: Make pub(crate)
 #[derive(Debug, Clone)]
 pub struct Session {
+    pub auth: Option<Auth>,
     pub ip: Option<String>,
     pub country: Option<String>,
     pub referrer_header: Option<String>,
     pub os: Option<String>,
 }
 
 #[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
-    }
 }
diff --git a/sentry/src/middleware/auth.rs b/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.
@@ -21,15 +21,6 @@ pub(crate) async fn for_request(
         .map(|hv| hv.to_str().ok().map(ToString::to_string))
         .flatten();
 
-    let session = Session {
-        ip: get_request_ip(&req),
-        country: None,
-        referrer_header: referrer,
-        os: None,
-    };
-
-    req.extensions_mut().insert(session.clone());
-
     let authorization = req.headers().get(AUTHORIZATION);
 
     let prefix = "Bearer ";
@@ -73,13 +64,20 @@ 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);
+        let session = Session {
+            ip: get_request_ip(&req),
+            country: None,
+            referrer_header: referrer,
+            os: None,
+            auth: Some(auth),
+        };
+
+        req.extensions_mut().insert(session);
     }
 
     Ok(req)
@@ -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"
         );
 
@@ -180,7 +178,7 @@ mod test {
             .expect("Valid requests should succeed");
         let session = altered_request
             .extensions()
-            .get::<AuthSession>()
+            .get::<Auth>()
             .expect("There should be a Session set inside the request");
 
         assert_eq!(IDS["leader"], session.uid);
diff --git a/sentry/src/payout.rs b/sentry/src/payout.rs
@@ -34,24 +34,18 @@ 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)) => rule.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);
+            let multiplier = rules.iter().filter_map(|&rule| rule.multiplier).product();
             let value: u64 = (multiplier * exponent.powi(18)) as u64;
             let result = min_price * BigNum::from(value);
 
diff --git a/sentry/src/routes/analytics.rs b/sentry/src/routes/analytics.rs
@@ -1,11 +1,11 @@
 use crate::db::analytics::{
     advertiser_channel_ids, get_advanced_reports, get_analytics, AnalyticsType,
 };
-use crate::success_response;
 use crate::Application;
-use crate::AuthSession;
+use crate::Auth;
 use crate::ResponseError;
 use crate::RouteParams;
+use crate::{success_response, Session};
 use hyper::{Body, Request, Response};
 use primitives::adapter::Adapter;
 use primitives::analytics::{AnalyticsQuery, AnalyticsResponse};
@@ -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 session = req
+        .extensions()
+        .get::<Session>()
+        .ok_or(ResponseError::Unauthorized)?;
+    let auth = session.auth.ok_or(ResponseError::Unauthorized)?;
 
-    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(""))?;
diff --git a/sentry/src/routes/channel.rs b/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_session = 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, &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();
 
diff --git a/sentry/src/routes/event_aggregate.rs b/sentry/src/routes/event_aggregate.rs
