Merge pull request #203 from AdExNetwork/issue-9-channel-list

Issue #9 channel list

Author: samparsky

primitives/src/sentry.rs

@@ -75,9 +75,9 @@ pub struct AggregateEvents {
     pub event_payouts: HashMap<String, BigNum>,
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct ChannelAllResponse {
+pub struct ChannelListResponse {
     pub channels: Vec<Channel>,
     pub total_pages: u64,
 }

sentry/migrations/20190806011140_initial_tables/down.sql renamed to sentry/migrations/20190806011140_initial-tables/down.sql

@@ -3,37 +3,33 @@ CREATE TABLE channels
     id             VARCHAR(66)              NOT NULL,
     creator        VARCHAR(255)             NOT NULL,
     deposit_asset  VARCHAR(42)              NOT NULL,
-    deposit_amount VARCHAR(255)             NOT NULL, -- @TODO change the deposit to BigNum compatible field
+    deposit_amount VARCHAR(255)             NOT NULL,
     valid_until    TIMESTAMP WITH TIME ZONE NOT NULL,
     spec           JSONB                    NOT NULL,
 
     PRIMARY KEY (id)
 );
 
-CREATE INDEX idx_valid_until ON channels (valid_until);
-CREATE INDEX idx_spec ON channels ((spec -> 'validator' ->> 'id'));
+CREATE INDEX idx_channel_valid_until ON channels (valid_until);
+CREATE INDEX idx_channels_spec_created ON channels ((spec ->> 'created'));
 
 CREATE TABLE validator_messages
 (
     channel_id VARCHAR(66)              NOT NULL REFERENCES channels (id) ON DELETE RESTRICT,
     "from"     VARCHAR(255)             NOT NULL,
     msg        JSONB                    NOT NULL,
-    received   TIMESTAMP WITH TIME ZONE NOT NULL,
-
-    PRIMARY KEY (channel_id)
+    received   TIMESTAMP WITH TIME ZONE NOT NULL
 );
 
-CREATE INDEX idx_received ON validator_messages (received);
-CREATE INDEX ON validator_messages ((msg ->> 'type'));
-CREATE INDEX ON validator_messages ((msg ->> 'stateRoot'));
+CREATE INDEX idx_validator_messages_received ON validator_messages (received);
+CREATE INDEX idx_validator_messages_msg_type ON validator_messages ((msg ->> 'type'));
+CREATE INDEX idx_validator_messages_msg_state_root ON validator_messages ((msg ->> 'stateRoot'));
 
 CREATE TABLE event_aggregates
 (
     channel_id VARCHAR(66)              NOT NULL REFERENCES channels (id) ON DELETE RESTRICT,
     created    TIMESTAMP WITH TIME ZONE NOT NULL,
-    events     JSONB                    NOT NULL,
-
-    PRIMARY KEY (channel_id)
+    events     JSONB                    NOT NULL
 );
 
-CREATE INDEX idx_created ON event_aggregates (created);
+CREATE INDEX idx_event_aggregates_created ON event_aggregates (created);

sentry/migrations/20190806011140_initial_tables/up.sql renamed to sentry/migrations/20190806011140_initial-tables/up.sql

@@ -78,7 +78,7 @@ pub async fn setup_migrations() {
 
     // Define Migrations
     config
-        .use_migrations(&[make_migration!("20190806011140_initial_tables")])
+        .use_migrations(&[make_migration!("20190806011140_initial-tables")])
         .expect("Loading migrations failed");
 
     // Reload config, ping the database for applied migrations

sentry/src/db.rs

@@ -3,6 +3,8 @@ use bb8::RunError;
 use primitives::{Channel, ChannelId, ValidatorId};
 use std::str::FromStr;
 
+pub use list_channels::list_channels;
+
 pub async fn get_channel_by_id(
     pool: &DbPool,
     id: &ChannelId,
@@ -67,3 +69,123 @@ pub async fn insert_channel(
         })
         .await
 }
+
+mod list_channels {
+    use crate::db::DbPool;
+    use bb8::RunError;
+    use bb8_postgres::tokio_postgres::types::{accepts, FromSql, ToSql, Type};
+    use chrono::{DateTime, Utc};
+    use primitives::sentry::ChannelListResponse;
+    use primitives::{Channel, ValidatorId};
+    use std::error::Error;
+    use std::str::FromStr;
+
+    struct TotalCount(pub u64);
+    impl<'a> FromSql<'a> for TotalCount {
+        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(Self(u64::from_str(str_slice)?))
+        }
+
+        // Use a varchar or text, since otherwise `int8` fails deserialization
+        accepts!(VARCHAR, TEXT);
+    }
+
+    pub async fn list_channels(
+        pool: &DbPool,
+        skip: u64,
+        limit: u32,
+        creator: &Option<String>,
+        validator: &Option<ValidatorId>,
+        valid_until_ge: &DateTime<Utc>,
+    ) -> Result<ChannelListResponse, RunError<bb8_postgres::tokio_postgres::Error>> {
+        let validator = validator.as_ref().map(|validator_id| {
+            serde_json::Value::from_str(&format!(r#"[{{"id": "{}"}}]"#, validator_id))
+                .expect("Not a valid json")
+        });
+        let (where_clauses, params) =
+            channel_list_query_params(creator, validator.as_ref(), valid_until_ge);
+        let total_count_params = (where_clauses.clone(), params.clone());
+
+        let channels = pool
+            .run(move |connection| {
+                async move {
+                    // To understand why we use Order by, see Postgres Documentation: https://www.postgresql.org/docs/8.1/queries-limit.html
+                    let statement = format!("SELECT id, creator, deposit_asset, deposit_amount, valid_until, spec FROM channels WHERE {} ORDER BY spec->>'created' DESC LIMIT {} OFFSET {}", where_clauses.join(" AND "), limit, skip);
+                    match connection.prepare(&statement).await {
+                        Ok(stmt) => {
+                            match connection.query(&stmt, params.as_slice()).await {
+                                Ok(rows) => {
+                                    let channels = rows.iter().map(Channel::from).collect();
+
+                                    Ok((channels, connection))
+                                },
+                                Err(e) => Err((e, connection)),
+                            }
+                        },
+                        Err(e) => Err((e, connection)),
+                    }
+                }
+            })
+            .await?;
+
+        let total_count =
+            list_channels_total_count(&pool, (&total_count_params.0, total_count_params.1)).await?;
+
+        // fast ceil for total_pages
+        let total_pages = if total_count == 0 {
+            1
+        } else {
+            1 + ((total_count - 1) / limit as u64)
+        };
+
+        Ok(ChannelListResponse {
+            total_pages,
+            channels,
+        })
+    }
+
+    async fn list_channels_total_count<'a>(
+        pool: &DbPool,
+        (where_clauses, params): (&'a [String], Vec<&'a (dyn ToSql + Sync)>),
+    ) -> Result<u64, RunError<bb8_postgres::tokio_postgres::Error>> {
+        pool.run(move |connection| {
+            async move {
+                let statement = format!(
+                    "SELECT COUNT(id)::varchar FROM channels WHERE {}",
+                    where_clauses.join(" AND ")
+                );
+                match connection.prepare(&statement).await {
+                    Ok(stmt) => match connection.query_one(&stmt, params.as_slice()).await {
+                        Ok(row) => Ok((row.get::<_, TotalCount>(0).0, connection)),
+                        Err(e) => Err((e, connection)),
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+    }
+
+    fn channel_list_query_params<'a>(
+        creator: &'a Option<String>,
+        validator: Option<&'a serde_json::Value>,
+        valid_until_ge: &'a DateTime<Utc>,
+    ) -> (Vec<String>, Vec<&'a (dyn ToSql + Sync)>) {
+        let mut where_clauses = vec!["valid_until >= $1".to_string()];
+        let mut params: Vec<&(dyn ToSql + Sync)> = vec![valid_until_ge];
+
+        if let Some(creator) = creator {
+            where_clauses.push(format!("creator = ${}", params.len() + 1));
+            params.push(creator);
+        }
+
+        if let Some(validator) = validator {
+            where_clauses.push(format!("spec->'validators' @> ${}", params.len() + 1));
+            params.push(validator);
+        }
+
+        (where_clauses, params)
+    }
+}

sentry/src/db/channel.rs

@@ -14,7 +14,7 @@ use primitives::Config;
 use redis::aio::MultiplexedConnection;
 use regex::Regex;
 use routes::cfg::config;
-use routes::channel::{create_channel, last_approved};
+use routes::channel::{channel_list, create_channel, last_approved};
 use slog::{error, Logger};
 use std::collections::HashMap;
 
@@ -109,7 +109,7 @@ impl<A: Adapter + 'static> Application<A> {
         let mut response = match (req.uri().path(), req.method()) {
             ("/cfg", &Method::GET) => config(req, &self).await,
             ("/channel", &Method::POST) => create_channel(req, &self).await,
-            ("/channel/list", &Method::GET) => Err(ResponseError::NotFound),
+            ("/channel/list", &Method::GET) => channel_list(req, &self).await,
             // This is important becuase it prevents us from doing
             // expensive regex matching for routes without /channel
             (route, method) if route.starts_with("/channel") => {

sentry/src/lib.rs

@@ -1,5 +1,5 @@
 use self::channel_list::ChannelListQuery;
-use crate::db::{get_channel_by_id, insert_channel};
+use crate::db::{get_channel_by_id, insert_channel, list_channels};
 use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
@@ -62,16 +62,27 @@ pub async fn create_channel<A: Adapter>(
     Ok(success_response(serde_json::to_string(&create_response)?))
 }
 
-pub async fn channel_list(req: Request<Body>) -> Result<Response<Body>, ResponseError> {
-    // @TODO: Get from Config
-    let _channel_find_limit = 5;
-
+pub async fn channel_list<A: Adapter>(
+    req: Request<Body>,
+    app: &Application<A>,
+) -> Result<Response<Body>, ResponseError> {
     let query = serde_urlencoded::from_str::<ChannelListQuery>(&req.uri().query().unwrap_or(""))?;
-
-    // @TODO: List all channels returned from the DB
-    println!("{:?}", query);
-
-    Err(ResponseError::NotFound)
+    let skip = query
+        .page
+        .checked_mul(app.config.channels_find_limit.into())
+        .ok_or_else(|| ResponseError::BadRequest("Page and/or limit is too large".into()))?;
+
+    let list_response = list_channels(
+        &app.pool,
+        skip,
+        app.config.channels_find_limit,
+        &query.creator,
+        &query.validator,
+        &query.valid_until_ge,
+    )
+    .await?;
+
+    Ok(success_response(serde_json::to_string(&list_response)?))
 }
 
 pub async fn last_approved<A: Adapter>(
@@ -93,40 +104,29 @@ pub async fn last_approved<A: Adapter>(
 }
 
 mod channel_list {
+    use chrono::serde::ts_seconds::deserialize as ts_seconds;
     use chrono::{DateTime, Utc};
-    use serde::{Deserialize, Deserializer};
+    use primitives::ValidatorId;
+    use serde::Deserialize;
 
     #[derive(Debug, Deserialize)]
-    pub(crate) struct ChannelListQuery {
-        /// page to show, should be >= 1
+    pub(super) struct ChannelListQuery {
         #[serde(default = "default_page")]
         pub page: u64,
-        /// channels limit per page, should be >= 1
-        #[serde(default = "default_limit")]
-        pub limit: u32,
         /// filters the list on `valid_until >= valid_until_ge`
-        #[serde(default = "Utc::now")]
+        /// It should be the same timestamp format as the `Channel.valid_until`: **seconds**
+        #[serde(
+            deserialize_with = "ts_seconds",
+            default = "Utc::now",
+            rename = "validUntil"
+        )]
         pub valid_until_ge: DateTime<Utc>,
+        pub creator: Option<String>,
         /// filters the channels containing a specific validator if provided
-        #[serde(default, deserialize_with = "deserialize_validator")]
-        pub validator: Option<String>,
-    }
-
-    /// Deserialize the `Option<String>`, but if the `String` is empty it will return `None`
-    fn deserialize_validator<'de, D>(de: D) -> Result<Option<String>, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        let value: String = Deserialize::deserialize(de)?;
-        let option = Some(value).filter(|string| !string.is_empty());
-        Ok(option)
-    }
-
-    fn default_limit() -> u32 {
-        1
+        pub validator: Option<ValidatorId>,
     }
 
     fn default_page() -> u64 {
-        1
+        0
     }
 }

sentry/src/routes/channel.rs

@@ -6,7 +6,7 @@ use futures_legacy::Future as LegacyFuture;
 use primitives::adapter::Adapter;
 use primitives::channel::SpecValidator;
 use primitives::sentry::{
-    ChannelAllResponse, EventAggregateResponse, LastApprovedResponse, SuccessResponse,
+    ChannelListResponse, EventAggregateResponse, LastApprovedResponse, SuccessResponse,
     ValidatorMessageResponse,
 };
 use primitives::validator::MessageTypes;
@@ -230,7 +230,7 @@ pub async fn all_channels(
     if first_page.total_pages < 2 {
         Ok(first_page.channels)
     } else {
-        let mut all: Vec<ChannelAllResponse> = try_join_all(
+        let mut all: Vec<ChannelListResponse> = try_join_all(
             (1..first_page.total_pages).map(|i| fetch_page(url.clone(), i, whoami.clone())),
         )
         .await?;
@@ -248,7 +248,7 @@ async fn fetch_page(
     sentry_url: String,
     page: u64,
     validator: String,
-) -> Result<ChannelAllResponse, reqwest::Error> {
+) -> Result<ChannelListResponse, reqwest::Error> {
     let client = Client::new();
 
     let mut query = vec![format!("page={}", page)];
@@ -257,7 +257,7 @@ async fn fetch_page(
     let future = client
         .get(format!("{}/channel/list?{}", sentry_url, query.join("&")).as_str())
         .send()
-        .and_then(|mut res: Response| res.json::<ChannelAllResponse>());
+        .and_then(|mut res: Response| res.json::<ChannelListResponse>());
 
     future.compat().await
 }