fix: /analytics query, add get_channel_id middleware

samparsky · samparsky · commit da0dda9b23e9 · 2020-02-17T02:38:10.000+01:00
diff --git a/sentry/src/db/analytics.rs b/sentry/src/db/analytics.rs
@@ -3,7 +3,7 @@ use crate::epoch;
 use crate::Session;
 use bb8::RunError;
 use chrono::Utc;
-use primitives::analytics::{AnalyticsQuery, AnalyticsResponse, ANALYTICS_QUERY_LIMIT};
+use primitives::analytics::{AnalyticsQuery, AnalyticsData, ANALYTICS_QUERY_LIMIT};
 use primitives::sentry::{AdvancedAnalyticsResponse, ChannelReport, PublisherReport};
 use primitives::{ChannelId, ValidatorId};
 use redis;
@@ -14,12 +14,10 @@ use std::error::Error;
 pub enum AnalyticsType {
     Advertiser {
         session: Session,
-        channel: Option<String>,
     },
     Global,
     Publisher {
         session: Session,
-        channel: Option<String>,
     },
 }
 
@@ -50,57 +48,85 @@ pub async fn get_analytics(
     pool: &DbPool,
     analytics_type: AnalyticsType,
     segment_by_channel: bool,
-) -> Result<Vec<AnalyticsResponse>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    channel_id: Option<&ChannelId>,
+) -> Result<Vec<AnalyticsData>, RunError<bb8_postgres::tokio_postgres::Error>> {
     let applied_limit = query.limit.min(ANALYTICS_QUERY_LIMIT);
     let (interval, period) = get_time_frame(&query.timeframe);
     let time_limit = Utc::now().timestamp() - period;
 
     let mut where_clauses = vec![format!("created > to_timestamp({})", time_limit)];
+    if let Some(id) = channel_id {
+        where_clauses.push(format!("channel_id = {}", id));
+    }
+
     let mut group_clause = "time".to_string();
     let mut select_clause = match analytics_type {
-        AnalyticsType::Advertiser { session, channel } => {
-            if let Some(id) = channel {
-                where_clauses.push(format!("channel_id = {}", id));
-            } else {
+        AnalyticsType::Advertiser { session } => {
+            if channel_id.is_none() {
                 where_clauses.push(format!(
                     "channel_id IN (SELECT id FROM channels WHERE creator = {})",
                     session.uid
                 ));
             }
 
             where_clauses.push(format!(
-                "events->'{}'->'{}' IS NOT NULL",
-                query.event_type, query.metric
+                "event_type = '{}'",
+                query.event_type
+            ));
+
+            where_clauses.push(format!(
+                "{} IS NOT NULL",
+                query.metric
             ));
 
             format!(
-                "SUM(value::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates, jsonb_each_text(events->'{}'->'{}')", 
-                interval, query.event_type, query.metric
+                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates", 
+                query.metric, interval
             )
         }
-        AnalyticsType::Global => {
+        AnalyticsType::Global  => {
             where_clauses.push(format!(
-                "events->'{}'->'{}' IS NOT NULL",
-                query.event_type, query.metric
+                "event_type = '{}'",
+                query.event_type
             ));
+
+            where_clauses.push(format!(
+                "{} IS NOT NULL",
+                query.metric
+            ));
+
+            where_clauses.push("earner IS NULL".to_string());
+
             format!(
-                "SUM(value::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates, jsonb_each_text(events->'{}'->'{}')", 
-                interval, query.event_type, query.metric
+                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates", 
+                query.metric, interval
             )
         }
-        AnalyticsType::Publisher { session, channel } => {
-            if let Some(id) = channel {
-                where_clauses.push(format!("channel_id = {}", id));
-            }
+        AnalyticsType::Publisher { session } => {
+            where_clauses.push(format!(
+                "event_type = '{}'",
+                query.event_type
+            ));
 
             where_clauses.push(format!(
-                "events->'{}'->'{}'->'{}' IS NOT NULL",
-                query.event_type, query.metric, session.uid
+                "{} IS NOT NULL",
+                query.metric
             ));
 
+            where_clauses.push(format!(
+                "earner = '{}'",
+                session.uid
+            ));
+            
+
+            // where_clauses.push(format!(
+            //     "events->'{}'->'{}'->'{}' IS NOT NULL",
+            //     query.event_type, query.metric, session.uid
+            // ));
+
             format!(
-                "SUM((events->'{}'->'{}'->>'{}')::numeric) as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates", 
-                query.event_type, query.metric, session.uid, interval
+                "SUM(({}::numeric) as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates", 
+                query.metric, interval
             )
         }
     };
@@ -118,13 +144,15 @@ pub async fn get_analytics(
         applied_limit,
     );
 
+    println!("{}", sql_query);
+
     // execute query
     pool.run(move |connection| async move {
         match connection.prepare(&sql_query).await {
             Ok(stmt) => match connection.query(&stmt, &[]).await {
                 Ok(rows) => {
-                    let analytics: Vec<AnalyticsResponse> =
-                        rows.iter().map(AnalyticsResponse::from).collect();
+                    let analytics: Vec<AnalyticsData> =
+                        rows.iter().map(AnalyticsData::from).collect();
                     Ok((analytics, connection))
                 }
                 Err(e) => Err((e, connection)),
diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs
@@ -5,7 +5,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::channel::{channel_load, get_channel_id};
 use crate::middleware::cors::{cors, Cors};
 use crate::routes::channel::channel_status;
 use crate::routes::event_aggregate::list_channel_event_aggregates;
@@ -198,15 +198,23 @@ async fn analytics_router<A: Adapter + 'static>(
                     .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?;
+                let req = chain(req, app, vec![
+                    Box::new(channel_load),
+                    Box::new(get_channel_id)
+                ]).await?;
+
                 analytics(req, app).await
             } else if let Some(caps) = ADVERTISER_ANALYTICS_BY_CHANNEL_ID.captures(route) {
                 let param = RouteParams(vec![caps
                     .get(1)
                     .map_or("".to_string(), |m| m.as_str().to_string())]);
                 req.extensions_mut().insert(param);
 
-                let req = auth_required_middleware(req, app).await?;
+                let req = chain(req, app, vec![
+                    Box::new(auth_required_middleware),
+                    Box::new(get_channel_id)
+                ]).await?;
+
                 advertiser_analytics(req, app).await
             } else {
                 Err(ResponseError::NotFound)
diff --git a/sentry/src/middleware/auth.rs b/sentry/src/middleware/auth.rs
@@ -23,6 +23,7 @@ pub(crate) async fn for_request(
         .and_then(|hv| {
             hv.to_str()
                 .map(|token_str| {
+                    println!("token str {}", token_str );
                     if token_str.starts_with(prefix) {
                         Some(token_str[prefix.len()..].to_string())
                     } else {
diff --git a/sentry/src/middleware/channel.rs b/sentry/src/middleware/channel.rs
@@ -66,3 +66,24 @@ pub fn channel_if_active<'a, A: Adapter + 'static>(
     }
     .boxed()
 }
+
+
+pub fn get_channel_id<'a, A: Adapter + 'static>(
+    mut req: Request<Body>,
+    _: &'a Application<A> 
+) -> BoxFuture<'a, Result<Request<Body>, ResponseError>> {
+    async move {
+        match req.extensions().get::<RouteParams>() {
+                Some(param) => {
+                    let id = param.get(0).expect("should have channel id");
+                    let channel_id = ChannelId::from_hex(id)
+                            .map_err(|_| ResponseError::BadRequest("Invalid Channel Id".to_string()))?;
+                    req.extensions_mut().insert(channel_id);
+
+                    Ok(req)
+                },
+                None => Ok(req),
+        }
+    }
+    .boxed()
+}
diff --git a/sentry/src/routes/analytics.rs b/sentry/src/routes/analytics.rs
@@ -1,29 +1,27 @@
 use crate::db::analytics::{
-    advertiser_channel_ids, get_advanced_reports, get_analytics, AnalyticsType,
+    advertiser_channel_ids, get_advanced_reports, get_analytics, AnalyticsType
 };
 use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
 use crate::RouteParams;
 use crate::Session;
 use hyper::{Body, Request, Response};
+use primitives::ChannelId;
 use primitives::adapter::Adapter;
-use primitives::analytics::AnalyticsQuery;
+use primitives::analytics::{AnalyticsQuery, AnalyticsResponse};
 use redis::aio::MultiplexedConnection;
 use slog::{error, Logger};
 
+
 pub async fn publisher_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
     let sess = req.extensions().get::<Session>();
-    let channel = match req.extensions().get::<RouteParams>() {
-        Some(param) => param.get(0),
-        None => None,
-    };
+
     let analytics_type = AnalyticsType::Publisher {
         session: sess.cloned().ok_or(ResponseError::Unauthorized)?,
-        channel,
     };
 
     process_analytics(req, app, analytics_type)
@@ -46,8 +44,8 @@ pub async fn analytics<A: Adapter>(
         Ok(Some(response)) => Ok(success_response(response)),
         _ => {
             // checks if /:id route param is present
-            let cache_timeframe = match req.extensions().get::<RouteParams>() {
-                Some(_) => 600,
+            let cache_timeframe= match req.extensions().get::<RouteParams>() {
+                Some(param) => 600,
                 None => 300,
             };
             let response = process_analytics(req, app, AnalyticsType::Global).await?;
@@ -69,13 +67,8 @@ pub async fn advertiser_analytics<A: Adapter>(
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
     let sess = req.extensions().get::<Session>();
-    let channel = match req.extensions().get::<RouteParams>() {
-        Some(param) => param.get(0),
-        None => None,
-    };
     let analytics_type = AnalyticsType::Advertiser {
         session: sess.ok_or(ResponseError::Unauthorized)?.to_owned(),
-        channel,
     };
 
     process_analytics(req, app, analytics_type)
@@ -88,20 +81,30 @@ pub async fn process_analytics<A: Adapter>(
     app: &Application<A>,
     analytics_type: AnalyticsType,
 ) -> Result<String, ResponseError> {
-    let query = serde_urlencoded::from_str::<AnalyticsQuery>(&req.uri().query().unwrap_or(""))?;
+    let mut query = serde_urlencoded::from_str::<AnalyticsQuery>(&req.uri().query().unwrap_or(""))?;
     query
         .is_valid()
         .map_err(|e| ResponseError::BadRequest(e.to_string()))?;
+    
+    query.metric_to_column();
+
+    let channel_id = req.extensions().get::<ChannelId>();
 
     let segment_channel = query
         .segment_by_channel
         .clone()
         .map(|_| true)
         .unwrap_or_else(|| false);
+    let limit = query.limit;
+
+    let aggr = get_analytics(query, &app.pool, analytics_type, segment_channel, channel_id).await?;
 
-    let result = get_analytics(query, &app.pool, analytics_type, segment_channel).await?;
+    let response = AnalyticsResponse {
+        limit,
+        aggr
+    };
 
-    serde_json::to_string(&result)
+    serde_json::to_string(&response)
         .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))
 }
 
