Issue #61 sentry paginated channel list (#63)

* domain - channel - ChannelRepository - list_count:
- persistence - channel - memory - impl `list_count`

* application - channel resource - channel_list - show `total_pages`

* domain - channel - ChannelRepository list_count - change count to `u64` and:
- app - resource - channel_list - response of `total_pages` to be `u64`
- persistence - postgres - impl `list_count`

* sentry - util - bb8 - util for a `query_result`

* validator - persistence - channel AllResponse:
 - add `total_pages`
 - add `serde(rename_all = camelCase)` attribute

Author: elpiel

sentry/src/application/resource/channel/channel_list/handler.rs

@@ -34,7 +34,12 @@ impl ChannelListHandler {
         let list_fut = self.channel_repository.list(&channel_list_params);
         // @TODO: Proper error handling
         let channels = await!(list_fut).unwrap();
+        let channels_count =
+            await!(self.channel_repository.list_count(&channel_list_params)).unwrap();
 
-        Ok(ChannelListResponse { channels })
+        Ok(ChannelListResponse {
+            channels,
+            total_pages: channels_count,
+        })
     }
 }

sentry/src/application/resource/channel/channel_list/response.rs

@@ -6,4 +6,5 @@ use domain::Channel;
 #[derive(Debug, Response)]
 pub struct ChannelListResponse {
     pub channels: Vec<Channel>,
+    pub total_pages: u64,
 }

sentry/src/domain/channel.rs

@@ -53,6 +53,8 @@ pub trait ChannelRepository: Debug + Send + Sync {
     /// Returns a list of channels, based on the passed Parameters for this method
     fn list(&self, params: &ChannelListParams) -> RepositoryFuture<Vec<Channel>>;
 
+    fn list_count(&self, params: &ChannelListParams) -> RepositoryFuture<u64>;
+
     fn find(&self, channel_id: &ChannelId) -> RepositoryFuture<Option<Channel>>;
 
     fn create(&self, channel: Channel) -> RepositoryFuture<()>;

sentry/src/infrastructure.rs

@@ -1,2 +1,3 @@
 pub(crate) mod field;
 pub mod persistence;
+pub(crate) mod util;

sentry/src/infrastructure/persistence/channel/memory.rs

@@ -36,28 +36,10 @@ impl ChannelRepository for MemoryChannelRepository {
             Ok(reader) => {
                 let channels = reader
                     .iter()
-                    .filter_map(|channel| {
-                        let valid_until_filter = channel.valid_until >= params.valid_until_ge;
-
-                        let validator_filter_passed = match &params.validator {
-                            Some(validator_id) => {
-                                // check if there is any validator in the current
-                                // `channel.spec.validators` that has the same `id`
-                                channel.spec.validators.find(&validator_id).is_some()
-                            }
-                            // if None -> the current channel has passed, since we don't need to filter by anything
-                            None => true,
-                        };
-
-                        match (valid_until_filter, validator_filter_passed) {
-                            (true, true) => Some(channel.clone()),
-                            (_, _) => None,
-                        }
-                    })
+                    .filter_map(|channel| list_filter(&params, channel))
                     .skip(skip_results)
                     .take(take)
                     .collect();
-
                 ok(channels)
             }
             Err(error) => err(MemoryPersistenceError::from(error).into()),
@@ -66,6 +48,22 @@ impl ChannelRepository for MemoryChannelRepository {
         res_fut.boxed()
     }
 
+    fn list_count(&self, params: &ChannelListParams) -> RepositoryFuture<u64> {
+        let res_fut = match self.records.read() {
+            Ok(reader) => {
+                let filtered_count = reader
+                    .iter()
+                    .filter_map(|channel| list_filter(&params, channel))
+                    .count();
+                let pages = (filtered_count as f64 / f64::from(params.limit)).ceil() as u64;
+                ok(pages)
+            }
+            Err(error) => err(MemoryPersistenceError::from(error).into()),
+        };
+
+        res_fut.boxed()
+    }
+
     fn find(&self, channel_id: &ChannelId) -> RepositoryFuture<Option<Channel>> {
         let res_fut = match self.records.read() {
             Ok(reader) => {
@@ -113,3 +111,22 @@ impl ChannelRepository for MemoryChannelRepository {
         create_fut.boxed()
     }
 }
+
+fn list_filter(params: &ChannelListParams, channel: &Channel) -> Option<Channel> {
+    let valid_until_filter = channel.valid_until >= params.valid_until_ge;
+
+    let validator_filter_passed = match &params.validator {
+        Some(validator_id) => {
+            // check if there is any validator in the current
+            // `channel.spec.validators` that has the same `id`
+            channel.spec.validators.find(&validator_id).is_some()
+        }
+        // if None -> the current channel has passed, since we don't need to filter by anything
+        None => true,
+    };
+
+    match (valid_until_filter, validator_filter_passed) {
+        (true, true) => Some(channel.clone()),
+        (_, _) => None,
+    }
+}

sentry/src/infrastructure/persistence/channel/postgres.rs

@@ -1,9 +1,8 @@
 use futures::compat::Future01CompatExt;
 use futures::future::FutureExt;
-use futures_legacy::future::IntoFuture;
 use futures_legacy::Future as OldFuture;
-use futures_legacy::Stream as OldStream;
 use tokio_postgres::types::Json;
+use tokio_postgres::Row;
 
 use domain::{Channel, ChannelId, ChannelSpec, RepositoryFuture};
 use try_future::try_future;
@@ -12,6 +11,7 @@ use crate::domain::channel::{ChannelListParams, ChannelRepository};
 use crate::infrastructure::field::{asset::AssetPg, bignum::BigNumPg, channel_id::ChannelIdPg};
 use crate::infrastructure::persistence::postgres::PostgresPersistenceError;
 use crate::infrastructure::persistence::DbPool;
+use crate::infrastructure::util::bb8::query_result;
 
 #[derive(Debug)]
 pub struct PostgresChannelRepository {
@@ -31,31 +31,11 @@ impl ChannelRepository for PostgresChannelRepository {
             .run(move |mut conn| {
                 conn.prepare("SELECT channel_id, creator, deposit_asset, deposit_amount, valid_until, spec FROM channels")
                     .then(move |res| match res {
-                        Ok(stmt) => {
-                            conn
-                                .query(&stmt, &[])
-                                .collect()
-                                .into_future()
-                                .then(|res| match res {
-                                    Ok(rows) => Ok((rows, conn)),
-                                    Err(err) => Err((err, conn)),
-                                })
-                                .into()
-                        }
+                        Ok(stmt) => query_result(conn.query(&stmt, &[]), conn),
                         Err(err) => try_future!(Err((err, conn))),
                     })
                     .and_then(|(rows, conn)| {
-                        let channels = rows.iter().map(|row| {
-                            let spec: ChannelSpec = row.get::<_, Json<ChannelSpec>>("spec").0;
-                            Channel {
-                                id: row.get::<_, ChannelIdPg>("channel_id").into(),
-                                creator: row.get("creator"),
-                                deposit_asset: row.get::<_, AssetPg>("deposit_asset").into(),
-                                deposit_amount: row.get::<_, BigNumPg>("deposit_amount").into(),
-                                valid_until: row.get("valid_until"),
-                                spec,
-                            }
-                        }).collect();
+                        let channels = rows.iter().map(|row| channel_map(row)).collect();
 
                         Ok((channels, conn))
                     })
@@ -65,6 +45,29 @@ impl ChannelRepository for PostgresChannelRepository {
         fut.compat().boxed()
     }
 
+    fn list_count(&self, _params: &ChannelListParams) -> RepositoryFuture<u64> {
+        let fut = self
+            .db_pool
+            .run(move |mut conn| {
+                conn.prepare("SELECT COUNT(channel_id)::TEXT FROM channels")
+                    .then(move |res| match res {
+                        Ok(stmt) => query_result(conn.query(&stmt, &[]), conn),
+                        Err(err) => try_future!(Err((err, conn))),
+                    })
+                    .and_then(|(rows, conn)| {
+                        let count = rows[0]
+                            .get::<_, &str>(0)
+                            .parse::<u64>()
+                            .expect("Not possible to have that many rows");
+
+                        Ok((count, conn))
+                    })
+            })
+            .map_err(|err| PostgresPersistenceError::from(err).into());
+
+        fut.compat().boxed()
+    }
+
     fn find(&self, _channel_id: &ChannelId) -> RepositoryFuture<Option<Channel>> {
         unimplemented!("find() for Postgres still needs to be implemented")
     }
@@ -73,3 +76,15 @@ impl ChannelRepository for PostgresChannelRepository {
         unimplemented!("create() for Postgres still needs to be implemented")
     }
 }
+
+fn channel_map(row: &Row) -> Channel {
+    let spec: ChannelSpec = row.get::<_, Json<ChannelSpec>>("spec").0;
+    Channel {
+        id: row.get::<_, ChannelIdPg>("channel_id").into(),
+        creator: row.get("creator"),
+        deposit_asset: row.get::<_, AssetPg>("deposit_asset").into(),
+        deposit_amount: row.get::<_, BigNumPg>("deposit_amount").into(),
+        valid_until: row.get("valid_until"),
+        spec,
+    }
+}

sentry/src/infrastructure/util.rs

@@ -0,0 +1 @@
+pub(crate) mod bb8;

sentry/src/infrastructure/util/bb8.rs

@@ -0,0 +1,21 @@
+use futures_legacy::future::IntoFuture;
+use futures_legacy::stream::Stream;
+use futures_legacy::Future;
+use tokio_postgres::impls::Query;
+use tokio_postgres::{Client, Row};
+use try_future::TryFuture;
+
+pub(crate) fn query_result(
+    query: Query,
+    client: Client,
+) -> TryFuture<impl Future<Item = (Vec<Row>, Client), Error = (tokio_postgres::Error, Client)>> {
+    query
+        .collect()
+        .into_future()
+        .then(|res| match res {
+            Ok(rows) => Ok((rows, client)),
+            Err(err) => Err((err, client)),
+        })
+        .into_future()
+        .into()
+}

validator/src/infrastructure/persistence/channel/api.rs

@@ -40,6 +40,8 @@ impl ChannelRepository for ApiChannelRepository {
 }
 
 #[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
 struct AllResponse {
     pub channels: Vec<Channel>,
+    pub total_pages: u64,
 }