Issue #55 Initial validator crate skeleton (#57)

* validator crate

* validator - Cargo.toml - add `futures-preview`

* validator - domain, infrastructure skeleton modules

* validator - lib - add `deny` attr for `rust_2018_idioms` & `clippy::all`

* validator:
- lib - add `async_await` & `await_macro`
- Cargo.toml - add `reqwest` for the api calls towards Sentry

* validator - domain - ChannelRepository

* validator - domain - register `channel` module

* validator - domain - fix `ChannelRepository::all`

* validator - domain - ChannelRepository - fix clippy

* sentry - domain - move ChannelRepository to Sentry

* validator - add `repositories` feature + `futures 0.1` for `reqwest`

* validator - Cargo.toml - `serde` & `serde_json`

* validator - infrastructure - add simple WIP api implementation

* sentry - domain - ChannelRepository - require `Debug` to be used with `Arc`

* sentry - use `Arc` to pass the concrete impl of repository for a `Resource`

* validator - persistence - channel - api - `page` to `u32`

* domain - util - `ts_milliseconds_option` fix deserialization

* validator - Cargo.toml - add `tokio`

* validator - main.rs - Fetch Channels for validator by ApiChannelRepository

Author: elpiel

Cargo.lock

@@ -4,4 +4,5 @@ members = [
     "adapter",
     "domain",
     "sentry",
+    "validator",
 ]

Cargo.toml

@@ -10,9 +10,6 @@ use crate::bignum::BigNum;
 use crate::util::serde::ts_milliseconds_option;
 use crate::{AdUnit, Asset, DomainError, EventSubmission, TargetingTag, ValidatorDesc};
 
-#[cfg(feature = "repositories")]
-pub use self::repository::ChannelRepository;
-
 #[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Copy, Clone)]
 #[serde(transparent)]
 pub struct ChannelId {
@@ -209,66 +206,6 @@ impl From<[ValidatorDesc; 2]> for SpecValidators {
     }
 }
 
-pub struct ChannelListParams {
-    /// page to show, should be >= 1
-    pub page: u32,
-    /// channels limit per page, should be >= 1
-    pub limit: u32,
-    /// filters `valid_until` to be `>= valid_until_ge`
-    pub valid_until_ge: DateTime<Utc>,
-    /// filters the channels containing a specific validator if provided
-    pub validator: Option<String>,
-    /// Ensures that this struct can only be created by calling `new()`
-    _secret: (),
-}
-
-impl ChannelListParams {
-    pub fn new(
-        valid_until_ge: DateTime<Utc>,
-        limit: u32,
-        page: u32,
-        validator: Option<String>,
-    ) -> Result<Self, DomainError> {
-        if page < 1 {
-            return Err(DomainError::InvalidArgument(
-                "Page should be >= 1".to_string(),
-            ));
-        }
-
-        if limit < 1 {
-            return Err(DomainError::InvalidArgument(
-                "Limit should be >= 1".to_string(),
-            ));
-        }
-
-        let validator = validator.and_then(|s| if s.is_empty() { None } else { Some(s) });
-
-        Ok(Self {
-            valid_until_ge,
-            page,
-            limit,
-            validator,
-            _secret: (),
-        })
-    }
-}
-
-#[cfg(feature = "repositories")]
-pub mod repository {
-    use crate::RepositoryFuture;
-
-    use super::*;
-
-    pub trait ChannelRepository: Send + Sync {
-        /// Returns a list of channels, based on the passed Parameters for this method
-        fn list(&self, params: &ChannelListParams) -> RepositoryFuture<Vec<Channel>>;
-
-        fn save(&self, channel: Channel) -> RepositoryFuture<()>;
-
-        fn find(&self, channel_id: &ChannelId) -> RepositoryFuture<Option<Channel>>;
-    }
-}
-
 #[cfg(any(test, feature = "fixtures"))]
 #[path = "./channel_fixtures.rs"]
 pub mod fixtures;

domain/src/channel.rs

@@ -9,9 +9,7 @@ pub use util::tests as test_util;
 pub use self::ad_unit::AdUnit;
 pub use self::asset::Asset;
 pub use self::bignum::BigNum;
-#[cfg(feature = "repositories")]
-pub use self::channel::ChannelRepository;
-pub use self::channel::{Channel, ChannelId, ChannelListParams, ChannelSpec};
+pub use self::channel::{Channel, ChannelId, ChannelSpec, SpecValidators};
 pub use self::event_submission::EventSubmission;
 #[cfg(feature = "repositories")]
 pub use self::repository::*;

domain/src/lib.rs

@@ -6,7 +6,8 @@ pub mod serde {
         use chrono::serde::ts_milliseconds::deserialize as from_ts_milliseconds;
         use chrono::serde::ts_milliseconds::serialize as to_ts_milliseconds;
         use chrono::{DateTime, Utc};
-        use serde::{Deserializer, Serializer};
+        use serde::{de, Serializer};
+        use std::fmt;
 
         pub fn serialize<S>(opt: &Option<DateTime<Utc>>, serializer: S) -> Result<S::Ok, S::Error>
         where
@@ -20,9 +21,37 @@ pub mod serde {
 
         pub fn deserialize<'de, D>(de: D) -> Result<Option<DateTime<Utc>>, D::Error>
         where
-            D: Deserializer<'de>,
+            D: de::Deserializer<'de>,
         {
-            from_ts_milliseconds(de).map(Some)
+            Ok(de
+                .deserialize_option(OptionMilliSecondsTimestampVisitor)
+                .map(|opt| opt.map(|dt| dt.with_timezone(&Utc))))?
+        }
+
+        struct OptionMilliSecondsTimestampVisitor;
+
+        impl<'de> de::Visitor<'de> for OptionMilliSecondsTimestampVisitor {
+            type Value = Option<DateTime<Utc>>;
+
+            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+                formatter.write_str("a unix timestamp in milliseconds or none")
+            }
+
+            /// Deserialize a timestamp in seconds since the epoch
+            fn visit_none<E>(self) -> Result<Option<DateTime<Utc>>, E>
+            where
+                E: de::Error,
+            {
+                Ok(None)
+            }
+
+            /// Deserialize a timestamp in seconds since the epoch
+            fn visit_some<D>(self, de: D) -> Result<Option<DateTime<Utc>>, D::Error>
+            where
+                D: de::Deserializer<'de>,
+            {
+                from_ts_milliseconds(de).map(Some)
+            }
         }
     }
 }

domain/src/util.rs

@@ -9,6 +9,10 @@ name = "sentry"
 path = "src/lib.rs"
 
 [dependencies]
+# Domain
+chrono = { version = "0.4", features = ["serde"] }
+time = "0.1.42"
+domain = { version = "0.1", path = "../domain", features = ["repositories"] }
 # Futures
 futures-preview = { version = "=0.3.0-alpha.16", features = ["compat", "io-compat"] }
 futures_legacy = { version = "0.1", package = "futures" }
@@ -31,10 +35,6 @@ serde = { version = "^1.0", features = ['derive'] }
 serde_json = "1.0"
 # Utils
 try_future = "0.1.3"
-# Domain
-chrono = { version = "0.4", features = ["serde"] }
-time = "0.1.42"
-domain = { version = "0.1", path = "../domain", features = ["repositories"] }
 [dev-dependencies]
 domain = { version = "0.1", path = "../domain", features = ["fixtures", "repositories"] }
 fake = { version = "^1.3", features = ["chrono"] }

sentry/Cargo.toml

@@ -6,18 +6,16 @@ use tower_web::{derive_resource_impl, impl_web, Deserialize, Extract};
 use channel_create::{ChannelCreateHandler, ChannelCreateResponse, ChannelInput};
 use channel_list::{ChannelListHandler, ChannelListResponse};
 
-use crate::infrastructure::persistence::channel::{
-    MemoryChannelRepository, PostgresChannelRepository,
-};
-use crate::infrastructure::persistence::DbPool;
+use crate::domain::channel::ChannelRepository;
+use std::sync::Arc;
 
 mod channel_create;
 mod channel_list;
 
 #[derive(Clone, Debug)]
 pub struct ChannelResource {
-    pub db_pool: DbPool,
     pub channel_list_limit: u32,
+    pub channel_repository: Arc<dyn ChannelRepository>,
 }
 
 impl_web! {
@@ -26,21 +24,16 @@ impl_web! {
         #[post("/channel")]
         #[content_type("application/json")]
         async fn create_channel(&self, body: ChannelInput) -> ChannelCreateResponse {
-            let _channel_repository = PostgresChannelRepository::new(self.db_pool.clone());
-            let channel_repository = MemoryChannelRepository::new(None);
 
-            let handler = ChannelCreateHandler::new(&channel_repository);
+            let handler = ChannelCreateHandler::new(self.channel_repository.clone());
 
             await!(handler.handle(body).boxed().compat()).unwrap()
         }
 
         #[get("/channel/list")]
         #[content_type("application/json")]
         async fn channel_list(&self, query_string: ChannelListQuery) -> ChannelListResponse {
-            let _channel_repository = PostgresChannelRepository::new(self.db_pool.clone());
-            let channel_repository = MemoryChannelRepository::new(None);
-
-            let handler = ChannelListHandler::new(self.channel_list_limit, &channel_repository);
+            let handler = ChannelListHandler::new(self.channel_list_limit, self.channel_repository.clone());
 
             await!(handler.handle(query_string.page(), query_string.validator()).boxed().compat()).unwrap()
         }

sentry/src/application/resource/channel.rs

@@ -1,21 +1,24 @@
 use tokio::await;
 
-use domain::{Channel, ChannelRepository};
+use crate::domain::channel::ChannelRepository;
+use domain::Channel;
 
 use super::ChannelCreateResponse;
 use super::ChannelInput;
+use std::sync::Arc;
 
-pub struct ChannelCreateHandler<'a> {
-    channel_repository: &'a dyn ChannelRepository,
+pub struct ChannelCreateHandler {
+    channel_repository: Arc<dyn ChannelRepository>,
 }
 
-impl<'a> ChannelCreateHandler<'a> {
-    pub fn new(channel_repository: &'a dyn ChannelRepository) -> Self {
+impl ChannelCreateHandler {
+    pub fn new(channel_repository: Arc<dyn ChannelRepository>) -> Self {
         Self { channel_repository }
     }
 }
 
-impl<'a> ChannelCreateHandler<'a> {
+impl ChannelCreateHandler {
+    #[allow(clippy::needless_lifetimes)]
     pub async fn handle(&self, channel_input: ChannelInput) -> Result<ChannelCreateResponse, ()> {
         // @TODO: Creating Channel Validation
 
@@ -28,7 +31,7 @@ impl<'a> ChannelCreateHandler<'a> {
             spec: channel_input.spec,
         };
 
-        let success = await!(self.channel_repository.save(channel)).is_ok();
+        let success = await!(self.channel_repository.create(channel)).is_ok();
 
         Ok(ChannelCreateResponse { success })
     }

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

@@ -1,25 +1,27 @@
 use chrono::Utc;
 use tokio::await;
 
-use domain::{ChannelListParams, ChannelRepository};
+use crate::domain::channel::{ChannelListParams, ChannelRepository};
 
 use super::ChannelListResponse;
+use std::sync::Arc;
 
-pub struct ChannelListHandler<'a> {
+pub struct ChannelListHandler {
     limit_per_page: u32,
-    channel_repository: &'a dyn ChannelRepository,
+    channel_repository: Arc<dyn ChannelRepository>,
 }
 
-impl<'a> ChannelListHandler<'a> {
-    pub fn new(limit_per_page: u32, channel_repository: &'a dyn ChannelRepository) -> Self {
+impl ChannelListHandler {
+    pub fn new(limit_per_page: u32, channel_repository: Arc<dyn ChannelRepository>) -> Self {
         Self {
             limit_per_page,
             channel_repository,
         }
     }
 }
 
-impl<'a> ChannelListHandler<'a> {
+impl ChannelListHandler {
+    #[allow(clippy::needless_lifetimes)]
     pub async fn handle(
         &self,
         page: u32,

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

@@ -0,0 +1 @@
+pub mod channel;