Merge branch 'dev' into issue-9-validator-messages

Author: elpiel

Cargo.lock

@@ -1,8 +1,7 @@
-use chrono::serde::ts_milliseconds;
+use chrono::serde::{ts_milliseconds, ts_milliseconds_option};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 
-use crate::util::serde::ts_milliseconds_option;
 use crate::TargetingTag;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]

primitives/src/ad_unit.rs

@@ -1,13 +1,12 @@
 use std::error::Error;
 use std::fmt;
 
-use chrono::serde::{ts_milliseconds, ts_seconds};
+use chrono::serde::{ts_milliseconds, ts_milliseconds_option, ts_seconds};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_hex::{SerHex, StrictPfx};
 
 use crate::big_num::BigNum;
-use crate::util::serde::ts_milliseconds_option;
 use crate::{AdUnit, EventSubmission, TargetingTag, ValidatorDesc, ValidatorId};
 use hex::{FromHex, FromHexError};
 use std::ops::Deref;

primitives/src/channel.rs

@@ -22,7 +22,6 @@ pub mod util {
     }
 
     pub mod logging;
-    pub mod serde;
 }
 pub mod validator;
 

primitives/src/lib.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,
 }

primitives/src/sentry.rs

@@ -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);

primitives/src/util/serde.rs

@@ -81,7 +81,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/migrations/20190806011140_initial_tables/down.sql renamed to sentry/migrations/20190806011140_initial-tables/down.sql

@@ -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)
+    }
+}