fix: refactor analytics.rs

Author: samparsky

Cargo.lock

@@ -1,6 +1,8 @@
 use crate::DomainError;
 use serde::{Deserialize, Serialize};
 
+pub const ANALYTICS_QUERY_LIMIT: u32 = 200;
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct AnalyticsResponse {
     time: u32,
@@ -36,7 +38,7 @@ pub struct AnalyticsQuery {
 
 impl AnalyticsQuery {
     pub fn is_valid(&self) -> Result<(), DomainError> {
-        let valid_event_types = ["IMPRESSION"];
+        let valid_event_types = ["IMPRESSION", "CLICK"];
         let valid_metric = ["eventPayouts", "eventCounts"];
         let valid_timeframe = ["year", "month", "week", "day", "hour"];
 
@@ -55,6 +57,11 @@ impl AnalyticsQuery {
                 "invalid timeframe, possible values are: {}",
                 valid_timeframe.join(" ,")
             )))
+        } else if self.limit > ANALYTICS_QUERY_LIMIT {
+            Err(DomainError::InvalidArgument(format!(
+                "invalid limit {}, maximum value 200",
+                self.limit
+            )))
         } else {
             Ok(())
         }

primitives/src/analytics.rs

@@ -35,7 +35,7 @@ lazy_static! {
         auth.insert("user".into(), "x8c9v1b2".into());
         auth.insert("publisher".into(), "testing".into());
         auth.insert("publisher2".into(), "testing2".into());
-        auth.insert("creator".into(), "awesomeCreator".into());
+        auth.insert("creator".into(), "0x033ed90e0fec3f3ea1c9b005c724d704501e0196".into());
         auth.insert("tester".into(), "AUTH_awesomeTester".into());
 
         auth

primitives/src/util/tests/prep_db.rs

@@ -1 +1 @@
-stable
+nightly

rust-toolchain

@@ -4,45 +4,51 @@ use bb8::RunError;
 use chrono::Utc;
 use primitives::analytics::{AnalyticsQuery, AnalyticsResponse, ANALYTICS_QUERY_LIMIT};
 
+pub enum AnalyticsType {
+    Advertiser {
+        session: Session,
+        channel: Option<String>,
+    },
+    Global,
+    Publisher {
+        session: Session,
+        channel: Option<String>,
+    },
+}
+
 pub async fn get_analytics(
     query: AnalyticsQuery,
-    channel: Option<String>,
-    sess: Option<&Session>,
     pool: &DbPool,
-    is_advertiser: bool,
-    filter_publisher: bool,
+    analytics_type: AnalyticsType,
 ) -> Result<Vec<AnalyticsResponse>, 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 is_advertiser {
-        match (channel, sess) {
-            (Some(id), _) => where_clauses.push(format!("channel_id IN ({})", id)),
-            (None, Some(session)) => where_clauses.push(format!(
-                "channel_id IN (SELECT id FROM channels WHERE creator = {})",
-                session.uid
-            )),
-            _ => {}
-        };
-    } else if let Some(id) = channel {
-        where_clauses.push(format!("channel_id = {}", id));
-    }
+    let select_query = match analytics_type {
+        AnalyticsType::Advertiser { session, channel } => {
+            if let Some(id) = channel {
+                where_clauses.push(format!("channel_id = {}", id));
+            } else {
+                where_clauses.push(format!(
+                    "channel_id IN (SELECT id FROM channels WHERE creator = {})",
+                    session.uid
+                ));
+            }
 
-    let select_query = match (filter_publisher, sess) {
-        (true, Some(session)) => {
             where_clauses.push(format!(
-                "events->'{}'->'{}'->'{}' IS NOT NULL",
-                query.event_type, query.metric, session.uid
+                "events->'{}'->'{}' IS NOT NULL",
+                query.event_type, query.metric
             ));
+
             format!(
-                "select 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
+                "select 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
             )
         }
-        _ => {
+        AnalyticsType::Global => {
             where_clauses.push(format!(
                 "events->'{}'->'{}' IS NOT NULL",
                 query.event_type, query.metric
@@ -52,6 +58,21 @@ pub async fn get_analytics(
                 interval, query.event_type, query.metric
             )
         }
+        AnalyticsType::Publisher { session, channel } => {
+            if let Some(id) = channel {
+                where_clauses.push(format!("channel_id = {}", id));
+            }
+
+            where_clauses.push(format!(
+                "events->'{}'->'{}'->'{}' IS NOT NULL",
+                query.event_type, query.metric, session.uid
+            ));
+
+            format!(
+                "select 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
+            )
+        }
     };
 
     let sql_query = format!(

sentry/src/db/analytics.rs

@@ -182,7 +182,6 @@ async fn analytics_router<A: Adapter>(
 
                 let req = chain(req, app, vec![channel_load]).await?;
                 analytics(req, app).await
-
             } else if let Some(caps) = ADVERTISER_ANALYTICS_BY_CHANNEL_ID.captures(route) {
                 let param = RouteParams(vec![caps
                     .get(1)
@@ -291,7 +290,10 @@ pub fn map_response_error(error: ResponseError) -> Response<Body> {
     match error {
         ResponseError::NotFound => not_found(),
         ResponseError::BadRequest(e) => bad_response(e, StatusCode::BAD_REQUEST),
-        ResponseError::UnAuthorized => bad_response("invalid authorization".to_string(), StatusCode::UNAUTHORIZED),
+        ResponseError::UnAuthorized => bad_response(
+            "invalid authorization".to_string(),
+            StatusCode::UNAUTHORIZED,
+        ),
     }
 }
 
@@ -314,11 +316,10 @@ pub fn bad_response(response_body: String, status_code: StatusCode) -> Response<
         .insert("Content-type", "application/json".parse().unwrap());
 
     *response.status_mut() = status_code;
-    
+
     response
 }
 
-
 pub fn success_response(response_body: String) -> Response<Body> {
     let body = Body::from(response_body);
 

sentry/src/lib.rs

@@ -53,7 +53,7 @@ pub(crate) async fn for_request(
                     .arg(serde_json::to_string(&adapter_session)?)
                     .query_async(&mut redis.clone())
                     .await?;
-                
+
                 adapter_session
             }
         };

sentry/src/middleware/auth.rs

@@ -1,4 +1,4 @@
-use crate::db::analytics::get_analytics;
+use crate::db::analytics::{get_analytics, AnalyticsType};
 use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
@@ -14,7 +14,17 @@ pub async fn publisher_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
-    process_analytics(req, app, false, true)
+    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)
         .await
         .map(success_response)
 }
@@ -38,8 +48,15 @@ pub async fn analytics<A: Adapter>(
                 Some(_) => 600,
                 None => 300,
             };
-            let response = process_analytics(req, app, false, false).await?;
-            cache(&redis.clone(), request_uri, &response, cache_timeframe, &app.logger).await;
+            let response = process_analytics(req, app, AnalyticsType::Global).await?;
+            cache(
+                &redis.clone(),
+                request_uri,
+                &response,
+                cache_timeframe,
+                &app.logger,
+            )
+            .await;
             Ok(success_response(response))
         }
     }
@@ -49,43 +66,44 @@ pub async fn advertiser_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
-    process_analytics(req, app, true, false)
+    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)
         .await
         .map(success_response)
 }
 
 pub async fn process_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
-    is_advertiser: bool,
-    filter_publisher: bool,
+    analytics_type: AnalyticsType,
 ) -> Result<String, ResponseError> {
     let query = serde_urlencoded::from_str::<AnalyticsQuery>(&req.uri().query().unwrap_or(""))?;
     query
         .is_valid()
         .map_err(|e| ResponseError::BadRequest(e.to_string()))?;
 
-    let sess = req.extensions().get::<Session>();
-    let channels = match req.extensions().get::<RouteParams>() {
-        Some(param) => param.get(0),
-        None => None,
-    };
-
-    let result = get_analytics(
-        query,
-        channels,
-        sess,
-        &app.pool,
-        is_advertiser,
-        filter_publisher,
-    )
-    .await?;
+    let result = get_analytics(query, &app.pool, analytics_type).await?;
 
     serde_json::to_string(&result)
         .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))
 }
 
-async fn cache(redis: &MultiplexedConnection, key: String, value: &str, timeframe: i32, logger: &Logger) {
+async fn cache(
+    redis: &MultiplexedConnection,
+    key: String,
+    value: &str,
+    timeframe: i32,
+    logger: &Logger,
+) {
     if let Err(err) = redis::cmd("SETEX")
         .arg(&key)
         .arg(timeframe)