Merge pull request #496 from AmbireTech/analytics-multichain-support

Analytics multichain support

Author: simzzz

Cargo.lock

@@ -1,6 +1,6 @@
 use crate::{
     sentry::{EventType, IMPRESSION},
-    Address, CampaignId, ValidatorId, IPFS,
+    Address, CampaignId, ChainId, ValidatorId, IPFS,
 };
 use parse_display::Display;
 use serde::{Deserialize, Serialize};
@@ -115,6 +115,8 @@ pub struct AnalyticsQuery {
     pub hostname: Option<String>,
     pub country: Option<String>,
     pub os_name: Option<OperatingSystem>,
+    #[serde(default)]
+    pub chains: Vec<ChainId>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Display, Hash, Eq)]

primitives/src/analytics.rs

@@ -2,13 +2,14 @@ use serde::{Deserialize, Serialize};
 use std::fmt;
 
 use crate::{config::TokenInfo, util::ApiUrl, Address, Campaign, Channel};
+use parse_display::Display;
 
 /// The Id of the chain
 ///
 /// # Ethereum Virtual Machine
 ///
 /// For all the EVM-compatible Chain IDs visit <https://chainid.network>
-#[derive(Serialize, Deserialize, Hash, Clone, Copy, Eq, PartialEq)]
+#[derive(Serialize, Deserialize, Hash, Clone, Copy, Eq, PartialEq, Display)]
 #[serde(transparent)]
 pub struct ChainId(u32);
 
@@ -21,6 +22,10 @@ impl ChainId {
 
         Self(id)
     }
+
+    pub fn to_u32(self) -> u32 {
+        self.0
+    }
 }
 
 impl From<u32> for ChainId {
@@ -110,3 +115,34 @@ impl ChainOf<Campaign> {
         }
     }
 }
+
+#[cfg(feature = "postgres")]
+pub mod postgres {
+    use super::ChainId;
+    use bytes::BytesMut;
+    use std::error::Error;
+    use tokio_postgres::types::{accepts, to_sql_checked, FromSql, IsNull, ToSql, Type};
+
+    impl<'a> FromSql<'a> for ChainId {
+        fn from_sql(ty: &Type, raw: &'a [u8]) -> Result<ChainId, Box<dyn Error + Sync + Send>> {
+            let value = <i32 as FromSql>::from_sql(ty, raw)?;
+
+            Ok(ChainId(u32::try_from(value)?))
+        }
+        accepts!(INT4);
+    }
+
+    impl ToSql for ChainId {
+        fn to_sql(
+            &self,
+            ty: &Type,
+            w: &mut BytesMut,
+        ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
+            <i32 as ToSql>::to_sql(&self.0.try_into()?, ty, w)
+        }
+
+        accepts!(INT4);
+
+        to_sql_checked!();
+    }
+}

primitives/src/chain.rs

@@ -3,7 +3,7 @@ use crate::{
     balances::BalancesState,
     spender::Spender,
     validator::{ApproveState, Heartbeat, MessageTypes, NewState, Type as MessageType},
-    Address, Balances, CampaignId, UnifiedMap, UnifiedNum, ValidatorId, IPFS,
+    Address, Balances, CampaignId, ChainId, UnifiedMap, UnifiedNum, ValidatorId, IPFS,
 };
 use chrono::{
     serde::ts_milliseconds, Date, DateTime, Datelike, Duration, NaiveDate, TimeZone, Timelike, Utc,
@@ -284,6 +284,7 @@ pub struct UpdateAnalytics {
     pub hostname: Option<String>,
     pub country: Option<String>,
     pub os_name: OperatingSystem,
+    pub chain_id: ChainId,
     pub event_type: EventType,
     pub amount_to_add: UnifiedNum,
     pub count_to_add: i32,

primitives/src/sentry.rs

@@ -61,6 +61,7 @@ once_cell = "1.5.2"
 woothee = "0.13"
 # Making requests to the platform
 reqwest = { version = "0.11", features = ["json", "cookies"] }
+serde_qs = "0.9.2"
 
 [dev-dependencies]
 pretty_assertions = "1"

sentry/Cargo.toml

@@ -93,9 +93,10 @@ CREATE TABLE analytics (
     hostname varchar(255) NOT NULL,
     country varchar(255) NOT NULL,
     os_name varchar(255) NOT NULL,
+    chain_id integer NOT NULL,
     event_type varchar(255) NOT NULL,
     payout_amount bigint NOT NULL DEFAULT 0,
     payout_count integer NOT NULL DEFAULT 0,
     -- Do not rename the Primary key constraint (`analytics_pkey`)!
-    PRIMARY KEY (campaign_id, "time", ad_unit, ad_slot, ad_slot_type, advertiser, publisher, hostname, country, os_name, event_type)
+    PRIMARY KEY (campaign_id, "time", ad_unit, ad_slot, ad_slot_type, advertiser, publisher, hostname, country, os_name, chain_id, event_type)
 );

sentry/migrations/20190806011140_initial-tables/up.sql

@@ -5,14 +5,14 @@ use crate::{
 use primitives::{
     analytics::OperatingSystem,
     sentry::{DateHour, Event, UpdateAnalytics},
-    Address, Campaign, UnifiedNum,
+    Address, Campaign, ChainOf, UnifiedNum,
 };
 use std::collections::HashMap;
 
 /// Validator fees will not be included in analytics
 pub async fn record(
     pool: &DbPool,
-    campaign: &Campaign,
+    campaign_context: &ChainOf<Campaign>,
     session: &Session,
     events_with_payouts: Vec<(Event, Address, UnifiedNum)>,
 ) -> Result<(), PoolError> {
@@ -43,7 +43,8 @@ pub async fn record(
                     ad_slot,
                 } => (*publisher, *ad_unit, referrer.clone(), *ad_slot),
             };
-            let ad_unit = campaign
+            let ad_unit = campaign_context
+                .context
                 .ad_units
                 .iter()
                 .find(|ad_unit| ad_unit.ipfs == event_ad_unit);
@@ -70,16 +71,17 @@ pub async fn record(
                 analytics.count_to_add += 1;
             })
             .or_insert_with(|| UpdateAnalytics {
-                campaign_id: campaign.id,
+                campaign_id: campaign_context.context.id,
                 time: datehour,
                 ad_unit,
                 ad_slot,
                 ad_slot_type,
-                advertiser: campaign.creator,
+                advertiser: campaign_context.context.creator,
                 publisher,
                 hostname,
                 country: session.country.to_owned(),
                 os_name: os_name.clone(),
+                chain_id: campaign_context.chain.chain_id,
                 event_type,
                 amount_to_add: payout_amount,
                 count_to_add: 1,
@@ -95,6 +97,7 @@ pub async fn record(
 #[cfg(test)]
 mod test {
     use super::*;
+    use crate::test_util::setup_dummy_app;
     use primitives::{
         sentry::{Analytics, CLICK, IMPRESSION},
         test_util::{DUMMY_CAMPAIGN, DUMMY_IPFS, PUBLISHER},
@@ -164,6 +167,8 @@ mod test {
 
     #[tokio::test]
     async fn test_analytics_recording_with_empty_events() {
+        let app = setup_dummy_app().await;
+
         let test_events = get_test_events();
         let database = DATABASE_POOL.get().await.expect("Should get a DB pool");
 
@@ -186,9 +191,22 @@ mod test {
             test_events["impression"].clone(),
         ];
 
-        record(&database.clone(), &campaign, &session, input_events.clone())
-            .await
-            .expect("should record");
+        let dummy_channel = DUMMY_CAMPAIGN.channel;
+        let channel_chain = app
+            .config
+            .find_chain_of(dummy_channel.token)
+            .expect("Channel token should be whitelisted in config!");
+        let channel_context = channel_chain.with_channel(dummy_channel);
+        let campaign_context = channel_context.clone().with(campaign);
+
+        record(
+            &database.clone(),
+            &campaign_context,
+            &session,
+            input_events.clone(),
+        )
+        .await
+        .expect("should record");
 
         let analytics = get_all_analytics(&database.pool)
             .await
@@ -218,6 +236,8 @@ mod test {
 
     #[tokio::test]
     async fn test_recording_with_session() {
+        let app = setup_dummy_app().await;
+
         let database = DATABASE_POOL.get().await.expect("Should get a DB pool");
 
         setup_test_migrations(database.pool.clone())
@@ -247,9 +267,21 @@ mod test {
             test_events["impression"].clone(),
         ];
 
-        record(&database.clone(), &campaign, &session, input_events.clone())
-            .await
-            .expect("should record");
+        let dummy_channel = DUMMY_CAMPAIGN.channel;
+        let channel_chain = app
+            .config
+            .find_chain_of(dummy_channel.token)
+            .expect("Channel token should be whitelisted in config!");
+        let channel_context = channel_chain.with_channel(dummy_channel);
+        let campaign_context = channel_context.clone().with(campaign);
+        record(
+            &database.clone(),
+            &campaign_context,
+            &session,
+            input_events.clone(),
+        )
+        .await
+        .expect("should record");
 
         let analytics = get_all_analytics(&database.pool)
             .await