Merge pull request #441 from AdExNetwork/issue-392-get-all-spenders-info

Added Pagination, added created field to Spendable, added limit to config

Author: simzzz

docs/config/dev.toml

@@ -3,6 +3,7 @@ max_channels = 512
 
 channels_find_limit = 200
 campaigns_find_limit = 200
+spendable_find_limit = 200
 wait_time = 500
 
 # V4 Deprecated

docs/config/prod.toml

@@ -3,6 +3,7 @@ max_channels = 512
 
 channels_find_limit = 512
 campaigns_find_limit = 512
+spendable_find_limit = 512
 
 wait_time = 40000
 

primitives/src/config.rs

@@ -26,6 +26,7 @@ pub struct Config {
     pub max_channels: u32,
     pub channels_find_limit: u32,
     pub campaigns_find_limit: u32,
+    pub spendable_find_limit: u32,
     pub wait_time: u32,
     #[deprecated = "redundant V4 value. No aggregates are needed for V5"]
     pub aggr_throttle: u32,

primitives/src/sentry.rs

@@ -185,7 +185,6 @@ pub struct AggregateEvents {
 #[serde(rename_all = "camelCase")]
 pub struct Pagination {
     pub total_pages: u64,
-    pub total: u64,
     pub page: u64,
 }
 
@@ -225,6 +224,12 @@ pub struct AllSpendersResponse {
     pub pagination: Pagination,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AllSpendersQuery {
+    // default is `u64::default()` = `0`
+    pub page: u64,
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ValidatorMessage {
     pub from: ValidatorId,

sentry/migrations/20190806011140_initial-tables/up.sql

@@ -43,6 +43,7 @@ CREATE TABLE spendable (
     channel_id varchar(66) NOT NULL,
     total bigint NOT NULL,
     still_on_create2 bigint NOT NULL,
+    created timestamp(2) with time zone NOT NULL,
     PRIMARY KEY (spender, channel_id),
     CONSTRAINT fk_spendable_channel_id FOREIGN KEY (channel_id) REFERENCES channels (id) ON DELETE RESTRICT ON UPDATE RESTRICT
 );

sentry/src/db/campaign.rs

@@ -95,7 +95,6 @@ pub async fn list_campaigns(
 
     let pagination = Pagination {
         total_pages,
-        total: total_pages,
         page: skip / limit as u64,
     };
 

sentry/src/db/channel.rs

@@ -147,7 +147,6 @@ mod list_channels {
             channels,
             pagination: Pagination {
                 total_pages,
-                total: total_pages,
                 page: skip / limit as u64,
             },
         })

sentry/src/db/spendable.rs

@@ -1,4 +1,6 @@
-use primitives::{spender::Spendable, Address, ChannelId};
+use crate::db::TotalCount;
+use chrono::Utc;
+use primitives::{sentry::Pagination, spender::Spendable, Address, ChannelId};
 
 use super::{DbPool, PoolError};
 
@@ -8,7 +10,7 @@ use super::{DbPool, PoolError};
 /// ```
 pub async fn insert_spendable(pool: DbPool, spendable: &Spendable) -> Result<bool, PoolError> {
     let client = pool.get().await?;
-    let stmt = client.prepare("INSERT INTO spendable (spender, channel_id, total, still_on_create2) values ($1, $2, $3, $4)").await?;
+    let stmt = client.prepare("INSERT INTO spendable (spender, channel_id, total, still_on_create2, created) values ($1, $2, $3, $4, $5)").await?;
 
     let row = client
         .execute(
@@ -18,6 +20,7 @@ pub async fn insert_spendable(pool: DbPool, spendable: &Spendable) -> Result<boo
                 &spendable.channel.id(),
                 &spendable.deposit.total,
                 &spendable.deposit.still_on_create2,
+                &Utc::now(),
             ],
         )
         .await?;
@@ -35,29 +38,50 @@ pub async fn fetch_spendable(
     channel_id: &ChannelId,
 ) -> Result<Option<Spendable>, PoolError> {
     let client = pool.get().await?;
-    let statement = client.prepare("SELECT spender, total, still_on_create2, channels.leader, channels.follower, channels.guardian, channels.token, channels.nonce FROM spendable INNER JOIN channels ON channels.id = spendable.channel_id WHERE spender = $1 AND channel_id = $2").await?;
+    let statement = client.prepare("SELECT spender, total, still_on_create2, spendable.created, channels.leader, channels.follower, channels.guardian, channels.token, channels.nonce FROM spendable INNER JOIN channels ON channels.id = spendable.channel_id WHERE spender = $1 AND channel_id = $2").await?;
 
     let row = client.query_opt(&statement, &[spender, channel_id]).await?;
 
     Ok(row.as_ref().map(Spendable::from))
 }
 
-static GET_ALL_SPENDERS_STATEMENT: &str = "SELECT spender, total, still_on_create2, channels.leader, channels.follower, channels.guardian, channels.token, channels.nonce FROM spendable INNER JOIN channels ON channels.id = spendable.channel_id WHERE channel_id = $1";
+static GET_ALL_SPENDERS_STATEMENT: &str = "SELECT spender, total, still_on_create2, spendable.created, channels.leader, channels.follower, channels.guardian, channels.token, channels.nonce FROM spendable INNER JOIN channels ON channels.id = spendable.channel_id WHERE channel_id = $1 ORDER BY spendable.created ASC LIMIT $2 OFFSET $3";
 
-// TODO: Include pagination
 pub async fn get_all_spendables_for_channel(
     pool: DbPool,
     channel_id: &ChannelId,
-) -> Result<Vec<Spendable>, PoolError> {
+    skip: u64,
+    limit: u64,
+) -> Result<(Vec<Spendable>, Pagination), PoolError> {
     let client = pool.get().await?;
     let statement = client.prepare(GET_ALL_SPENDERS_STATEMENT).await?;
 
-    let rows = client.query(&statement, &[channel_id]).await?;
+    let rows = client
+        .query(
+            &statement,
+            &[channel_id, &limit.to_string(), &skip.to_string()],
+        )
+        .await?;
+    let spendables = rows.iter().map(Spendable::from).collect();
+
+    let total_count = list_spendable_total_count(&pool, channel_id).await?;
+
+    // fast ceil for total_pages
+    let total_pages = if total_count == 0 {
+        1
+    } else {
+        1 + ((total_count - 1) / limit as u64)
+    };
+
+    let pagination = Pagination {
+        total_pages,
+        page: skip / limit as u64,
+    };
 
-    Ok(rows.iter().map(Spendable::from).collect())
+    Ok((spendables, pagination))
 }
 
-static UPDATE_SPENDABLE_STATEMENT: &str = "WITH inserted_spendable AS (INSERT INTO spendable(spender, channel_id, total, still_on_create2) VALUES($1, $2, $3, $4) ON CONFLICT ON CONSTRAINT spendable_pkey DO UPDATE SET total = $3, still_on_create2 = $4 WHERE spendable.spender = $1 AND spendable.channel_id = $2 RETURNING *) SELECT inserted_spendable.*, channels.leader, channels.follower, channels.guardian, channels.token, channels.nonce FROM inserted_spendable INNER JOIN channels ON inserted_spendable.channel_id = channels.id";
+static UPDATE_SPENDABLE_STATEMENT: &str = "WITH inserted_spendable AS (INSERT INTO spendable(spender, channel_id, total, still_on_create2, created) VALUES($1, $2, $3, $4, $5) ON CONFLICT ON CONSTRAINT spendable_pkey DO UPDATE SET total = $3, still_on_create2 = $4 WHERE spendable.spender = $1 AND spendable.channel_id = $2 RETURNING *) SELECT inserted_spendable.*, channels.leader, channels.follower, channels.guardian, channels.token, channels.nonce FROM inserted_spendable INNER JOIN channels ON inserted_spendable.channel_id = channels.id";
 
 // Updates spendable entry deposit or inserts a new spendable entry if it doesn't exist
 pub async fn update_spendable(pool: DbPool, spendable: &Spendable) -> Result<Spendable, PoolError> {
@@ -72,13 +96,28 @@ pub async fn update_spendable(pool: DbPool, spendable: &Spendable) -> Result<Spe
                 &spendable.channel.id(),
                 &spendable.deposit.total,
                 &spendable.deposit.still_on_create2,
+                &Utc::now(),
             ],
         )
         .await?;
 
     Ok(Spendable::from(&row))
 }
 
+async fn list_spendable_total_count<'a>(
+    pool: &DbPool,
+    channel_id: &ChannelId,
+) -> Result<u64, PoolError> {
+    let client = pool.get().await?;
+
+    let statement =
+        "SELECT COUNT(spendable.id)::varchar FROM spendable INNER JOIN channels ON spendable.channel_id=channels.id WHERE spendable.channel_id = $1";
+    let stmt = client.prepare(statement).await?;
+    let row = client.query_one(&stmt, &[&channel_id]).await?;
+
+    Ok(row.get::<_, TotalCount>(0).0)
+}
+
 #[cfg(test)]
 mod test {
     use primitives::{

sentry/src/routes/campaign.rs

@@ -225,10 +225,14 @@ pub async fn campaign_list<A: Adapter>(
     let mut query =
         serde_urlencoded::from_str::<CampaignListQuery>(req.uri().query().unwrap_or(""))?;
 
-    query.validator = match (query.validator, query.is_leader, req.extensions().get::<Auth>()) {
-        (None, Some(true), Some(session)) => Some(session.uid), // only case where session.uid is used
+    query.validator = match (
+        query.validator,
+        query.is_leader,
+        req.extensions().get::<Auth>(),
+    ) {
+        (None, Some(true), Some(auth)) => Some(auth.uid), // only case where Auth.uid is used
         (Some(validator), _, _) => Some(validator), // for all cases with a validator passed
-        _ => None, // default, no filtration by validator
+        _ => None,                                  // default, no filtration by validator
     };
 
     let limit = app.config.campaigns_find_limit;

sentry/src/routes/channel.rs

@@ -13,8 +13,8 @@ use primitives::{
     balances::{Balances, CheckedState, UncheckedState},
     config::TokenInfo,
     sentry::{
-        channel_list::ChannelListQuery, AccountingResponse, AllSpendersResponse, LastApproved,
-        LastApprovedQuery, LastApprovedResponse, Pagination, SpenderResponse, SuccessResponse,
+        channel_list::ChannelListQuery, AccountingResponse, AllSpendersQuery, AllSpendersResponse,
+        LastApproved, LastApprovedQuery, LastApprovedResponse, SpenderResponse, SuccessResponse,
     },
     spender::{Spendable, Spender, SpenderLeaf},
     validator::{MessageTypes, NewState},
@@ -267,11 +267,19 @@ pub async fn get_all_spender_limits<A: Adapter + 'static>(
         .expect("Request should have Channel")
         .to_owned();
 
+    let query = serde_urlencoded::from_str::<AllSpendersQuery>(req.uri().query().unwrap_or(""))?;
+    let limit = app.config.spendable_find_limit;
+    let skip = query
+        .page
+        .checked_mul(limit.into())
+        .ok_or_else(|| ResponseError::FailedValidation("Page and/or limit is too large".into()))?;
+
     let new_state = get_corresponding_new_state(&app.pool, &app.logger, &channel).await?;
 
     let mut all_spender_limits: HashMap<Address, Spender> = HashMap::new();
 
-    let all_spendables = get_all_spendables_for_channel(app.pool.clone(), &channel.id()).await?;
+    let (all_spendables, pagination) =
+        get_all_spendables_for_channel(app.pool.clone(), &channel.id(), skip, limit.into()).await?;
 
     // Using for loop to avoid async closures
     for spendable in all_spendables {
@@ -300,12 +308,7 @@ pub async fn get_all_spender_limits<A: Adapter + 'static>(
 
     let res = AllSpendersResponse {
         spenders: all_spender_limits,
-        pagination: Pagination {
-            // TODO
-            page: 1,
-            total: 1,
-            total_pages: 1,
-        },
+        pagination,
     };
 
     Ok(success_response(serde_json::to_string(&res)?))