fix: analytics db query

Author: samparsky

primitives/src/analytics.rs

@@ -1,18 +1,22 @@
 use crate::DomainError;
 use serde::{Deserialize, Serialize};
+use crate::ChannelId;
 
 pub const ANALYTICS_QUERY_LIMIT: u32 = 200;
 
 #[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct AnalyticsData {
     pub time: f64,
     pub value: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub channel_id: Option<ChannelId>
 }
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct AnalyticsResponse {
     pub aggr: Vec<AnalyticsData>,
-    pub limit: u32
+    pub limit: u32,
 }
 
 #[cfg(feature = "postgres")]
@@ -25,6 +29,7 @@ pub mod postgres {
             Self {
                 time: row.get("time"),
                 value: row.get("value"),
+                channel_id: row.try_get("channel_id").ok(),
             }
         }
     }

sentry/src/db/analytics.rs

@@ -27,7 +27,7 @@ pub async fn advertiser_channel_ids(
 ) -> Result<Vec<ChannelId>, RunError<bb8_postgres::tokio_postgres::Error>> {
     pool.run(move |connection| async move {
         match connection
-            .prepare("SELECT id FROM channels WHERE creator = {}")
+            .prepare("SELECT id FROM channels WHERE creator = $1")
             .await
         {
             Ok(stmt) => match connection.query(&stmt, &[creator]).await {
@@ -81,7 +81,7 @@ pub async fn get_analytics(
             ));
 
             format!(
-                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates", 
+                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time", 
                 query.metric, interval
             )
         }
@@ -99,7 +99,7 @@ pub async fn get_analytics(
             where_clauses.push("earner IS NULL".to_string());
 
             format!(
-                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates", 
+                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time", 
                 query.metric, interval
             )
         }
@@ -119,26 +119,20 @@ pub async fn get_analytics(
                 session.uid
             ));
 
-
-            // where_clauses.push(format!(
-            //     "events->'{}'->'{}'->'{}' IS NOT NULL",
-            //     query.event_type, query.metric, session.uid
-            // ));
-
             format!(
-                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time from event_aggregates", 
+                "SUM({}::numeric)::varchar as value, (extract(epoch from created) - (MOD( CAST (extract(epoch from created) AS NUMERIC), {}))) as time", 
                 query.metric, interval
             )
         }
     };
 
     if segment_by_channel {
-        select_clause = format!("{}, channelId", select_clause);
-        group_clause = format!("{}, channelId", group_clause);
+        select_clause = format!("{}, channel_id", select_clause);
+        group_clause = format!("{}, channel_id", group_clause);
     }
 
     let sql_query = format!(
-        "SELECT {} WHERE {} GROUP BY {} LIMIT {}",
+        "SELECT {} FROM event_aggregates WHERE {} GROUP BY {} LIMIT {}",
         select_clause,
         where_clauses.join(" AND "),
         group_clause,
@@ -208,6 +202,7 @@ pub async fn get_advanced_reports(
     publisher: &ValidatorId,
     channel_ids: &[ChannelId],
 ) -> Result<AdvancedAnalyticsResponse, Box<dyn Error>> {
+    println!("get advnaces");
     let publisher_reports = [
         PublisherReport::ReportPublisherToAdUnit,
         PublisherReport::ReportPublisherToAdSlot,
@@ -216,7 +211,7 @@ pub async fn get_advanced_reports(
         PublisherReport::ReportPublisherToHostname,
     ];
 
-    let mut publisher_stats = HashMap::new();
+    let mut publisher_stats: HashMap<PublisherReport, HashMap<String, f64>> = HashMap::new();
 
     for publisher_report in publisher_reports.iter() {
         let pair = match publisher_report {
@@ -255,6 +250,7 @@ pub async fn get_advanced_reports(
 
         by_channel_stats.insert(channel_id.to_owned(), channel_stat);
     }
+    
 
     Ok(AdvancedAnalyticsResponse {
         publisher_stats,

sentry/src/lib.rs

@@ -57,6 +57,7 @@ lazy_static! {
     static ref CHANNEL_EVENTS_AGGREGATES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events-aggregates/?$").expect("The regex should be valid");
     static ref ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
     static ref ADVERTISER_ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/for-advertiser/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
+    static ref PUBLISHER_ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/for-publisher/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");    
     static ref CREATE_EVENTS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events(/.*)?$").expect("The regex should be valid");
 
 }
@@ -216,6 +217,18 @@ async fn analytics_router<A: Adapter + 'static>(
                 ]).await?;
 
                 advertiser_analytics(req, app).await
+            } else if let Some(caps) = PUBLISHER_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 = chain(req, app, vec![
+                    Box::new(auth_required_middleware),
+                    Box::new(get_channel_id)
+                ]).await?;
+
+                publisher_analytics(req, app).await
             } else {
                 Err(ResponseError::NotFound)
             }

sentry/src/routes/analytics.rs

@@ -115,9 +115,9 @@ pub async fn advanced_analytics<A: Adapter>(
     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 query = serde_urlencoded::from_str::<AnalyticsQuery>(&req.uri().query().unwrap_or(""))?;
 
-    let response = get_advanced_reports(&app.redis, &event_type, &sess.uid, &advertiser_channels)
+    let response = get_advanced_reports(&app.redis, &query.event_type, &sess.uid, &advertiser_channels)
         .await
         .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))?;