Merge pull request #191 from AdExNetwork/issue-189-channel-load

Issue #189 Channel load

Author: samparsky

Cargo.lock

@@ -4,6 +4,9 @@ version = "0.1.0"
 authors = ["Lachezar Lechev <[email protected]>, Omidiora Samuel <[email protected]>"]
 edition = "2018"
 
+[features]
+postgres = ["postgres-types", "bytes"]
+
 [dependencies]
 # Futures
 futures-preview = { version = "=0.3.0-alpha.19", features = ["compat"]}
@@ -33,5 +36,8 @@ num-derive = "0.2"
 # Fixtures
 fake = { version = "^1.3", features = ["chrono"] }
 rand = { version = "^0.6" }
+# postgres feature
+postgres-types = {version = "0.1.0-alpha.1", optional = true}
+bytes = { version = "0.4.12", optional = true}
 # Other
 lazy_static = "1.4.0"

primitives/Cargo.toml

@@ -222,6 +222,55 @@ where
     serializer.serialize_str(&num.to_str_radix(10))
 }
 
+#[cfg(feature = "postgres")]
+pub mod postgres {
+    use super::BigNum;
+    use bytes::BytesMut;
+    use postgres_types::{FromSql, IsNull, ToSql, Type};
+    use std::error::Error;
+
+    impl<'a> FromSql<'a> for BigNum {
+        fn from_sql(ty: &Type, raw: &'a [u8]) -> Result<BigNum, Box<dyn Error + Sync + Send>> {
+            use std::convert::TryInto;
+
+            let str_slice = <&str as FromSql>::from_sql(ty, raw)?;
+
+            Ok(str_slice.try_into()?)
+        }
+
+        fn accepts(ty: &Type) -> bool {
+            match *ty {
+                Type::TEXT | Type::VARCHAR => true,
+                _ => false,
+            }
+        }
+    }
+
+    impl ToSql for BigNum {
+        fn to_sql(
+            &self,
+            ty: &Type,
+            w: &mut BytesMut,
+        ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
+            <String as ToSql>::to_sql(&self.0.to_string(), ty, w)
+        }
+
+        fn accepts(ty: &Type) -> bool {
+            match *ty {
+                Type::TEXT | Type::VARCHAR => true,
+                _ => false,
+            }
+        }
+
+        fn to_sql_checked(
+            &self,
+            ty: &Type,
+            out: &mut BytesMut,
+        ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
+            <String as ToSql>::to_sql_checked(&self.0.to_string(), ty, out)
+        }
+    }
+}
 #[cfg(test)]
 mod test {
     use super::*;

primitives/src/big_num.rs

@@ -9,12 +9,47 @@ use serde_hex::{SerHex, StrictPfx};
 use crate::big_num::BigNum;
 use crate::util::serde::ts_milliseconds_option;
 use crate::{AdUnit, EventSubmission, TargetingTag, ValidatorDesc, ValidatorId};
+use hex::{FromHex, FromHexError};
+use std::ops::Deref;
+
+#[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Copy, Clone)]
+#[serde(transparent)]
+pub struct ChannelId(#[serde(with = "SerHex::<StrictPfx>")] [u8; 32]);
+
+impl Deref for ChannelId {
+    type Target = [u8];
+
+    fn deref(&self) -> &[u8] {
+        &self.0
+    }
+}
+
+impl From<[u8; 32]> for ChannelId {
+    fn from(array: [u8; 32]) -> Self {
+        Self(array)
+    }
+}
+
+impl AsRef<[u8]> for ChannelId {
+    fn as_ref(&self) -> &[u8] {
+        &self.0
+    }
+}
+
+impl FromHex for ChannelId {
+    type Error = FromHexError;
+
+    fn from_hex<T: AsRef<[u8]>>(hex: T) -> Result<Self, Self::Error> {
+        let array = hex::FromHex::from_hex(hex.as_ref())?;
+
+        Ok(Self(array))
+    }
+}
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct Channel {
-    #[serde(with = "SerHex::<StrictPfx>")]
-    pub id: [u8; 32],
+    pub id: ChannelId,
     pub creator: String,
     pub deposit_asset: String,
     pub deposit_amount: BigNum,
@@ -158,3 +193,29 @@ impl Error for ChannelError {
         None
     }
 }
+
+#[cfg(feature = "postgres")]
+pub mod postgres {
+    use super::ChannelId;
+    use hex::FromHex;
+    use postgres_types::{FromSql, Type};
+    use std::error::Error;
+
+    impl<'a> FromSql<'a> for ChannelId {
+        fn from_sql(ty: &Type, raw: &'a [u8]) -> Result<Self, Box<dyn Error + Sync + Send>> {
+            let str_slice = <&str as FromSql>::from_sql(ty, raw)?;
+
+            // FromHex::from_hex for fixed-sized arrays will guard against the length of the string!
+            let id: [u8; 32] = <[u8; 32] as FromHex>::from_hex(str_slice)?;
+
+            Ok(ChannelId(id))
+        }
+
+        fn accepts(ty: &Type) -> bool {
+            match *ty {
+                Type::TEXT | Type::VARCHAR => true,
+                _ => false,
+            }
+        }
+    }
+}

primitives/src/channel.rs

@@ -29,7 +29,7 @@ pub mod validator;
 pub use self::ad_unit::AdUnit;
 pub use self::balances_map::BalancesMap;
 pub use self::big_num::BigNum;
-pub use self::channel::{Channel, ChannelSpec, SpecValidator, SpecValidators};
+pub use self::channel::{Channel, ChannelId, ChannelSpec, SpecValidator, SpecValidators};
 pub use self::config::Config;
 pub use self::event_submission::EventSubmission;
 pub use self::targeting_tag::TargetingTag;

primitives/src/lib.rs

@@ -1,8 +1,7 @@
 use crate::validator::{ApproveState, Heartbeat, MessageTypes, NewState};
-use crate::{BigNum, Channel};
+use crate::{BigNum, Channel, ChannelId};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
-use serde_hex::{SerHex, StrictPfx};
 use std::collections::HashMap;
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -63,8 +62,7 @@ pub struct Earner {
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct EventAggregate {
-    #[serde(with = "SerHex::<StrictPfx>")]
-    pub channel_id: [u8; 32],
+    pub channel_id: ChannelId,
     pub created: DateTime<Utc>,
     pub events: HashMap<String, AggregateEvents>,
 }

primitives/src/sentry.rs

@@ -1,5 +1,6 @@
 use crate::{
-    BigNum, Channel, ChannelSpec, EventSubmission, SpecValidators, ValidatorDesc, ValidatorId,
+    BigNum, Channel, ChannelId, ChannelSpec, EventSubmission, SpecValidators, ValidatorDesc,
+    ValidatorId,
 };
 use chrono::{TimeZone, Utc};
 use fake::faker::{Faker, Number};
@@ -56,7 +57,7 @@ lazy_static! {
         let nonce = BigNum::from(<Faker as Number>::between(100_000_000, 999_999_999));
 
         Channel {
-            id: <[u8; 32]>::from_hex("061d5e2a67d0a9a10f1c732bca12a676d83f79663a396f7d87b3e30b9b411088").expect("prep_db: failed to deserialize channel id"),
+            id: ChannelId::from_hex("061d5e2a67d0a9a10f1c732bca12a676d83f79663a396f7d87b3e30b9b411088").expect("prep_db: failed to deserialize channel id"),
             creator: "0x033ed90e0fec3f3ea1c9b005c724d704501e0196".to_string(),
             deposit_asset: "0x89d24A6b4CcB1B6fAA2625fE562bDD9a23260359".to_string(),
             deposit_amount: 1_000.into(),

primitives/src/util/tests/prep_db.rs

@@ -8,7 +8,7 @@ edition = "2018"
 # Futures
 futures-preview = { version = "=0.3.0-alpha.19", features = ["compat"]}
 # Primitives
-primitives = {path = "../primitives"}
+primitives = {path = "../primitives", features = ["postgres"]}
 adapter = { version = "0.1", path = "../adapter" }
 chrono = { version = "0.4", features = ["serde"] }
 hex = "0.3.2"
@@ -21,7 +21,7 @@ regex = "1"
 # Database
 redis = {version = "0.13.1-alpha.0", features = ["tokio-executor"]}
 bb8 = {git = "https://github.com/djc/bb8", branch = "async-await"}
-bb8-postgres = {git = "https://github.com/djc/bb8", branch = "async-await"}
+bb8-postgres = {git = "https://github.com/djc/bb8", branch = "async-await", features = ["with-chrono-0_4", "with-serde_json-1"]}
 # Logger
 slog = { version = "^2.2.3" , features = ["max_level_trace"] }
 # Serde

sentry/Cargo.toml

@@ -2,12 +2,12 @@ use redis::aio::MultiplexedConnection;
 use redis::RedisError;
 
 use bb8::Pool;
-use bb8_postgres::{
-    tokio_postgres::{Error as PostgresError, NoTls},
-    PostgresConnectionManager,
-};
+use bb8_postgres::tokio_postgres::NoTls;
+use bb8_postgres::PostgresConnectionManager;
 use lazy_static::lazy_static;
 
+pub type DbPool = Pool<PostgresConnectionManager<NoTls>>;
+
 lazy_static! {
     static ref REDIS_URL: String =
         std::env::var("REDIS_URL").unwrap_or_else(|_| String::from("redis://127.0.0.1:6379"));
@@ -20,8 +20,7 @@ pub async fn redis_connection() -> Result<MultiplexedConnection, RedisError> {
     client.get_multiplexed_tokio_connection().await
 }
 
-pub async fn postgres_connection() -> Result<Pool<PostgresConnectionManager<NoTls>>, PostgresError>
-{
+pub async fn postgres_connection() -> Result<DbPool, bb8_postgres::tokio_postgres::Error> {
     let pg_mgr = PostgresConnectionManager::new_from_stringlike(POSTGRES_URL.as_str(), NoTls)?;
 
     Pool::builder().build(pg_mgr).await

sentry/src/db.rs

@@ -1,10 +1,9 @@
 #![deny(clippy::all)]
 #![deny(rust_2018_idioms)]
 
+use crate::db::DbPool;
 use crate::middleware::auth;
 use crate::middleware::cors::{cors, Cors};
-use bb8::Pool;
-use bb8_postgres::{tokio_postgres::NoTls, PostgresConnectionManager};
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Error, Method, Request, Response, Server, StatusCode};
 use primitives::adapter::Adapter;
@@ -14,8 +13,10 @@ use slog::{error, info, Logger};
 
 pub mod middleware {
     pub mod auth;
+    pub mod channel;
     pub mod cors;
 }
+
 pub mod routes {
     pub mod channel;
     pub mod cfg {
@@ -43,7 +44,7 @@ pub struct Application<A: Adapter> {
     adapter: A,
     logger: Logger,
     redis: MultiplexedConnection,
-    _postgres: Pool<PostgresConnectionManager<NoTls>>,
+    pool: DbPool,
     _clustered: bool,
     port: u16,
     config: Config,
@@ -55,7 +56,7 @@ impl<A: Adapter + 'static> Application<A> {
         config: Config,
         logger: Logger,
         redis: MultiplexedConnection,
-        postgres: Pool<PostgresConnectionManager<NoTls>>,
+        pool: DbPool,
         clustered: bool,
         port: u16,
     ) -> Self {
@@ -64,7 +65,7 @@ impl<A: Adapter + 'static> Application<A> {
             config,
             logger,
             redis,
-            _postgres: postgres,
+            pool,
             _clustered: clustered,
             port,
         }
@@ -79,17 +80,19 @@ impl<A: Adapter + 'static> Application<A> {
             let adapter_config = (self.adapter.clone(), self.config.clone());
             let redis = self.redis.clone();
             let logger = self.logger.clone();
+            let pool = self.pool.clone();
             async move {
                 Ok::<_, Error>(service_fn(move |req| {
                     let adapter_config = adapter_config.clone();
                     let redis = redis.clone();
                     let logger = logger.clone();
+                    let pool = pool.clone();
                     async move {
                         Ok::<_, Error>(
                             handle_routing(
                                 req,
                                 (&adapter_config.0, &adapter_config.1),
-                                redis,
+                                (pool, redis),
                                 &logger,
                             )
                             .await,
@@ -125,7 +128,7 @@ where
 async fn handle_routing(
     req: Request<Body>,
     (adapter, config): (&impl Adapter, &Config),
-    redis: MultiplexedConnection,
+    (pool, redis): (DbPool, MultiplexedConnection),
     logger: &Logger,
 ) -> Response<Body> {
     let headers = match cors(&req) {
@@ -147,7 +150,7 @@ async fn handle_routing(
     let mut response = match (req.uri().path(), req.method()) {
         ("/cfg", &Method::GET) => crate::routes::cfg::return_config(&config),
         (route, _) if route.starts_with("/channel") => {
-            crate::routes::channel::handle_channel_routes(req, adapter).await
+            crate::routes::channel::handle_channel_routes(req, (&pool, adapter)).await
         }
         _ => Err(ResponseError::NotFound),
     }
@@ -172,7 +175,8 @@ pub fn not_found() -> Response<Body> {
     response
 }
 
-pub fn bad_request(_: Box<dyn std::error::Error>) -> Response<Body> {
+pub fn bad_request(err: Box<dyn std::error::Error>) -> Response<Body> {
+    println!("{:#?}", err);
     let body = Body::from("Bad Request: try again later");
     let mut response = Response::new(body);
     let status = response.status_mut();