sentry - returning a Channel

elpiel · elpiel · commit f53468a9aa80 · 2019-11-14T12:35:04.000+02:00
diff --git a/sentry/src/lib.rs b/sentry/src/lib.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;
@@ -45,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,
@@ -57,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 {
@@ -66,7 +65,7 @@ impl<A: Adapter + 'static> Application<A> {
             config,
             logger,
             redis,
-            _postgres: postgres,
+            pool,
             _clustered: clustered,
             port,
         }
@@ -81,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,
@@ -127,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) {
@@ -149,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),
     }
@@ -174,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();
diff --git a/sentry/src/middleware/channel.rs b/sentry/src/middleware/channel.rs
@@ -1,34 +1,45 @@
-use primitives::postgres::field::{BigNumPg, ChannelIdPg};
 use crate::db::DbPool;
+use crate::ResponseError;
+use bb8::RunError;
 use bb8_postgres::tokio_postgres::{types::Json, Row};
 use hyper::{Body, Request};
 use primitives::channel::ChannelId;
+use primitives::postgres::field::{BigNumPg, ChannelIdPg};
 use primitives::{Channel, ChannelSpec};
-use std::error;
 
 pub async fn channel_load(
     mut req: Request<Body>,
-    (pool, id): (DbPool, &ChannelId),
-) -> Result<Request<Body>, Box<dyn error::Error>> {
-    let channel_id = hex::encode(id);
+    (pool, id): (&DbPool, &ChannelId),
+) -> Result<Request<Body>, ResponseError> {
+    let channel = get_channel(pool, id)
+        .await?
+        .ok_or(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>> {
+    let id = hex::encode(id);
 
-    let channel = pool
+    pool
         .run(move |connection| {
             async move {
-                match connection.prepare("SELECT channel_id, creator, deposit_asset, deposit_amount, valid_until, spec FROM channels WHERE channel_id = $1").await {
-                    Ok(select) => match connection.query_one(&select, &[&channel_id]).await {
-                            Ok(row) => Ok((channel_map(&row), connection)),
-                            Err(e) => Err((e, connection)),
-                        },
+                match connection.prepare("SELECT channel_id, creator, deposit_asset, deposit_amount, valid_until, spec FROM channels WHERE channel_id = $1 LIMIT 1").await {
+                    Ok(select) => match connection.query(&select, &[&id]).await {
+                        Ok(results) => Ok((results.get(0).map(|row| channel_map(&row)), connection)),
+                        Err(e) => Err((e, connection)),
+                    },
                     Err(e) => Err((e, connection)),
                 }
             }
         })
-        .await?;
-
-    req.extensions_mut().insert(channel);
-
-    Ok(req)
+        .await
 }
 
 fn channel_map(row: &Row) -> Channel {
diff --git a/sentry/src/routes/channel.rs b/sentry/src/routes/channel.rs
@@ -2,14 +2,23 @@ use futures::TryStreamExt;
 use hyper::{Body, Method, Request, Response};
 
 use primitives::adapter::Adapter;
-use primitives::Channel;
+use primitives::{Channel, ChannelId};
 
 use self::channel_list::ChannelListQuery;
+use crate::db::DbPool;
+use crate::middleware::channel::get_channel;
 use crate::ResponseError;
+use hex::FromHex;
+use lazy_static::lazy_static;
+use regex::Regex;
 
+lazy_static! {
+    static ref CHANNEL_GET_BY_ID: Regex =
+        Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
+}
 pub async fn handle_channel_routes(
     req: Request<Body>,
-    adapter: &impl Adapter,
+    (pool, adapter): (&DbPool, &impl Adapter),
 ) -> Result<Response<Body>, ResponseError> {
     // Channel Creates
     if req.uri().path() == "/channel" && req.method() == Method::POST {
@@ -24,6 +33,18 @@ pub async fn handle_channel_routes(
         return Ok(Response::builder().status(200).body(body).unwrap());
     }
 
+    // @TODO: This is only a PoC, see https://github.com/AdExNetwork/adex-validator-stack-rust/issues/9
+    if let (Some(caps), &Method::GET) = (CHANNEL_GET_BY_ID.captures(req.uri().path()), req.method())
+    {
+        let channel_id = ChannelId::from_hex(caps.get(1).unwrap().as_str())?;
+        let channel = get_channel(&pool, &channel_id).await?.unwrap();
+
+        return Ok(Response::builder()
+            .header("Content-type", "application/json")
+            .body(serde_json::to_string(&channel)?.into())
+            .unwrap());
+    }
+
     // Channel List
     if req.uri().path().starts_with("/channel/list") {
         // @TODO: Get from Config
