add: approve_state, latest_new_state, last_heartbeats db queries

Author: samparsky

sentry/src/analytics_recorder.rs

@@ -12,7 +12,11 @@ fn get_payout(channel: &Channel, event: &Event) -> BigNum {
         Event::Impression { .. } => channel.spec.min_per_impression.clone(),
         Event::Click { .. } => {
             if let Some(pricing) = channel.spec.pricing_bounds.clone() {
-                pricing.click.min
+                if let Some(click) = pricing.click {
+                    click.min
+                } else {
+                    BigNum::from(0)
+                }
             } else {
                 BigNum::from(0)
             }

sentry/src/db.rs

@@ -84,9 +84,10 @@ pub async fn setup_migrations(environment: &str) {
 
     let mut migrations = vec![make_migration!("20190806011140_initial-tables")];
 
+
     if environment == "development" {
         // seeds database tables for testing
-        migrations.push(make_migration!("20190806011140_initial-tables/seed"))
+        migrations.push(make_migration!("20190806011140_initial-tables/seed"));
     }
 
     // Define Migrations
@@ -97,6 +98,18 @@ pub async fn setup_migrations(environment: &str) {
     // Reload config, ping the database for applied migrations
     let config = config.reload().expect("Should reload applied migrations");
 
+    if environment == "development" {
+        // delete all existing data  to make tests reproducible
+        Migrator::with_config(&config)
+            .all(true)
+            .direction(Direction::Down)
+            .swallow_completion(true)
+            .apply()
+            .expect("Applying migrations failed");
+    }
+    
+    let config = config.reload().expect("Should reload applied migrations");
+
     Migrator::with_config(&config)
         // set `swallow_completion` to `true`
         // so no error will be returned if all migrations have already been ran

sentry/src/db/event_aggregate.rs

@@ -5,11 +5,73 @@ use bb8_postgres::tokio_postgres::types::{ToSql, Type};
 use bb8_postgres::tokio_postgres::Error;
 use chrono::{DateTime, Utc};
 use futures::pin_mut;
-use primitives::sentry::EventAggregate;
+use primitives::sentry::{EventAggregate, LastApprovedResponse, ApproveStateValidatorMessage, NewStateValidatorMessage, HeartbeatValidatorMessage};
 use primitives::BigNum;
-use primitives::{ChannelId, ValidatorId};
+use primitives::{ChannelId, ValidatorId, Channel};
 use std::ops::Add;
 
+
+pub async fn approve_state( 
+    pool: &DbPool,
+    channel: &Channel
+) -> Result<Vec<ApproveStateValidatorMessage>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    /// select (from, msg, received) from validator_messages where channel_id = channel_id, from = from, msg ->> 'type'->>'ApproveState'
+    /// select (from, msg, received)  from validator_messages where channel_id = channel_id, from = from, msg ->> 'type'->>'NewState', msg ->> 'stateRoot'->>'0xx' 
+    pool
+        .run(move |connection| {
+            async move {
+                match connection.prepare("SELECT from, msg, received FROM validator_messages WHERE channel_id = $1 AND from = $2 AND msg ->> 'type' = 'ApproveState' ORDER BY received DESC").await {
+                    Ok(select) => match connection.query(&select, &[&channel.id, &channel.spec.validators.follower().id]).await {
+                        Ok(rows) => Ok((rows.iter().map(ApproveStateValidatorMessage::from).collect(), connection)),
+                        Err(e) => Err((e, connection)),
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+}
+
+pub async fn latest_new_state(
+    pool: &DbPool,
+    channel: &Channel,
+    state_root: &str,
+) -> Result<Option<NewStateValidatorMessage>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool
+    .run(move |connection| {
+        async move {
+            match connection.prepare("SELECT from, msg, received FROM validator_messages WHERE channel_id = $1 AND from = $2 AND msg ->> 'type' = 'NewState' AND msg->> 'stateRoot' = $3 ORDER BY received DESC LIMIT 1").await {
+                Ok(select) => match connection.query(&select, &[&channel.id, &channel.spec.validators.leader().id, &state_root]).await {
+                    Ok(rows) => Ok((rows.get(0).map(NewStateValidatorMessage::from), connection)),
+                    Err(e) => Err((e, connection)),
+                },
+                Err(e) => Err((e, connection)),
+            }
+        }
+    })
+    .await
+}
+
+pub async fn last_heartbeats(
+    pool: &DbPool,
+    channel_id: &ChannelId,
+    validator_id: &ValidatorId
+) -> Result<Vec<HeartbeatValidatorMessage>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool
+    .run(move |connection| {
+        async move {
+            match connection.prepare("SELECT from, msg, received FROM validator_messages WHERE channel_id = $1 AND from = $2 AND msg ->> 'type' = 'Heartbeat' ORDER BY received DESC LIMIT 2").await {
+                Ok(select) => match connection.query(&select, &[&channel_id, &validator_id]).await {
+                    Ok(rows) => Ok((rows.iter().map(HeartbeatValidatorMessage::from).collect(), connection)),
+                    Err(e) => Err((e, connection)),
+                },
+                Err(e) => Err((e, connection)),
+            }
+        }
+    })
+    .await
+}
+
 pub async fn list_event_aggregates(
     pool: &DbPool,
     channel_id: &ChannelId,
@@ -153,8 +215,8 @@ pub async fn insert_event_aggregate(
                 pin_mut!(writer);
                 for item in data {
                     if let Err(e) = writer.as_mut().write(&[&item.id, &item.created, &item.event_type, &item.event_count, &item.event_payout, &item.earner]).await {
-                            err = Some(e);
-                            break;
+                        err = Some(e);
+                        break;
                     }
                 }
 

sentry/src/event_aggregator.rs

@@ -156,6 +156,7 @@ impl EventAggregator {
         drop(channel_recorder);
 
         if aggr_throttle == 0 {
+            println!("in store");
             store(
                 &app.pool,
                 &channel_id,

sentry/src/routes/channel.rs

@@ -101,6 +101,8 @@ 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();
 
+    
+
     Ok(Response::builder()
         .header("Content-type", "application/json")
         .body(serde_json::to_string(&channel)?.into())
@@ -126,7 +128,9 @@ pub async fn insert_events<A: Adapter + 'static>(
 
     let into_body = req.into_body();
     let body = hyper::body::to_bytes(into_body).await?;
+    println!("deserializing request body");
     let request_body = serde_json::from_slice::<HashMap<String, Vec<Event>>>(&body)?;
+    println!("deserializing request body 2");
     let events = request_body
         .get("events")
         .ok_or_else(|| ResponseError::BadRequest("invalid request".to_string()))?;