fix: validator sql, analytics query

Author: samparsky

adapter/src/ethereum.rs

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

primitives/src/analytics.rs

@@ -3,18 +3,24 @@ use serde::{Deserialize, Serialize};
 
 pub const ANALYTICS_QUERY_LIMIT: u32 = 200;
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AnalyticsData {
+    pub time: f64,
+    pub value: String,
+}
+
 #[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"),
@@ -39,6 +45,13 @@ pub struct AnalyticsQuery {
 }
 
 impl AnalyticsQuery {
+    pub fn metric_to_column(&mut self) {
+        self.metric = match self.metric.as_str() {
+            "eventCounts"=> "count".to_string(),
+            "eventPayouts" => "payout".to_string(),
+            _ => "count".to_string(),
+        };
+    }
     pub fn is_valid(&self) -> Result<(), DomainError> {
         let valid_event_types = ["IMPRESSION", "CLICK"];
         let valid_metric = ["eventPayouts", "eventCounts"];

primitives/src/channel.rs

@@ -72,8 +72,8 @@ pub struct Pricing {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "UPPERCASE")]
 pub struct PricingBounds {
-    pub impression: Pricing,
-    pub click: Pricing,
+    pub impression: Option<Pricing>,
+    pub click: Option<Pricing>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]

primitives/src/sentry.rs

@@ -12,7 +12,7 @@ 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)]
@@ -29,6 +29,13 @@ pub struct ApproveStateValidatorMessage {
     pub msg: ApproveState,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct HeartbeatValidatorMessage {
+    pub from: String,
+    pub received: DateTime<Utc>,
+    pub msg: Heartbeat,
+}
+
 #[serde(tag = "type", rename_all = "SCREAMING_SNAKE_CASE")]
 #[derive(Serialize, Deserialize, Clone)]
 pub enum Event {
@@ -126,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)]
@@ -202,9 +209,12 @@ impl fmt::Display for ChannelReport {
 
 #[cfg(feature = "postgres")]
 mod postgres {
-    use super::ValidatorMessage;
+    use super::{
+        ApproveStateValidatorMessage, HeartbeatValidatorMessage, NewStateValidatorMessage,
+        ValidatorMessage,
+    };
     use crate::sentry::EventAggregate;
-    use crate::validator::MessageTypes;
+    use crate::validator::{ApproveState, Heartbeat, MessageTypes, NewState};
     use bytes::BytesMut;
     use postgres_types::{accepts, to_sql_checked, IsNull, Json, ToSql, Type};
     use std::error::Error;
@@ -230,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<ApproveState>>("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<NewState>>("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<Heartbeat>>("msg").0,
+            }
+        }
+    }
+
     impl ToSql for MessageTypes {
         fn to_sql(
             &self,

primitives/src/validator.rs

@@ -178,6 +178,7 @@ pub mod postgres {
     use postgres_types::{FromSql, IsNull, ToSql, Type};
     use std::convert::TryFrom;
     use std::error::Error;
+    use crate::ToETHChecksum;
 
     impl<'a> FromSql<'a> for ValidatorId {
         fn from_sql(ty: &Type, raw: &'a [u8]) -> Result<Self, Box<dyn Error + Sync + Send>> {
@@ -201,7 +202,7 @@ pub mod postgres {
             ty: &Type,
             w: &mut BytesMut,
         ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
-            let string = format!("0x{}", self.to_hex_non_prefix_string());
+            let string = self.to_checksum();
 
             <String as ToSql>::to_sql(&string, ty, w)
         }
@@ -215,7 +216,7 @@ pub mod postgres {
             ty: &Type,
             out: &mut BytesMut,
         ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
-            let string = format!("0x{}", self.to_hex_non_prefix_string());
+            let string = self.to_checksum();
 
             <String as ToSql>::to_sql_checked(&string, ty, out)
         }

sentry/src/db/analytics.rs

@@ -55,16 +55,17 @@ pub async fn get_analytics(
     let time_limit = Utc::now().timestamp() - period;
 
     let mut where_clauses = vec![format!("created > to_timestamp({})", time_limit)];
+
     if let Some(id) = channel_id {
-        where_clauses.push(format!("channel_id = {}", id));
+        where_clauses.push(format!("channel_id = '{}'", id));
     }
 
     let mut group_clause = "time".to_string();
     let mut select_clause = match analytics_type {
         AnalyticsType::Advertiser { session } => {
             if channel_id.is_none() {
                 where_clauses.push(format!(
-                    "channel_id IN (SELECT id FROM channels WHERE creator = {})",
+                    "channel_id IN (SELECT id FROM channels WHERE creator = '{}')",
                     session.uid
                 ));
             }
@@ -125,7 +126,7 @@ pub async fn get_analytics(
             // ));
 
             format!(
-                "SUM(({}::numeric) 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 from event_aggregates", 
                 query.metric, interval
             )
         }

sentry/src/routes/analytics.rs

@@ -45,7 +45,7 @@ pub async fn analytics<A: Adapter>(
         _ => {
             // checks if /:id route param is present
             let cache_timeframe= match req.extensions().get::<RouteParams>() {
-                Some(param) => 600,
+                Some(_) => 600,
                 None => 300,
             };
             let response = process_analytics(req, app, AnalyticsType::Global).await?;