Merge pull request #198 from AdExNetwork/issue-189-channel-if-functions

Issue #189 channel if functions

Author: samparsky

adapter/src/dummy.rs

@@ -64,7 +64,7 @@ impl Adapter for DummyAdapter {
     }
 
     fn validate_channel(&self, channel: &Channel) -> AdapterResult<bool> {
-        match DummyAdapter::is_channel_valid(&self.config, channel) {
+        match DummyAdapter::is_channel_valid(&self.config, self.whoami(), channel) {
             Ok(_) => Ok(true),
             Err(e) => Err(AdapterError::InvalidChannel(e.to_string())),
         }

adapter/src/ethereum.rs

@@ -123,7 +123,7 @@ impl Adapter for EthereumAdapter {
 
     fn validate_channel(&self, channel: &Channel) -> AdapterResult<bool> {
         // check if channel is valid
-        if let Err(e) = EthereumAdapter::is_channel_valid(&self.config, channel) {
+        if let Err(e) = EthereumAdapter::is_channel_valid(&self.config, self.whoami(), channel) {
             return Err(AdapterError::InvalidChannel(e.to_string()));
         }
 

primitives/src/channel.rs

@@ -40,7 +40,7 @@ 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())?;
+        let array = hex::FromHex::from_hex(hex)?;
 
         Ok(Self(array))
     }
@@ -200,24 +200,9 @@ pub mod postgres {
     use super::{Channel, ChannelSpec};
     use bytes::BytesMut;
     use hex::FromHex;
-    use postgres_types::{FromSql, IsNull, ToSql, Type};
+    use postgres_types::{accepts, to_sql_checked, FromSql, IsNull, Json, ToSql, Type};
     use std::error::Error;
-    use tokio_postgres::{types::Json, Row};
-
-    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)?;
-
-            Ok(ChannelId::from_hex(&str_slice[2..])?)
-        }
-
-        fn accepts(ty: &Type) -> bool {
-            match *ty {
-                Type::TEXT | Type::VARCHAR => true,
-                _ => false,
-            }
-        }
-    }
+    use tokio_postgres::Row;
 
     impl From<&Row> for Channel {
         fn from(row: &Row) -> Self {
@@ -232,6 +217,16 @@ pub mod postgres {
         }
     }
 
+    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)?;
+
+            Ok(ChannelId::from_hex(&str_slice[2..])?)
+        }
+
+        accepts!(TEXT, VARCHAR);
+    }
+
     impl ToSql for ChannelId {
         fn to_sql(
             &self,
@@ -247,14 +242,19 @@ pub mod postgres {
             <String as ToSql>::accepts(ty)
         }
 
-        fn to_sql_checked(
+        to_sql_checked!();
+    }
+
+    impl ToSql for ChannelSpec {
+        fn to_sql(
             &self,
             ty: &Type,
-            out: &mut BytesMut,
+            w: &mut BytesMut,
         ) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
-            let string = format!("0x{}", hex::encode(self));
-
-            <String as ToSql>::to_sql_checked(&string, ty, out)
+            Json(self).to_sql(ty, w)
         }
+
+        accepts!(JSONB);
+        to_sql_checked!();
     }
 }

primitives/src/channel_validator.rs

@@ -2,15 +2,14 @@ use crate::channel::{Channel, ChannelError, SpecValidator, SpecValidators};
 use crate::config::Config;
 use crate::ValidatorId;
 use chrono::Utc;
-use std::convert::TryFrom;
 
 pub trait ChannelValidator {
-    fn is_channel_valid(config: &Config, channel: &Channel) -> Result<(), ChannelError> {
-        let identity = &config.clone().identity.unwrap_or_else(|| "".to_string());
-        let validator_identity = ValidatorId::try_from(identity).map_err(|_| {
-            ChannelError::InvalidArgument("Failed to deserialize identity".to_string())
-        })?;
-        let adapter_channel_validator = match channel.spec.validators.find(&validator_identity) {
+    fn is_channel_valid(
+        config: &Config,
+        validator_identity: &ValidatorId,
+        channel: &Channel,
+    ) -> Result<(), ChannelError> {
+        let adapter_channel_validator = match channel.spec.validators.find(validator_identity) {
             // check if the channel validators include our adapter identity
             SpecValidator::None => return Err(ChannelError::AdapterNotIncluded),
             SpecValidator::Leader(validator) | SpecValidator::Follower(validator) => validator,

primitives/src/config.rs

@@ -17,7 +17,6 @@ lazy_static! {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all(serialize = "SCREAMING_SNAKE_CASE"))]
 pub struct Config {
-    pub identity: Option<String>, // should not be here maybe?
     pub max_channels: u32,
     pub wait_time: u32,
     pub aggr_throttle: u32,

sentry/src/chain.rs

@@ -1,24 +1,27 @@
-use crate::ResponseError;
+use crate::{Application, ResponseError};
 use hyper::{Body, Request};
+use primitives::adapter::Adapter;
 use std::future::Future;
 
 // chain middleware function calls
 //
 // function signature
 // fn middleware(mut req: Request) -> Result<Request, ResponseError>
 
-pub async fn chain<M, MF>(
+pub async fn chain<'a, A, M, MF>(
     req: Request<Body>,
+    app: &'a Application<A>,
     middlewares: Vec<M>,
 ) -> Result<Request<Body>, ResponseError>
 where
+    A: Adapter,
     MF: Future<Output = Result<Request<Body>, ResponseError>> + Send,
-    M: FnMut(Request<Body>) -> MF,
+    M: FnMut(Request<Body>, &'a Application<A>) -> MF,
 {
     let mut req = Ok(req);
 
     for mut mw in middlewares.into_iter() {
-        match mw(req.unwrap()).await {
+        match mw(req.unwrap(), app).await {
             Ok(r) => {
                 req = Ok(r);
             }

sentry/src/db.rs

@@ -7,6 +7,9 @@ use std::env;
 
 use lazy_static::lazy_static;
 
+mod channel;
+pub use self::channel::*;
+
 pub type DbPool = Pool<PostgresConnectionManager<NoTls>>;
 
 lazy_static! {

sentry/src/db/channel.rs

@@ -0,0 +1,69 @@
+use crate::db::DbPool;
+use bb8::RunError;
+use primitives::{Channel, ChannelId, ValidatorId};
+use std::str::FromStr;
+
+pub async fn get_channel_by_id(
+    pool: &DbPool,
+    id: &ChannelId,
+) -> Result<Option<Channel>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool
+        .run(move |connection| {
+            async move {
+                match connection.prepare("SELECT id, creator, deposit_asset, deposit_amount, valid_until, spec FROM channels WHERE id = $1 LIMIT 1").await {
+                    Ok(select) => match connection.query(&select, &[&id]).await {
+                        Ok(results) => Ok((results.get(0).map(Channel::from), connection)),
+                        Err(e) => Err((e, connection)),
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+}
+
+pub async fn get_channel_by_id_and_validator(
+    pool: &DbPool,
+    id: &ChannelId,
+    validator_id: &ValidatorId,
+) -> Result<Option<Channel>, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool
+        .run(move |connection| {
+            async move {
+                let validator = serde_json::Value::from_str(&format!(r#"[{{"id": "{}"}}]"#, validator_id)).expect("Not a valid json");
+                let query = "SELECT id, creator, deposit_asset, deposit_amount, valid_until, spec FROM channels WHERE id = $1 AND spec->'validators' @> $2 LIMIT 1";
+                match connection.prepare(query).await {
+                    Ok(select) => {
+                        match connection.query(&select, &[&id, &validator]).await {
+                            Ok(results) => Ok((results.get(0).map(Channel::from), connection)),
+                            Err(e) => Err((e, connection)),
+                        }
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+}
+
+pub async fn insert_channel(
+    pool: &DbPool,
+    channel: &Channel,
+) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool
+        .run(move |connection| {
+            async move {
+                match connection.prepare("INSERT INTO channels (id, creator, deposit_asset, deposit_amount, valid_until, spec) values ($1, $2, $3, $4, $5, $6)").await {
+                    Ok(stmt) => match connection.execute(&stmt, &[&channel.id, &channel.creator, &channel.deposit_asset, &channel.deposit_amount, &channel.valid_until, &channel.spec]).await {
+                        Ok(row) => {
+                            let inserted = row == 1;
+                            Ok((inserted, connection))
+                        },
+                        Err(e) => Err((e, connection)),
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+}

sentry/src/lib.rs

@@ -4,7 +4,9 @@
 use crate::chain::chain;
 use crate::db::DbPool;
 use crate::middleware::auth;
+use crate::middleware::channel::channel_load;
 use crate::middleware::cors::{cors, Cors};
+use crate::routes::channel::channel_status;
 use hyper::{Body, Method, Request, Response, StatusCode};
 use lazy_static::lazy_static;
 use primitives::adapter::Adapter;
@@ -40,7 +42,10 @@ lazy_static! {
     // @TODO define other regex routes
 }
 
-async fn config_middleware(req: Request<Body>) -> Result<Request<Body>, ResponseError> {
+async fn config_middleware<A: Adapter>(
+    req: Request<Body>,
+    _: &Application<A>,
+) -> Result<Request<Body>, ResponseError> {
     Ok(req)
 }
 
@@ -121,14 +126,30 @@ impl<A: Adapter + 'static> Application<A> {
 
                     // example with middleware
                     // @TODO remove later
-                    let req = match chain(req, vec![config_middleware]).await {
+                    let req = match chain(req, &self, vec![config_middleware]).await {
                         Ok(req) => req,
                         Err(error) => {
                             return map_response_error(error);
                         }
                     };
 
                     last_approved(req, &self).await
+                } else if let (Some(caps), &Method::GET) =
+                    (CHANNEL_STATUS_BY_CHANNEL_ID.captures(route), method)
+                {
+                    let param = RouteParams(vec![caps
+                        .get(1)
+                        .map_or("".to_string(), |m| m.as_str().to_string())]);
+                    req.extensions_mut().insert(param);
+
+                    let req = match chain(req, &self, vec![channel_load]).await {
+                        Ok(req) => req,
+                        Err(error) => {
+                            return map_response_error(error);
+                        }
+                    };
+
+                    channel_status(req, &self).await
                 } else {
                     Err(ResponseError::NotFound)
                 }

sentry/src/middleware/channel.rs

@@ -1,38 +1,61 @@
-use crate::db::DbPool;
-use crate::ResponseError;
-use bb8::RunError;
+use crate::db::{get_channel_by_id, get_channel_by_id_and_validator};
+use crate::{Application, ResponseError, RouteParams};
+use hex::FromHex;
 use hyper::{Body, Request};
-use primitives::{Channel, ChannelId};
+use primitives::adapter::Adapter;
+use primitives::{ChannelId, ValidatorId};
+use std::convert::TryFrom;
 
-pub async fn channel_load(
+/// channel_load & channel_if_exist
+pub async fn channel_load<A: Adapter>(
     mut req: Request<Body>,
-    (pool, id): (&DbPool, &ChannelId),
+    app: &Application<A>,
 ) -> Result<Request<Body>, ResponseError> {
-    let channel = get_channel(pool, id)
+    let id = req
+        .extensions()
+        .get::<RouteParams>()
+        .ok_or_else(|| ResponseError::BadRequest("Route params not found".to_string()))?
+        .get(0)
+        .ok_or_else(|| ResponseError::BadRequest("No id".to_string()))?;
+
+    let channel_id = ChannelId::from_hex(id)
+        .map_err(|_| ResponseError::BadRequest("Wrong Channel Id".to_string()))?;
+    let channel = get_channel_by_id(&app.pool, &channel_id)
         .await?
-        .ok_or(ResponseError::NotFound)?;
+        .ok_or_else(|| ResponseError::NotFound)?;
 
     req.extensions_mut().insert(channel);
 
     Ok(req)
 }
 
-// @TODO: Maybe move this to more generic place?
-pub async fn get_channel(
-    pool: &DbPool,
-    id: &ChannelId,
-) -> Result<Option<Channel>, RunError<bb8_postgres::tokio_postgres::Error>> {
-    pool
-        .run(move |connection| {
-            async move {
-                match connection.prepare("SELECT id, creator, deposit_asset, deposit_amount, valid_until, spec FROM channels WHERE id = $1 LIMIT 1").await {
-                    Ok(select) => match connection.query(&select, &[&id]).await {
-                        Ok(results) => Ok(( results.get(0).map(Channel::from) , connection)),
-                        Err(e) => Err((e, connection)),
-                    },
-                    Err(e) => Err((e, connection)),
-                }
-            }
-        })
-        .await
+pub async fn channel_if_active<A: Adapter>(
+    mut req: Request<Body>,
+    app: &Application<A>,
+) -> Result<Request<Body>, ResponseError> {
+    let route_params = req
+        .extensions()
+        .get::<RouteParams>()
+        .ok_or_else(|| ResponseError::BadRequest("Route params not found".to_string()))?;
+
+    let id = route_params
+        .get(0)
+        .ok_or_else(|| ResponseError::BadRequest("No id".to_string()))?;
+
+    let channel_id = ChannelId::from_hex(id)
+        .map_err(|_| ResponseError::BadRequest("Wrong Channel Id".to_string()))?;
+
+    let validator_id = route_params
+        .get(1)
+        .ok_or_else(|| ResponseError::BadRequest("No Validator Id".to_string()))?;
+    let validator_id = ValidatorId::try_from(&validator_id)
+        .map_err(|_| ResponseError::BadRequest("Wrong Validator Id".to_string()))?;
+
+    let channel = get_channel_by_id_and_validator(&app.pool, &channel_id, &validator_id)
+        .await?
+        .ok_or_else(|| ResponseError::NotFound)?;
+
+    req.extensions_mut().insert(channel);
+
+    Ok(req)
 }