Merge branch 'dev' into validator-worker

Author: samparsky

README.md

@@ -45,6 +45,11 @@ you need to be running those in order to run the automated tests:
 
 TODO
 
+### Bug
+
+
+
+
 ### Run Validator Worker
 
 TODO

adapter/src/ethereum.rs

@@ -598,7 +598,10 @@ mod test {
         token_contract
             .call(
                 "setBalanceTo",
-                (Token::Address(leader_account), Token::Uint(2000.into())),
+                (
+                    Token::Address(leader_account),
+                    Token::Uint(U256::from(2000 as u64)),
+                ),
                 leader_account,
                 Options::default(),
             )
@@ -649,6 +652,7 @@ mod test {
                 nonce: None,
                 withdraw_period_start: Utc::now() + Duration::days(1),
                 ad_units: vec![],
+                pricing_bounds: None,
             },
         };
 

primitives/src/analytics.rs

@@ -1,30 +1,42 @@
+use crate::ChannelId;
 use crate::DomainError;
 use serde::{Deserialize, Serialize};
 
 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 {
-    time: u32,
-    value: String,
+    pub aggr: Vec<AnalyticsData>,
+    pub limit: u32,
 }
 
 #[cfg(feature = "postgres")]
 pub mod postgres {
-    use super::AnalyticsResponse;
+    use super::AnalyticsData;
     use tokio_postgres::Row;
 
-    impl From<&Row> for AnalyticsResponse {
+    impl From<&Row> for AnalyticsData {
         fn from(row: &Row) -> Self {
             Self {
                 time: row.get("time"),
                 value: row.get("value"),
+                channel_id: row.try_get("channel_id").ok(),
             }
         }
     }
 }
 
 #[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct AnalyticsQuery {
     #[serde(default = "default_limit")]
     pub limit: u32,
@@ -34,6 +46,7 @@ pub struct AnalyticsQuery {
     pub metric: String,
     #[serde(default = "default_timeframe")]
     pub timeframe: String,
+    pub segment_by_channel: Option<String>,
 }
 
 impl AnalyticsQuery {

primitives/src/big_num.rs

@@ -10,7 +10,19 @@ use num_derive::{Num, NumOps, One, Zero};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 
 #[derive(
-    Serialize, Deserialize, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, NumOps, One, Zero, Num,
+    Serialize,
+    Deserialize,
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    NumOps,
+    One,
+    Zero,
+    Num,
+    Default,
 )]
 pub struct BigNum(
     #[serde(

primitives/src/channel.rs

@@ -63,6 +63,19 @@ pub struct Channel {
     pub spec: ChannelSpec,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Pricing {
+    pub max: BigNum,
+    pub min: BigNum,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(rename_all = "UPPERCASE")]
+pub struct PricingBounds {
+    pub impression: Option<Pricing>,
+    pub click: Option<Pricing>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct ChannelSpec {
@@ -73,6 +86,8 @@ pub struct ChannelSpec {
     pub max_per_impression: BigNum,
     /// Minimum payment offered per impression
     pub min_per_impression: BigNum,
+    // Event pricing bounds
+    pub pricing_bounds: Option<PricingBounds>,
     /// An array of TargetingTag (optional)
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub targeting: Vec<TargetingTag>,
@@ -270,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,

primitives/src/sentry.rs

@@ -1,42 +1,56 @@
-use crate::validator::{ApproveState, Heartbeat, MessageTypes, NewState};
+use crate::validator::MessageTypes;
 use crate::{BigNum, Channel, ChannelId, ValidatorId};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::fmt;
+use std::hash::Hash;
 
 #[derive(Serialize, Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct LastApproved {
     /// NewState can be None if the channel is brand new
     pub new_state: Option<NewStateValidatorMessage>,
     /// ApproveState can be None if the channel is brand new
-    pub approved_state: Option<ApproveStateValidatorMessage>,
+    pub approve_state: Option<ApproveStateValidatorMessage>,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct NewStateValidatorMessage {
-    pub from: String,
+    pub from: ValidatorId,
     pub received: DateTime<Utc>,
-    pub msg: NewState,
+    pub msg: MessageTypes,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ApproveStateValidatorMessage {
-    pub from: String,
+    pub from: ValidatorId,
+    pub received: DateTime<Utc>,
+    pub msg: MessageTypes,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct HeartbeatValidatorMessage {
+    pub from: ValidatorId,
     pub received: DateTime<Utc>,
-    pub msg: ApproveState,
+    pub msg: MessageTypes,
 }
 
 #[serde(tag = "type", rename_all = "SCREAMING_SNAKE_CASE")]
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone, Debug)]
 pub enum Event {
     #[serde(rename_all = "camelCase")]
     Impression {
         publisher: ValidatorId,
         ad_unit: Option<String>,
+        ad_slot: Option<String>,
+        referrer: Option<String>,
     },
     Click {
-        publisher: String,
+        publisher: ValidatorId,
+        ad_unit: Option<String>,
+        ad_slot: Option<String>,
+        referrer: Option<String>,
     },
     ImpressionWithCommission {
         earners: Vec<Earner>,
@@ -55,7 +69,37 @@ pub enum Event {
     Close,
 }
 
-#[derive(Serialize, Deserialize)]
+impl Event {
+    pub fn is_click_event(&self) -> bool {
+        match *self {
+            Event::Click { .. } => true,
+            _ => false,
+        }
+    }
+
+    pub fn is_impression_event(&self) -> bool {
+        match *self {
+            Event::Impression { .. } => true,
+            _ => false,
+        }
+    }
+}
+
+impl fmt::Display for Event {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match *self {
+            Event::Impression { .. } => write!(f, "IMPRESSION"),
+            Event::Click { .. } => write!(f, "CLICK"),
+            Event::ImpressionWithCommission { .. } => write!(f, "IMPRESSION_WITH_COMMMISION"),
+            Event::UpdateImpressionPrice { .. } => write!(f, "UPDATE_IMPRESSION_PRICE"),
+            Event::Pay { .. } => write!(f, "PAY"),
+            Event::PauseChannel => write!(f, "PAUSE_CHANNEL"),
+            Event::Close => write!(f, "CLOSE"),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct Earner {
     #[serde(rename = "publisher")]
     pub address: String,
@@ -89,7 +133,7 @@ pub struct ChannelListResponse {
 #[serde(rename_all = "camelCase")]
 pub struct LastApprovedResponse {
     pub last_approved: Option<LastApproved>,
-    pub heartbeats: Option<Vec<Heartbeat>>,
+    pub heartbeats: Option<Vec<HeartbeatValidatorMessage>>,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -116,9 +160,59 @@ pub struct EventAggregateResponse {
     pub events: Vec<EventAggregate>,
 }
 
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AdvancedAnalyticsResponse {
+    pub by_channel_stats: HashMap<ChannelId, HashMap<ChannelReport, HashMap<String, f64>>>,
+    pub publisher_stats: HashMap<PublisherReport, HashMap<String, f64>>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Hash, PartialEq, Eq, Clone)]
+#[serde(rename_all = "camelCase")]
+pub enum PublisherReport {
+    AdUnit,
+    AdSlot,
+    AdSlotPay,
+    Country,
+    Hostname,
+}
+
+impl fmt::Display for PublisherReport {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match *self {
+            PublisherReport::AdUnit => write!(f, "reportPublisherToAdUnit"),
+            PublisherReport::AdSlot => write!(f, "reportPublisherToAdSlot"),
+            PublisherReport::AdSlotPay => write!(f, "reportPublisherToAdSlotPay"),
+            PublisherReport::Country => write!(f, "reportPublisherToCountry"),
+            PublisherReport::Hostname => write!(f, "reportPublisherToHostname"),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug, Hash, PartialEq, Eq, Clone)]
+#[serde(rename_all = "camelCase")]
+pub enum ChannelReport {
+    AdUnit,
+    Hostname,
+    HostnamePay,
+}
+
+impl fmt::Display for ChannelReport {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match *self {
+            ChannelReport::AdUnit => write!(f, "reportPublisherToAdUnit"),
+            ChannelReport::Hostname => write!(f, "reportChannelToHostname"),
+            ChannelReport::HostnamePay => write!(f, "reportChannelToHostnamePay"),
+        }
+    }
+}
+
 #[cfg(feature = "postgres")]
 mod postgres {
-    use super::ValidatorMessage;
+    use super::{
+        ApproveStateValidatorMessage, HeartbeatValidatorMessage, NewStateValidatorMessage,
+        ValidatorMessage,
+    };
     use crate::sentry::EventAggregate;
     use crate::validator::MessageTypes;
     use bytes::BytesMut;
@@ -146,6 +240,36 @@ mod postgres {
         }
     }
 
+    impl From<&Row> for ApproveStateValidatorMessage {
+        fn from(row: &Row) -> Self {
+            Self {
+                from: row.get("from"),
+                received: row.get("received"),
+                msg: row.get::<_, Json<MessageTypes>>("msg").0,
+            }
+        }
+    }
+
+    impl From<&Row> for NewStateValidatorMessage {
+        fn from(row: &Row) -> Self {
+            Self {
+                from: row.get("from"),
+                received: row.get("received"),
+                msg: row.get::<_, Json<MessageTypes>>("msg").0,
+            }
+        }
+    }
+
+    impl From<&Row> for HeartbeatValidatorMessage {
+        fn from(row: &Row) -> Self {
+            Self {
+                from: row.get("from"),
+                received: row.get("received"),
+                msg: row.get::<_, Json<MessageTypes>>("msg").0,
+            }
+        }
+    }
+
     impl ToSql for MessageTypes {
         fn to_sql(
             &self,

primitives/src/util/tests/prep_db.rs

@@ -1,4 +1,5 @@
 use crate::{
+    channel::{Pricing, PricingBounds},
     BigNum, Channel, ChannelId, ChannelSpec, EventSubmission, SpecValidators, ValidatorDesc,
     ValidatorId,
 };
@@ -69,7 +70,7 @@ lazy_static! {
                 title: None,
                 validators: SpecValidators::new(DUMMY_VALIDATOR_LEADER.clone(), DUMMY_VALIDATOR_FOLLOWER.clone()),
                 max_per_impression: 10.into(),
-                min_per_impression: 10.into(),
+                min_per_impression: 1.into(),
                 targeting: vec![],
                 min_targeting_score: None,
                 event_submission: Some(EventSubmission { allow: vec![] }),
@@ -79,6 +80,7 @@ lazy_static! {
                 nonce: Some(nonce),
                 withdraw_period_start: Utc.timestamp_millis(4_073_414_400_000),
                 ad_units: vec![],
+                pricing_bounds: Some(PricingBounds {impression: None, click: Some(Pricing { max: 0.into(), min: 0.into()})}),
             },
         }
     };