Merge pull request #229 from AdExNetwork/issue-9-get-event-aggregates

elpiel · web-flow · commit 72f5be4b3cb3 · 2020-01-03T14:54:09.000+02:00
Issue #9 GET Channel Events aggregates
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,12 +1,12 @@
 [workspace]
 
 members = [
-    "adapter",
     "primitives",
+    "adapter",
+    "adview-manager",
     "validator_worker",
     "sentry",
-    "adview-manager",
 ]
 
 [patch.crates-io]
-redis = {version = "0.13.1-alpha.0", git = "https://github.com/Marwes/redis-rs", branch = "async_await_api"}
+redis = {version = "0.13.1-alpha.0", git = "https://github.com/mitsuhiko/redis-rs"}
diff --git a/adapter/src/ethereum.rs b/adapter/src/ethereum.rs
@@ -564,7 +564,8 @@ mod test {
             )
             .expect("prep_db: failed to deserialize channel id"),
             // leader_account
-            creator: "0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39".to_string(),
+            creator: ValidatorId::try_from("Df08F82De32B8d460adbE8D72043E3a7e25A3B39")
+                .expect("should be valid ValidatorId"),
             deposit_asset: eth_checksum::checksum(&format!("{:?}", token_contract.address())),
             deposit_amount: 2_000.into(),
             valid_until: Utc::now() + Duration::days(2),
diff --git a/adapter/src/lib.rs b/adapter/src/lib.rs
@@ -94,8 +94,7 @@ impl TryFrom<&Channel> for EthereumChannel {
             .map(|v| &v.id)
             .collect::<Vec<_>>();
 
-        let creator = <[u8; 20]>::from_hex(&channel.creator[2..])
-            .map_err(|_| ChannelError::InvalidArgument("failed to parse creator".into()))?;
+        let creator = channel.creator.inner();
         let deposit_asset = <[u8; 20]>::from_hex(&channel.deposit_asset[2..])
             .map_err(|_| ChannelError::InvalidArgument("failed to parse deposit asset".into()))?;
 
diff --git a/primitives/Cargo.toml b/primitives/Cargo.toml
@@ -35,7 +35,7 @@ num-derive = "0.2"
 fake = { version = "^1.3", features = ["chrono"] }
 rand = "^0.6"
 # postgres feature
-postgres-types = { version = "0.1.0-alpha.2", optional = true }
+postgres-types = { version = "0.1.0", optional = true }
 bytes = { version = "0.5", optional = true }
 tokio-postgres = { version = "0.5.1", optional = true, features = ["with-chrono-0_4", "with-serde_json-1"] }
 # Futures
diff --git a/primitives/src/channel.rs b/primitives/src/channel.rs
@@ -49,7 +49,7 @@ impl FromHex for ChannelId {
 #[serde(rename_all = "camelCase")]
 pub struct Channel {
     pub id: ChannelId,
-    pub creator: String,
+    pub creator: ValidatorId,
     pub deposit_asset: String,
     pub deposit_amount: BigNum,
     #[serde(with = "ts_seconds")]
diff --git a/primitives/src/channel_validator.rs b/primitives/src/channel_validator.rs
@@ -2,6 +2,7 @@ use crate::channel::{Channel, ChannelError, SpecValidator, SpecValidators};
 use crate::config::Config;
 use crate::ValidatorId;
 use chrono::Utc;
+use std::cmp::PartialEq;
 
 pub trait ChannelValidator {
     fn is_channel_valid(
@@ -60,10 +61,10 @@ pub fn all_validators_listed(validators: &SpecValidators, whitelist: &[Validator
     }
 }
 
-pub fn creator_listed(channel: &Channel, whitelist: &[String]) -> bool {
+pub fn creator_listed(channel: &Channel, whitelist: &[ValidatorId]) -> bool {
     // if the list is empty, return true, as we don't have a whitelist to restrict us to
     // or if we have a list, check if it includes the `channel.creator`
-    whitelist.is_empty() || whitelist.iter().any(|allowed| allowed == &channel.creator)
+    whitelist.is_empty() || whitelist.iter().any(|allowed| allowed.eq(&channel.creator))
 }
 
 pub fn asset_listed(channel: &Channel, whitelist: &[String]) -> bool {
diff --git a/primitives/src/config.rs b/primitives/src/config.rs
@@ -33,7 +33,7 @@ pub struct Config {
     pub validator_tick_timeout: u32,
     pub ip_rate_limit: RateLimit,  // HashMap??
     pub sid_rate_limit: RateLimit, // HashMap ??
-    pub creators_whitelist: Vec<String>,
+    pub creators_whitelist: Vec<ValidatorId>,
     pub minimal_deposit: BigNum,
     pub minimal_fee: BigNum,
     pub token_address_whitelist: Vec<String>,
diff --git a/primitives/src/sentry.rs b/primitives/src/sentry.rs
@@ -62,6 +62,8 @@ pub struct Earner {
     pub promilles: u64,
 }
 
+pub type EarnerAddress = ValidatorId;
+
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct EventAggregate {
@@ -112,18 +114,30 @@ pub struct ValidatorMessageResponse {
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct EventAggregateResponse {
+    pub channel: Channel,
     pub events: Vec<EventAggregate>,
 }
 
 #[cfg(feature = "postgres")]
 mod postgres {
     use super::ValidatorMessage;
+    use crate::sentry::EventAggregate;
     use crate::validator::MessageTypes;
     use bytes::BytesMut;
     use postgres_types::{accepts, to_sql_checked, IsNull, Json, ToSql, Type};
     use std::error::Error;
     use tokio_postgres::Row;
 
+    impl From<&Row> for EventAggregate {
+        fn from(row: &Row) -> Self {
+            Self {
+                channel_id: row.get("channel_id"),
+                created: row.get("created"),
+                events: row.get::<_, Json<_>>("events").0,
+            }
+        }
+    }
+
     impl From<&Row> for ValidatorMessage {
         fn from(row: &Row) -> Self {
             Self {
diff --git a/primitives/src/util/tests/prep_db.rs b/primitives/src/util/tests/prep_db.rs
@@ -58,7 +58,7 @@ lazy_static! {
 
         Channel {
             id: ChannelId::from_hex("061d5e2a67d0a9a10f1c732bca12a676d83f79663a396f7d87b3e30b9b411088").expect("prep_db: failed to deserialize channel id"),
-            creator: "0x033ed90e0fec3f3ea1c9b005c724d704501e0196".to_string(),
+            creator: ValidatorId::try_from("033ed90e0fec3f3ea1c9b005c724d704501e0196").expect("Should be valid ValidatorId"),
             deposit_asset: "0x89d24A6b4CcB1B6fAA2625fE562bDD9a23260359".to_string(),
             deposit_amount: 1_000.into(),
             // UNIX timestamp for 2100-01-01
diff --git a/sentry/src/access.rs b/sentry/src/access.rs
@@ -5,6 +5,7 @@ use redis::aio::MultiplexedConnection;
 use primitives::event_submission::{RateLimit, Rule};
 use primitives::sentry::Event;
 use primitives::Channel;
+use std::cmp::PartialEq;
 
 use crate::Session;
 
@@ -54,7 +55,7 @@ pub async fn check_access(
 
     let default_rules = [
         Rule {
-            uids: Some(vec![channel.creator.clone()]),
+            uids: Some(vec![channel.creator.to_string()]),
             rate_limit: None,
         },
         Rule {
@@ -74,7 +75,7 @@ pub async fn check_access(
     let rules = allow_rules
         .iter()
         .filter(|r| match &r.uids {
-            Some(uids) => uids.iter().any(|uid| &session.uid == uid),
+            Some(uids) => uids.iter().any(|uid| uid.eq(&session.uid.to_string())),
             None => true,
         })
         .collect::<Vec<_>>();
@@ -107,16 +108,11 @@ async fn apply_rule(
     match &rule.rate_limit {
         Some(rate_limit) => {
             let key = if &rate_limit.limit_type == "sid" {
-                // @TODO: Is this really necessary?
-                if session.uid.is_empty() {
-                    Err("rateLimit: unauthenticated request".to_string())
-                } else {
-                    Ok(format!(
-                        "adexRateLimit:{}:{}",
-                        hex::encode(channel.id),
-                        session.uid
-                    ))
-                }
+                Ok(format!(
+                    "adexRateLimit:{}:{}",
+                    hex::encode(channel.id),
+                    session.uid
+                ))
             } else if &rate_limit.limit_type == "ip" {
                 if events.len() != 1 {
                     Err("rateLimit: only allows 1 event".to_string())
@@ -163,7 +159,7 @@ mod test {
     use primitives::config::configuration;
     use primitives::event_submission::{RateLimit, Rule};
     use primitives::sentry::Event;
-    use primitives::util::tests::prep_db::DUMMY_CHANNEL;
+    use primitives::util::tests::prep_db::{DUMMY_CHANNEL, IDS};
     use primitives::{Channel, Config, EventSubmission};
 
     use crate::db::redis_connection;
@@ -207,7 +203,7 @@ mod test {
 
         let session = Session {
             era: 0,
-            uid: "response".to_string(),
+            uid: IDS["follower"].clone(),
             ip: Default::default(),
         };
 
@@ -241,7 +237,7 @@ mod test {
 
         let session = Session {
             era: 0,
-            uid: "response".to_string(),
+            uid: IDS["follower"].clone(),
             ip: Default::default(),
         };
 
diff --git a/sentry/src/db.rs b/sentry/src/db.rs
@@ -9,9 +9,11 @@ use lazy_static::lazy_static;
 
 pub mod analytics;
 mod channel;
+mod event_aggregate;
 mod validator_message;
 
 pub use self::channel::*;
+pub use self::event_aggregate::*;
 pub use self::validator_message::*;
 
 pub type DbPool = Pool<PostgresConnectionManager<NoTls>>;
diff --git a/sentry/src/db/event_aggregate.rs b/sentry/src/db/event_aggregate.rs
@@ -0,0 +1,54 @@
+use crate::db::DbPool;
+use bb8::RunError;
+use bb8_postgres::tokio_postgres::types::ToSql;
+use chrono::{DateTime, Utc};
+use primitives::sentry::EventAggregate;
+use primitives::ValidatorId;
+
+pub async fn list_event_aggregates(
+    pool: &DbPool,
+    limit: u32,
+    from: &Option<ValidatorId>,
+    after: &Option<DateTime<Utc>>,
+) -> Result<Vec<EventAggregate>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    let (mut where_clauses, mut params) = (vec![], Vec::<&(dyn ToSql + Sync)>::new());
+    if let Some(from) = from {
+        let key_counts = format!(
+            "events->'IMPRESSION'->'eventPayouts'->'{}'",
+            from.to_string()
+        );
+        where_clauses.push(format!("{} IS NOT NULL", key_counts));
+    }
+    if let Some(after) = after {
+        params.push(after);
+        where_clauses.push(format!("created > {}", params.len()));
+    }
+
+    let event_aggregates = pool
+        .run(move |connection| {
+            async move {
+                let where_clause = if !where_clauses.is_empty() {
+                    format!("WHERE {}", where_clauses.join(" AND "))
+                } else {
+                    "".to_string()
+                };
+                let statement = format!("SELECT channel_id, created, events FROM event_aggregates {} ORDER BY created DESC LIMIT {}", where_clause, limit);
+                match connection.prepare(&statement).await {
+                    Ok(stmt) => {
+                        match connection.query(&stmt, params.as_slice()).await {
+                            Ok(rows) => {
+                                let event_aggregates = rows.iter().map(EventAggregate::from).collect();
+
+                                Ok((event_aggregates, connection))
+                            },
+                            Err(e) => Err((e, connection)),
+                        }
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await?;
+
+    Ok(event_aggregates)
+}
diff --git a/sentry/src/event_reducer.rs b/sentry/src/event_reducer.rs
@@ -13,7 +13,7 @@ pub(crate) fn reduce(channel: &Channel, initial_aggr: &mut EventAggregate, ev: &
             initial_aggr.events.insert("IMPRESSION".to_owned(), merge);
         }
         Event::Close => {
-            let creator = channel.creator.clone();
+            let creator = channel.creator.to_string();
             let close_event = AggregateEvents {
                 event_counts: Some(vec![(creator.clone(), 1.into())].into_iter().collect()),
                 event_payouts: vec![(creator, channel.deposit_amount.clone())]
@@ -36,14 +36,14 @@ fn merge_impression_ev(
     let event_count = impression
         .event_counts
         .get_or_insert_with(Default::default)
-        .entry(earner.into())
+        .entry(earner.to_string())
         .or_insert_with(|| 0.into());
 
     *event_count += &1.into();
 
     let event_payouts = impression
         .event_payouts
-        .entry(earner.into())
+        .entry(earner.to_string())
         .or_insert_with(|| 0.into());
     *event_payouts += &channel.spec.min_per_impression;
 
diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs
diff --git a/sentry/src/middleware/auth.rs b/sentry/src/middleware/auth.rs
diff --git a/sentry/src/middleware/event_aggregate.rs b/sentry/src/middleware/event_aggregate.rs
diff --git a/sentry/src/routes/analytics.rs b/sentry/src/routes/analytics.rs
diff --git a/sentry/src/routes/event_aggregate.rs b/sentry/src/routes/event_aggregate.rs

Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@ use crate::channel::{Channel, ChannelError, SpecValidator, SpecValidators};`
`2`	`2`	`use crate::config::Config;`
`3`	`3`	`use crate::ValidatorId;`
`4`	`4`	`use chrono::Utc;`
	`5`	`+use std::cmp::PartialEq;`
`5`	`6`
`6`	`7`	`pub trait ChannelValidator {`
`7`	`8`	`fn is_channel_valid(`
`@@ -60,10 +61,10 @@ pub fn all_validators_listed(validators: &SpecValidators, whitelist: &[Validator`
`60`	`61`	`}`
`61`	`62`	`}`
`62`	`63`
`63`		`-pub fn creator_listed(channel: &Channel, whitelist: &[String]) -> bool {`
	`64`	`+pub fn creator_listed(channel: &Channel, whitelist: &[ValidatorId]) -> bool {`
`64`	`65`	`// if the list is empty, return true, as we don't have a whitelist to restrict us to`
`65`	`66`	// or if we have a list, check if it includes the `channel.creator`
`66`		`- whitelist.is_empty() \|\| whitelist.iter().any(\|allowed\| allowed == &channel.creator)`
	`67`	`+ whitelist.is_empty() \|\| whitelist.iter().any(\|allowed\| allowed.eq(&channel.creator))`
`67`	`68`	`}`
`68`	`69`
`69`	`70`	`pub fn asset_listed(channel: &Channel, whitelist: &[String]) -> bool {`