Merge pull request #209 from samparsky/analytics-route

Analytics route

Author: elpiel

Cargo.lock

@@ -37,7 +37,7 @@ rand = "^0.6"
 # postgres feature
 postgres-types = { version = "0.1.0-alpha.2", optional = true }
 bytes = { version = "0.5", optional = true }
-tokio-postgres = { version = "0.5.0-alpha.2", optional = true, features = ["with-chrono-0_4", "with-serde_json-1"] }
+tokio-postgres = { version = "0.5.1", optional = true, features = ["with-chrono-0_4", "with-serde_json-1"] }
 # Futures
 futures = "0.3.1"
 # Other

primitives/Cargo.toml

@@ -0,0 +1,85 @@
+use crate::DomainError;
+use serde::{Deserialize, Serialize};
+
+pub const ANALYTICS_QUERY_LIMIT: u32 = 200;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AnalyticsResponse {
+    time: u32,
+    value: String,
+}
+
+#[cfg(feature = "postgres")]
+pub mod postgres {
+    use super::AnalyticsResponse;
+    use tokio_postgres::Row;
+
+    impl From<&Row> for AnalyticsResponse {
+        fn from(row: &Row) -> Self {
+            Self {
+                time: row.get("time"),
+                value: row.get("value"),
+            }
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+pub struct AnalyticsQuery {
+    #[serde(default = "default_limit")]
+    pub limit: u32,
+    #[serde(default = "default_event_type")]
+    pub event_type: String,
+    #[serde(default = "default_metric")]
+    pub metric: String,
+    #[serde(default = "default_timeframe")]
+    pub timeframe: String,
+}
+
+impl AnalyticsQuery {
+    pub fn is_valid(&self) -> Result<(), DomainError> {
+        let valid_event_types = ["IMPRESSION", "CLICK"];
+        let valid_metric = ["eventPayouts", "eventCounts"];
+        let valid_timeframe = ["year", "month", "week", "day", "hour"];
+
+        if !valid_event_types.contains(&self.event_type.as_str()) {
+            Err(DomainError::InvalidArgument(format!(
+                "invalid event_type, possible values are: {}",
+                valid_event_types.join(" ,")
+            )))
+        } else if !valid_metric.contains(&self.metric.as_str()) {
+            Err(DomainError::InvalidArgument(format!(
+                "invalid metric, possible values are: {}",
+                valid_metric.join(" ,")
+            )))
+        } else if !valid_timeframe.contains(&self.timeframe.as_str()) {
+            Err(DomainError::InvalidArgument(format!(
+                "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(())
+        }
+    }
+}
+
+fn default_limit() -> u32 {
+    100
+}
+
+fn default_event_type() -> String {
+    "IMPRESSION".into()
+}
+
+fn default_metric() -> String {
+    "eventCounts".into()
+}
+
+fn default_timeframe() -> String {
+    "hour".into()
+}

primitives/src/analytics.rs

@@ -23,6 +23,7 @@ pub mod util {
 
     pub mod logging;
 }
+pub mod analytics;
 pub mod validator;
 
 pub use self::ad_unit::AdUnit;
@@ -42,7 +43,10 @@ pub enum DomainError {
 
 impl fmt::Display for DomainError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "Domain error",)
+        match self {
+            DomainError::InvalidArgument(err) => write!(f, "{}", err),
+            DomainError::RuleViolation(err) => write!(f, "{}", err),
+        }
     }
 }
 

primitives/src/lib.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

@@ -20,8 +20,8 @@ hyper = { version = "0.13", features = ["stream"] }
 regex = "1"
 # Database
 redis = { version = "0.13.1-alpha.0", features = ["tokio-rt-core"] }
-bb8 = { git = "https://github.com/djc/bb8", branch = "async-await" }
-bb8-postgres = { git = "https://github.com/djc/bb8", branch = "async-await", features = ["with-chrono-0_4", "with-serde_json-1"] }
+bb8 = { git = "https://github.com/khuey/bb8" }
+bb8-postgres = { git = "https://github.com/khuey/bb8", features = ["with-chrono-0_4", "with-serde_json-1"] }
 # Migrations
 migrant_lib = { version = "0.27", features = ["d-postgres"] }
 # Logger

sentry/Cargo.toml

@@ -7,6 +7,7 @@ use std::env;
 
 use lazy_static::lazy_static;
 
+pub mod analytics;
 mod channel;
 mod validator_message;
 

sentry/src/db.rs

@@ -0,0 +1,117 @@
+use crate::db::DbPool;
+use crate::Session;
+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,
+    pool: &DbPool,
+    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)];
+
+    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
+                ));
+            }
+
+            where_clauses.push(format!(
+                "events->'{}'->'{}' IS NOT NULL",
+                query.event_type, query.metric
+            ));
+
+            format!(
+                "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
+            ));
+            format!(
+                "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::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!(
+        "{} WHERE {} GROUP BY time LIMIT {}",
+        select_query,
+        where_clauses.join(" AND "),
+        applied_limit
+    );
+
+    // 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();
+                        Ok((analytics, connection))
+                    }
+                    Err(e) => Err((e, connection)),
+                },
+                Err(e) => Err((e, connection)),
+            }
+        }
+    })
+    .await
+}
+
+fn get_time_frame(timeframe: &str) -> (i64, i64) {
+    let minute = 60 * 1000;
+    let hour = 60 * minute;
+    let day = 24 * hour;
+
+    match timeframe {
+        "year" => (30 * day, 365 * day),
+        "month" => (day, 30 * day),
+        "week" => (6 * hour, 7 * day),
+        "day" => (hour, day),
+        "hour" => (minute, hour),
+        _ => (hour, day),
+    }
+}