sentry - db - list_channels & list_channels_total_pages

Author: elpiel

sentry/src/db/channel.rs

@@ -1,8 +1,23 @@
 use crate::db::DbPool;
 use bb8::RunError;
+use bb8_postgres::tokio_postgres::types::{accepts, FromSql, ToSql, Type};
+use chrono::{DateTime, Utc};
 use primitives::{Channel, ChannelId, ValidatorId};
+use std::error::Error;
 use std::str::FromStr;
 
+struct TotalPages(pub u64);
+impl<'a> FromSql<'a> for TotalPages {
+    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 get_channel_by_id(
     pool: &DbPool,
     id: &ChannelId,
@@ -67,3 +82,93 @@ pub async fn insert_channel(
         })
         .await
 }
+
+pub async fn list_channels(
+    pool: &DbPool,
+    skip: u64,
+    limit: u32,
+    creator: &Option<String>,
+    validator: &Option<ValidatorId>,
+    valid_until_ge: &DateTime<Utc>,
+) -> Result<Vec<Channel>, 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);
+
+    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 id 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
+}
+
+pub async fn list_channels_total_pages(
+    pool: &DbPool,
+    creator: &Option<String>,
+    validator: &Option<ValidatorId>,
+    valid_until_ge: &DateTime<Utc>,
+) -> Result<u64, 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);
+
+    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::<_, TotalPages>(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)
+}