fix: /last-approved route

Author: samparsky

primitives/src/big_num.rs

@@ -10,7 +10,7 @@ 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(

sentry/src/analytics_recorder.rs

@@ -17,9 +17,9 @@ pub fn get_payout(channel: &Channel, event: &Event) -> BigNum {
                 }
             }
 
-            BigNum::from(0)
+            Default::default()
         }
-        _ => BigNum::from(0),
+        _ => Default::default(),
     }
 }
 

sentry/src/db/analytics.rs

@@ -7,7 +7,7 @@ use primitives::analytics::{AnalyticsData, AnalyticsQuery, ANALYTICS_QUERY_LIMIT
 use primitives::sentry::{AdvancedAnalyticsResponse, ChannelReport, PublisherReport};
 use bb8_postgres::tokio_postgres::types::{ToSql};
 use primitives::{ChannelId, ValidatorId};
-use redis;
+use redis::cmd;
 use redis::aio::MultiplexedConnection;
 use std::collections::HashMap;
 use std::error::Error;
@@ -157,15 +157,15 @@ fn get_time_frame(timeframe: &str) -> (i64, i64) {
 }
 
 async fn stat_pair(
-    conn: MultiplexedConnection,
+    mut conn: MultiplexedConnection,
     key: &str,
 ) -> Result<HashMap<String, f64>, Box<dyn Error>> {
-    let data = redis::cmd("ZRANGE")
+    let data = cmd("ZRANGE")
         .arg(key)
         .arg(0 as u64)
         .arg(-1 as i64)
         .arg("WITHSCORES")
-        .query_async::<_, Vec<String>>(&mut conn.clone())
+        .query_async::<_, Vec<String>>(&mut conn)
         .await?;
 
     Ok(data

sentry/src/event_aggregator.rs

@@ -21,8 +21,7 @@ use std::time::Duration;
 use tokio::time::delay_for;
 
 lazy_static! {
-    pub static ref ANALYTICS_RECORDER: String =
-        env::var("ANALYTICS_RECORDER").unwrap_or_else(|_| "false".to_string());
+    pub static ref ANALYTICS_RECORDER: Option<String> = env::var("ANALYTICS_RECORDER").ok();
 }
 
 #[derive(Debug)]
@@ -140,7 +139,7 @@ impl EventAggregator {
             .iter()
             .for_each(|ev| event_reducer::reduce(&record.channel, &mut record.aggregate, ev));
 
-        if ANALYTICS_RECORDER.ne(&"false".to_string()) {
+        if ANALYTICS_RECORDER.is_some() {
             let logger = app.logger.clone();
             tokio::spawn(analytics_recorder::record(
                 redis.clone(),

sentry/src/routes/analytics.rs

@@ -89,9 +89,8 @@ pub async fn process_analytics<A: Adapter>(
 
     let segment_channel = query
         .segment_by_channel
-        .clone()
-        .map(|_| true)
-        .unwrap_or_else(|| false);
+        .is_some();
+        
     let limit = query.limit;
 
     let aggr = get_analytics(

sentry/src/routes/channel.rs

@@ -103,42 +103,30 @@ pub async fn last_approved<A: Adapter>(
     let channel_id = ChannelId::from_hex(route_params.index(0))?;
     let channel = get_channel_by_id(&app.pool, &channel_id).await?.unwrap();
 
-    let approve_state = lastest_approve_state(&app.pool, &channel).await?;
-    if approve_state.is_none() {
-        return Ok(Response::builder()
-            .header("Content-type", "application/json")
-            .body(
-                serde_json::to_string(&LastApprovedResponse {
-                    last_approved: None,
-                    heartbeats: None,
-                })?
-                .into(),
-            )
-            .unwrap());
-    }
+    let default_response = Response::builder()
+        .header("Content-type", "application/json")
+        .body(
+            serde_json::to_string(&LastApprovedResponse {
+                last_approved: None,
+                heartbeats: None,
+            })?
+            .into(),
+        )
+        .expect("should build response");
+
+    let approve_state = match lastest_approve_state(&app.pool, &channel).await? {
+        Some(approve_state) => approve_state,
+        None => return Ok(default_response)
+    };
 
-    let state_root = match approve_state
-        .as_ref()
-        .expect("value should be present")
-        .msg
-        .clone()
-    {
+    let state_root = match approve_state.msg.clone() {
         MessageTypes::ApproveState(approve_state) => approve_state.state_root,
         _ => return Err(ResponseError::BadRequest("invalid request".to_string())),
     };
 
     let new_state = latest_new_state(&app.pool, &channel, &state_root).await?;
     if new_state.is_none() {
-        return Ok(Response::builder()
-            .header("Content-type", "application/json")
-            .body(
-                serde_json::to_string(&LastApprovedResponse {
-                    last_approved: None,
-                    heartbeats: None,
-                })?
-                .into(),
-            )
-            .unwrap());
+        return Ok(default_response);
     }
 
     let query = serde_urlencoded::from_str::<LastApprovedQuery>(&req.uri().query().unwrap_or(""))?;
@@ -159,10 +147,10 @@ pub async fn last_approved<A: Adapter>(
     Ok(Response::builder()
         .header("Content-type", "application/json")
         .body(
-            serde_json::to_string(&&LastApprovedResponse {
+            serde_json::to_string(&LastApprovedResponse {
                 last_approved: Some(LastApproved {
                     new_state,
-                    approve_state,
+                    approve_state: Some(approve_state),
                 }),
                 heartbeats,
             })?

validator.env.dist

@@ -4,5 +4,3 @@ DUMMY_IDENTITY=ce07CbB7e054514D590a0262C93070D838bFBA2e
 # for Ethereum adapter
 #KEYSTORE_FILE="/app/keystore.json"
 #KEYSTORE_PWD="adexvalidator"
-
-

validator_worker/src/follower.rs

@@ -82,7 +82,6 @@ async fn on_new_state<'a, A: Adapter + 'static>(
         },
         _ => Default::default(),
     };
-    // .map_or(Default::default(), |new_state| new_state.msg.balances);
 
     if !is_valid_transition(&iface.channel, &prev_balances, &proposed_balances) {
         return Ok(on_error(&iface, &new_state, InvalidNewState::Transition).await);