add: impl /analytics/advanced route

Author: samparsky

primitives/src/channel.rs

@@ -285,6 +285,12 @@ pub mod postgres {
         accepts!(TEXT, VARCHAR);
     }
 
+    impl From<&Row> for ChannelId {
+        fn from(row: &Row) -> Self {
+            row.get("id")
+        }
+    }
+
     impl ToSql for ChannelId {
         fn to_sql(
             &self,

sentry/src/analytics_recorder.rs

@@ -5,7 +5,6 @@ use primitives::sentry::{ChannelReport, PublisherReport};
 use primitives::{BigNum, Channel};
 use redis;
 use redis::aio::MultiplexedConnection;
-
 use slog::{error, Logger};
 
 fn get_payout(channel: &Channel, event: &Event) -> BigNum {
@@ -47,10 +46,12 @@ pub async fn record(
                 ad_slot,
                 referrer,
             } => {
-                let payout = get_payout(channel, event)
+                let divisor = BigNum::from(10u64.pow(18));
+                let pay_amount = get_payout(channel, event)
+                    .div_floor(&divisor)
                     .to_u64()
                     .expect("should always have a payout");
-                let pay_amount = payout / 10u64.pow(18);
+                // let pay_amount = payout / 10u64.pow(18);
 
                 if let Some(ad_unit) = ad_unit {
                     db.zincr(

sentry/src/db/analytics.rs

@@ -4,7 +4,7 @@ use crate::Session;
 use bb8::RunError;
 use chrono::Utc;
 use primitives::analytics::{AnalyticsQuery, AnalyticsResponse, ANALYTICS_QUERY_LIMIT};
-use primitives::sentry::{AdvancedAnalyticsResponse, ChannelReport, Event, PublisherReport};
+use primitives::sentry::{AdvancedAnalyticsResponse, ChannelReport, PublisherReport};
 use primitives::{ChannelId, ValidatorId};
 use redis;
 use redis::aio::MultiplexedConnection;
@@ -23,6 +23,24 @@ pub enum AnalyticsType {
     },
 }
 
+pub async fn advertiser_channel_ids(
+    pool: &DbPool,
+    creator: &ValidatorId
+) -> Result<Vec<ChannelId>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool.run(move |connection| async move {
+        match connection.prepare("SELECT id FROM channels WHERE creator = {}").await {
+            Ok(stmt) => match connection.query(&stmt, &[creator]).await {
+                Ok(rows) => {
+                    let channel_ids: Vec<ChannelId> = rows.iter().map(ChannelId::from).collect();
+                    Ok((channel_ids, connection))
+                },
+                Err(e) => Err((e, connection)),
+            },
+            Err(e) => Err((e, connection)),
+        }
+    })
+    .await
+}
 pub async fn get_analytics(
     query: AnalyticsQuery,
     pool: &DbPool,
@@ -146,7 +164,7 @@ async fn stat_pair(
 
 pub async fn get_advanced_reports(
     redis: &MultiplexedConnection,
-    event: &Event,
+    event_type: &str,
     publisher: &ValidatorId,
     channel_ids: &[ChannelId],
 ) -> Result<AdvancedAnalyticsResponse, Box<dyn Error>> {
@@ -166,10 +184,10 @@ pub async fn get_advanced_reports(
                 "{}:{}:{}:{}",
                 epoch().floor(),
                 publisher_report,
-                event,
+                event_type,
                 publisher
             ),
-            _ => format!("{}:{}:{}", publisher_report, event, publisher),
+            _ => format!("{}:{}:{}", publisher_report, event_type, publisher),
         };
         let result = stat_pair(redis.clone(), &pair).await?;
         publisher_stats.insert(publisher_report.clone(), result);
@@ -189,7 +207,7 @@ pub async fn get_advanced_reports(
         for channel_report in channel_reports.iter() {
             let result = stat_pair(
                 redis.clone(),
-                &format!("{}:{}:{}", channel_report, event, channel_id),
+                &format!("{}:{}:{}", channel_report, event_type, channel_id),
             )
             .await?;
             channel_stat.insert(channel_report.clone(), result);

sentry/src/lib.rs

@@ -18,7 +18,7 @@ use primitives::adapter::Adapter;
 use primitives::{Config, ValidatorId};
 use redis::aio::MultiplexedConnection;
 use regex::Regex;
-use routes::analytics::{advertiser_analytics, analytics, publisher_analytics};
+use routes::analytics::{advertiser_analytics, analytics, publisher_analytics, advanced_analytics};
 use routes::cfg::config;
 use routes::channel::{
     channel_list, create_channel, create_validator_messages, insert_events, last_approved,
@@ -141,6 +141,16 @@ impl<A: Adapter + 'static> Application<A> {
             ("/channel/list", &Method::GET) => channel_list(req, &self).await,
 
             ("/analytics", &Method::GET) => analytics(req, &self).await,
+            ("/analytics/advanced", &Method::GET) =>  {
+                let req = match chain(req, &self, vec![Box::new(auth_required_middleware)]).await {
+                    Ok(req) => req,
+                    Err(error) => {
+                        return map_response_error(error);
+                    }
+                };
+
+                advanced_analytics(req, &self).await
+            },
             ("/analytics/for-advertiser", &Method::GET) => {
                 let req = match chain(req, &self, vec![Box::new(auth_required_middleware)]).await {
                     Ok(req) => req,

sentry/src/routes/analytics.rs

@@ -1,4 +1,4 @@
-use crate::db::analytics::{get_analytics, AnalyticsType};
+use crate::db::analytics::{get_analytics, AnalyticsType, advertiser_channel_ids, get_advanced_reports};
 use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
@@ -97,6 +97,29 @@ pub async fn process_analytics<A: Adapter>(
         .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))
 }
 
+
+pub async fn advanced_analytics<A: Adapter>(
+    req: Request<Body>,
+    app: &Application<A>,
+) -> Result<Response<Body>, ResponseError> {
+    let sess = req.extensions().get::<Session>().expect("auth is required");
+    let advertiser_channels = advertiser_channel_ids(&app.pool, &sess.uid).await?;
+
+    let event_type = serde_urlencoded::from_str::<String>(&req.uri().query().unwrap_or(""))?;
+
+    let response = get_advanced_reports(
+        &app.redis,
+        &event_type,
+        &sess.uid,
+        &advertiser_channels
+    )
+    .await
+    .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))?;
+
+    Ok(success_response(serde_json::to_string(&response)?))
+
+}
+
 async fn cache(
     redis: &MultiplexedConnection,
     key: String,