changes to analytics - nullable slot/unit, DateHour struct, separate analytics struct for update/get, test changes

Author: simzzz

primitives/src/analytics.rs

@@ -80,7 +80,7 @@ pub struct AnalyticsQuery {
     pub segment_by_channel: Option<String>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Display)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Display, Hash, Eq)]
 #[serde(untagged, into = "String", from = "String")]
 pub enum OperatingSystem {
     Linux,

primitives/src/campaign.rs

@@ -29,7 +29,7 @@ mod campaign_id {
     use thiserror::Error;
     use uuid::Uuid;
 
-    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
     /// an Id of 16 bytes, (de)serialized as a `0x` prefixed hex
     /// In this implementation of the `CampaignId` the value is generated from a `Uuid::new_v4().to_simple()`
     pub struct CampaignId([u8; 16]);

primitives/src/sentry.rs

@@ -5,9 +5,11 @@ use crate::{
     validator::{ApproveState, Heartbeat, MessageTypes, NewState, Type as MessageType},
     Address, Balances, BigNum, CampaignId, Channel, ChannelId, UnifiedNum, ValidatorId, IPFS,
 };
-use chrono::{DateTime, Utc};
+use bytes::BytesMut;
+use chrono::{Date, DateTime, Datelike, TimeZone, Timelike, Utc};
+use postgres_types::{accepts, to_sql_checked, FromSql, IsNull, ToSql, Type};
 use serde::{Deserialize, Serialize};
-use std::{collections::HashMap, fmt, hash::Hash};
+use std::{collections::HashMap, error::Error, fmt, hash::Hash};
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
@@ -137,10 +139,10 @@ pub enum Event {
     },
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Hash, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct UpdateAnalytics {
-    pub time: DateTime<Utc>,
+    pub time: DateHour,
     pub campaign_id: CampaignId,
     pub ad_unit: Option<IPFS>,
     pub ad_slot: Option<IPFS>,
@@ -158,7 +160,7 @@ pub struct UpdateAnalytics {
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct Analytics {
-    pub time: DateTime<Utc>,
+    pub time: DateHour,
     pub campaign_id: CampaignId,
     pub ad_unit: Option<IPFS>,
     pub ad_slot: Option<IPFS>,
@@ -173,29 +175,34 @@ pub struct Analytics {
     pub payout_count: i32,
 }
 
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Hash, Eq)]
 pub struct DateHour {
-    date: Date,
-    hour: u64,
+    pub date: DateTime<Utc>,
+    pub hour: u32,
 }
 
-impl <'a> FromSql<'a> for DateHour {
+impl<'a> FromSql<'a> for DateHour {
     fn from_sql(ty: &Type, raw: &'a [u8]) -> Result<Self, Box<dyn Error + Sync + Send>> {
-        let str_slice = <&str as FromSql>::from_sql(ty, raw)?;
-
-        Ok(json.0)
+        let date_time = <DateTime<Utc> as FromSql>::from_sql(ty, raw)?;
+        assert_eq!(date_time.time().minute(), 0);
+        assert_eq!(date_time.time().second(), 0);
+        assert_eq!(date_time.time().nanosecond(), 0);
+        Ok(Self {
+            date: date_time,
+            hour: date_time.hour(),
+        })
     }
+    accepts!(TIMESTAMPTZ);
 }
 
 impl ToSql for DateHour {
-    fn to_sql(
-        &self,
-        ty: &Type,
-        w: &mut BytesMut,
-    ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
-        self.to_string().to_sql(ty, w)
+    fn to_sql(&self, ty: &Type, w: &mut BytesMut) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
+        Utc.ymd(self.date.year(), self.date.month(), self.date.day())
+            .and_hms(self.hour, 0, 0)
+            .to_sql(ty, w)
     }
 
-    accepts!(TEXT, VARCHAR);
+    accepts!(TIMESTAMPTZ);
     to_sql_checked!();
 }
 
@@ -665,7 +672,7 @@ pub mod campaign_create {
 
 #[cfg(feature = "postgres")]
 mod postgres {
-    use super::{Analytics, UpdateAnalytics, MessageResponse, ValidatorMessage};
+    use super::{Analytics, MessageResponse, ValidatorMessage};
     use crate::{
         sentry::EventAggregate,
         validator::{messages::Type as MessageType, MessageTypes},

primitives/src/unified_num.rs

@@ -59,6 +59,7 @@ use std::{
     FromStr,
     Serialize,
     Deserialize,
+    Hash,
 )]
 #[serde(into = "String", try_from = "String")]
 pub struct UnifiedNum(u64);

sentry/src/analytics.rs

@@ -5,9 +5,10 @@ use crate::{
 use chrono::{Timelike, Utc};
 use primitives::{
     analytics::OperatingSystem,
-    sentry::{Event, Analytics, UpdateAnalytics},
+    sentry::{DateHour, Event, UpdateAnalytics},
     Address, Campaign, UnifiedNum,
 };
+use std::collections::HashSet;
 
 /// Validator fees will not be included in analytics
 pub async fn record(
@@ -16,14 +17,22 @@ pub async fn record(
     session: &Session,
     events_with_payouts: Vec<(Event, Address, UnifiedNum)>,
 ) -> Result<(), PoolError> {
-    let os_name = session.os.as_ref().map(|os| OperatingSystem::map_os(os)).unwrap_or_default();
+    let os_name = session
+        .os
+        .as_ref()
+        .map(|os| OperatingSystem::map_os(os))
+        .unwrap_or_default();
     let time = {
         let full_utc = Utc::now();
 
         // leave only the Hour portion and erase the minutes & seconds
-        full_utc.date().and_hms(full_utc.hour(), 0, 0)
+        DateHour {
+            date: full_utc.date().and_hms(0, 0, 0), // TODO: Fix
+            hour: full_utc.hour(),
+        }
     };
 
+    let mut analytics_set: HashSet<UpdateAnalytics> = HashSet::new();
     for (event, _payout_addr, payout_amount) in events_with_payouts {
         let event_type = event.to_string();
         let (publisher, ad_unit, referrer, ad_slot, ad_slot_type) = {
@@ -62,7 +71,7 @@ pub async fn record(
         };
 
         // DB: Insert or Update all events
-        let event_for_db = UpdateAnalytics {
+        let mut analytics = UpdateAnalytics {
             campaign_id: campaign.id,
             time,
             ad_unit,
@@ -77,18 +86,28 @@ pub async fn record(
             amount_to_add: payout_amount,
             count_to_add: 1,
         };
-
-        insert_analytics(pool, &event_for_db).await?;
+        // TODO: tidy up this operation
+        match analytics_set.get(&analytics) {
+            Some(a) => {
+                analytics.amount_to_add += &a.amount_to_add;
+                analytics.count_to_add = a.count_to_add + 1;
+                let _ = &analytics_set.replace(a.to_owned());
+            }
+            None => {
+                let _ = &analytics_set.insert(analytics);
+            }
+        }
+    }
+    for a in analytics_set.iter() {
+        insert_analytics(pool, a).await?;
     }
-
     Ok(())
 }
 
 #[cfg(test)]
 mod test {
     use super::*;
     use primitives::{
-        analytics::OperatingSystem,
         util::tests::prep_db::{ADDRESSES, DUMMY_CAMPAIGN},
         UnifiedNum,
     };
@@ -146,53 +165,23 @@ mod test {
             .await
             .expect("should record");
 
-        let query_click_event = Analytics {
-            time: Utc::now(),
-            campaign_id: DUMMY_CAMPAIGN.id,
-            ad_unit: None,
-            ad_slot: None,
-            ad_slot_type: None,
-            advertiser: campaign.creator,
-            publisher: ADDRESSES["leader"],
-            hostname: None,
-            country: None,
-            os_name: OperatingSystem::Other,
-            event_type: "CLICK".to_string(),
-            payout_amount: Default::default(),
-            payout_count: 1,
-        };
-
-        let query_impression_event = Analytics {
-            time: Utc::now(),
-            campaign_id: DUMMY_CAMPAIGN.id,
-            ad_unit: None,
-            ad_slot: None,
-            ad_slot_type: None,
-            advertiser: campaign.creator,
-            publisher: ADDRESSES["leader"],
-            hostname: None,
-            country: None,
-            os_name: OperatingSystem::Other,
-            event_type: "IMPRESSION".to_string(),
-            payout_amount: Default::default(),
-            payout_count: 1,
-        };
-
-        let click_analytics = find_analytics(&database.pool, &query_click_event)
+        let analytics = find_analytics(&database.pool)
             .await
             .expect("should find analytics");
-        let impression_analytics =
-            find_analytics(&database.pool, &query_impression_event)
-                .await
-                .expect("should find analytics");
-        assert_eq!(click_analytics.event_type, "CLICK".to_string());
+        let click_analytics = analytics
+            .iter()
+            .find(|a| a.event_type == "CLICK")
+            .expect("There should be a click event");
+        let impression_analytics = analytics
+            .iter()
+            .find(|a| a.event_type == "IMPRESSION")
+            .expect("There should be an impression event");
         assert_eq!(
             click_analytics.payout_amount,
             UnifiedNum::from_u64(1_000_000)
         );
         assert_eq!(click_analytics.payout_count, 1);
 
-        assert_eq!(impression_analytics.event_type, "IMPRESSION".to_string());
         assert_eq!(
             impression_analytics.payout_amount,
             UnifiedNum::from_u64(1_000_000)
@@ -203,13 +192,17 @@ mod test {
             .await
             .expect("should record");
 
-        let click_analytics = find_analytics(&database.pool, &query_click_event)
+        let analytics = find_analytics(&database.pool)
             .await
             .expect("should find analytics");
-        let impression_analytics =
-            find_analytics(&database.pool, &query_impression_event)
-                .await
-                .expect("should find analytics");
+        let click_analytics = analytics
+            .iter()
+            .find(|a| a.event_type == "CLICK")
+            .expect("There should be a click event");
+        let impression_analytics = analytics
+            .iter()
+            .find(|a| a.event_type == "IMPRESSION")
+            .expect("There should be an impression event");
         assert_eq!(
             click_analytics.payout_amount,
             UnifiedNum::from_u64(2_000_000)

sentry/src/db/analytics.rs

@@ -2,7 +2,9 @@ use crate::{epoch, Auth};
 use chrono::Utc;
 use primitives::{
     analytics::{AnalyticsData, AnalyticsQuery, ANALYTICS_QUERY_LIMIT},
-    sentry::{AdvancedAnalyticsResponse, ChannelReport, UpdateAnalytics, Analytics, PublisherReport},
+    sentry::{
+        AdvancedAnalyticsResponse, Analytics, ChannelReport, PublisherReport, UpdateAnalytics,
+    },
     ChannelId, ValidatorId,
 };
 use redis::{aio::MultiplexedConnection, cmd};
@@ -274,45 +276,16 @@ pub async fn insert_analytics(
     Ok(event_analytics)
 }
 
-pub async fn find_analytics(
-    pool: &DbPool,
-    event: &Analytics,
-) -> Result<Analytics, PoolError> {
+// Currently used for testing only
+pub async fn find_analytics(pool: &DbPool) -> Result<Vec<Analytics>, PoolError> {
     let client = pool.get().await?;
 
-    let query = "SELECT campaign_id, time, ad_unit, ad_slot, ad_slot_type, advertiser, publisher, hostname, country, os, event_type, payout_amount, payout_count
-    FROM analytics WHERE campaign_id = $1 AND time = date_trunc('hour', cast($2 as timestamp with time zone)) AND ad_unit = $3 AND ad_slot = $4 AND ad_slot_type = $5 AND advertiser = $6 AND publisher = $7 AND hostname = $8 AND country = $9 AND os = $10 AND event_type = $11";
-
+    let query = "SELECT * FROM analytics";
     let stmt = client.prepare(query).await?;
-    let ad_unit = match event.ad_unit {
-        Some(ipfs) => ipfs.to_string(),
-        None => "".to_string(),
-    };
 
-    let ad_slot = match event.ad_slot {
-        Some(ipfs) => ipfs.to_string(),
-        None => "".to_string(),
-    };
-    let row = client
-        .query_one(
-            &stmt,
-            &[
-                &event.campaign_id,
-                &event.time,
-                &ad_unit,
-                &ad_slot,
-                &event.ad_slot_type.as_ref().unwrap_or(&"".to_string()),
-                &event.advertiser,
-                &event.publisher,
-                &event.hostname.as_ref().unwrap_or(&"".to_string()),
-                &event.country.as_ref().unwrap_or(&"".to_string()),
-                &event.os_name.to_string(),
-                &event.event_type,
-            ],
-        )
-        .await?;
+    let rows = client.query(&stmt, &[]).await?;
 
-    let event_analytics = Analytics::from(&row);
+    let event_analytics: Vec<Analytics> = rows.iter().map(Analytics::from).collect();
     Ok(event_analytics)
 }
 
@@ -321,6 +294,7 @@ mod test {
     use super::*;
     use primitives::{
         analytics::OperatingSystem,
+        sentry::DateHour,
         util::tests::prep_db::{ADDRESSES, DUMMY_CAMPAIGN},
         UnifiedNum, IPFS,
     };
@@ -338,7 +312,10 @@ mod test {
             .expect("Migrations should succeed");
         {
             let analytics = UpdateAnalytics {
-                time: Utc.ymd(2021, 2, 1).and_hms(7, 0, 0),
+                time: DateHour {
+                    date: Utc.ymd(2021, 2, 1).and_hms(1, 0, 0),
+                    hour: 1,
+                },
                 campaign_id: DUMMY_CAMPAIGN.id,
                 ad_unit: Some(
                     IPFS::try_from("Qmasg8FrbuSQpjFu3kRnZF9beg8rEBFrqgi1uXDRwCbX5f")
@@ -359,13 +336,13 @@ mod test {
                 count_to_add: 1,
             };
 
-            let insert_res =
-                insert_analytics(&database.clone(), &analytics.clone())
-                    .await
-                    .expect("Should insert");
+            let insert_res = insert_analytics(&database.clone(), &analytics.clone())
+                .await
+                .expect("Should insert");
 
             assert_eq!(insert_res.campaign_id, analytics.campaign_id);
-            assert_eq!(insert_res.time.date(), analytics.time.date());
+            assert_eq!(insert_res.time.date, analytics.time.date);
+            assert_eq!(insert_res.time.hour, analytics.time.hour);
             assert_eq!(insert_res.ad_unit, analytics.ad_unit);
             assert_eq!(insert_res.ad_slot, analytics.ad_slot);
             assert_eq!(insert_res.ad_slot_type, analytics.ad_slot_type);
@@ -386,7 +363,10 @@ mod test {
         }
         {
             let analytics_with_empty_fields = UpdateAnalytics {
-                time: Utc.ymd(2021, 2, 1).and_hms(7, 0, 0),
+                time: DateHour {
+                    date: Utc.ymd(2021, 2, 1).and_hms(1, 0, 0),
+                    hour: 1,
+                },
                 campaign_id: DUMMY_CAMPAIGN.id,
                 ad_unit: None,
                 ad_slot: None,