diff --git a/Cargo.toml b/Cargo.toml index 44e21ab..d23d070 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,7 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread"] } tokio-cron-scheduler = { version = "0" } uuid = { version = "0", features = ["serde", "v4", "v5"] } -beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.25-snapshot" } -beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.5-snapshot" } +beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.27-snapshot" } +beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.10-snapshot" } [build-dependencies] diff --git a/src/compositions/identity/composition.rs b/src/compositions/identity/composition.rs index 134f5a6..8ffa83e 100644 --- a/src/compositions/identity/composition.rs +++ b/src/compositions/identity/composition.rs @@ -39,21 +39,8 @@ impl Composition { let expiration_at = issued_at + Duration::days(1); let claims = models::Claims { iss: issuer, - iat: issued_at - .date() - .with_hms_milli(issued_at.hour(), issued_at.minute(), issued_at.second(), 0) - .unwrap() - .assume_utc(), - exp: expiration_at - .date() - .with_hms_milli( - expiration_at.hour(), - expiration_at.minute(), - expiration_at.second(), - 0, - ) - .unwrap() - .assume_utc(), + iat: issued_at.unix_timestamp(), + exp: expiration_at.unix_timestamp(), session_id, }; diff --git a/src/compositions/identity/models.rs b/src/compositions/identity/models.rs index 108fab8..a2444cb 100644 --- a/src/compositions/identity/models.rs +++ b/src/compositions/identity/models.rs @@ -1,5 +1,4 @@ use serde::{Deserialize, Serialize}; -use time::OffsetDateTime; // #[derive(Debug, Serialize, Deserialize)] // struct Claims { @@ -14,7 +13,7 @@ use time::OffsetDateTime; #[derive(Debug, Serialize, Deserialize)] pub struct Claims { pub iss: String, - pub iat: OffsetDateTime, - pub exp: OffsetDateTime, + pub iat: i64, + pub exp: i64, pub session_id: String, } diff --git a/src/main.rs b/src/main.rs index ae4c1dc..7bd9c6c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -67,10 +67,15 @@ async fn main() -> Result<(), Box> { password_salt.clone(), jwt_secret.clone(), ); + let site_service = services::site::service::Service::new( + connection_server_broker.clone(), + queue_server_broker.clone(), + pool.clone(), + ); println!("Server service [beteran-server-service] is started"); - futures::try_join!(identity_service.subscribe(),)?; + futures::try_join!(site_service.subscribe(), identity_service.subscribe(),)?; Ok(()) } diff --git a/src/repositories/site/models.rs b/src/repositories/site/models.rs index 7603b5a..ce5d8a7 100644 --- a/src/repositories/site/models.rs +++ b/src/repositories/site/models.rs @@ -21,6 +21,30 @@ pub struct Site { pub deleted_at: Option, } +/// +#[derive(Insertable, Debug, Clone)] +#[table_name = "sites"] +pub struct NewSite { + /// + pub url: String, + /// + pub show: bool, + /// + pub can_use: bool, +} + +/// +#[derive(AsChangeset, Debug, Clone)] +#[table_name = "sites"] +pub struct ModifySite { + /// + pub url: String, + /// + pub show: bool, + /// + pub can_use: bool, +} + /// #[derive(Debug, Clone)] pub struct FindAll { diff --git a/src/repositories/site/repository.rs b/src/repositories/site/repository.rs index 1f0347f..936095a 100644 --- a/src/repositories/site/repository.rs +++ b/src/repositories/site/repository.rs @@ -26,6 +26,19 @@ impl Repository { Repository {} } + /// + pub fn insert( + &self, + conn: &diesel::PgConnection, + new_member: &models::NewSite, + ) -> Result { + let inserted = diesel::insert_into(sites::table) + .values(new_member) + .get_result::(conn)?; + + Ok(inserted) + } + /// pub fn select( &self, @@ -147,4 +160,28 @@ impl Repository { q.load::(conn) } + + /// + pub fn update( + &self, + conn: &diesel::PgConnection, + id: uuid::Uuid, + modify: &models::ModifySite, + ) -> Result { + use sites::dsl; + + diesel::update(dsl::sites.filter(dsl::id.eq(id))) + .set(modify) + .execute(conn) + .map(|c| c as u64) + } + + /// + pub fn delete(&self, conn: &diesel::PgConnection, id: uuid::Uuid) -> Result { + use sites::dsl; + + diesel::delete(sites::table.filter(dsl::id.eq(id))) + .execute(conn) + .map(|c| c as u64) + } } diff --git a/src/services/mod.rs b/src/services/mod.rs index db53a0c..7ed452c 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -1 +1,2 @@ pub mod identity; +pub mod site; diff --git a/src/services/site/mod.rs b/src/services/site/mod.rs new file mode 100644 index 0000000..b9e52db --- /dev/null +++ b/src/services/site/mod.rs @@ -0,0 +1,2 @@ +pub mod models; +pub mod service; diff --git a/src/services/site/models.rs b/src/services/site/models.rs new file mode 100644 index 0000000..ac235bf --- /dev/null +++ b/src/services/site/models.rs @@ -0,0 +1,16 @@ +use crate::repositories; +use beteran_protobuf_rust as bpr; + +impl From<&repositories::site::models::Site> for bpr::models::domain::Site { + fn from(d: &repositories::site::models::Site) -> Self { + bpr::models::domain::Site { + id: d.id.to_string(), + url: d.url.clone(), + show: d.show, + can_use: d.can_use, + created_at: d.created_at as u64, + updated_at: d.updated_at as u64, + deleted_at: d.deleted_at.map(|d| d as u64), + } + } +} diff --git a/src/services/site/service.rs b/src/services/site/service.rs new file mode 100644 index 0000000..33300ce --- /dev/null +++ b/src/services/site/service.rs @@ -0,0 +1,556 @@ +//! +//! + +use std::str::FromStr; + +use super::models; +use crate::compositions; +use crate::repositories; +use beteran_common_rust as bcr; +use beteran_protobuf_rust as bpr; +use diesel::{ + r2d2::{ConnectionManager, Pool}, + PgConnection, +}; +use prost::Message; + +/// +pub struct Service { + connection_broker: nats::asynk::Connection, + queue_broker: String, + pool: Pool>, + site_repository: repositories::site::repository::Repository, + site_composition: compositions::site::composition::Composition, +} + +impl std::fmt::Debug for Service { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.debug_struct("Service of service.member.service.identity") + .finish() + } +} + +impl Service { + /// + pub fn new( + connection_broker: nats::asynk::Connection, + queue_broker: String, + pool: Pool>, + ) -> Service { + Service { + connection_broker, + queue_broker, + pool, + site_repository: repositories::site::repository::Repository::new(), + site_composition: compositions::site::composition::Composition::new(), + } + } + + pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box> { + futures::try_join!( + self.list_sites(), + self.create_site(), + self.update_site(), + self.delete_site(), + ) + .map(|_| ()) + } + + fn check_site( + &self, + conn: &diesel::PgConnection, + url: Option, + site_id: uuid::Uuid, + ) -> Result { + match self + .site_composition + .select_by_url(conn, url, site_id) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })? { + Some(s) => Ok(s), + None => Err(bcr::error::rpc::Error::InvalidParams( + bcr::error::rpc::InvalidParams { + message: "invalid site_url information".to_string(), + detail: bcr::error::rpc::InvalidParamsDetail { + location: "request".to_string(), + param: "client.site_url".to_string(), + value: "".to_string(), + error_type: bcr::error::rpc::InvalidParamsType::None, + message: "".to_string(), + }, + }, + )), + } + } + + async fn list_sites(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::ss::domain::site::SUBJECT_LIST_SITES, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let req = bpr::ss::domain::site::ListSitesRequest::decode(message.data.as_slice()) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let client = match req.client { + Some(c) => c, + None => { + return Err(bcr::error::rpc::Error::InvalidParams( + bcr::error::rpc::InvalidParams { + message: "invalid client information".to_string(), + detail: bcr::error::rpc::InvalidParamsDetail { + location: "request".to_string(), + param: "client".to_string(), + value: "".to_string(), + error_type: bcr::error::rpc::InvalidParamsType::Required, + message: "".to_string(), + }, + }, + )); + } + }; + + let conn = self.pool.get().map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let count = self + .site_repository + .select_all_count( + &conn, + repositories::site::models::FindAll { + show: None, + can_use: None, + url_like: None, + pagination: req + .pagination + .as_ref() + .map(|d| bcr::models::pagination::Pagination::from(d)), + sorts: Some( + req + .sorts + .iter() + .map(|d| beteran_common_rust::models::pagination::Sort::from(d)) + .collect(), + ), + }, + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let list = self + .site_repository + .select_all( + &conn, + repositories::site::models::FindAll { + show: None, + can_use: None, + url_like: None, + pagination: req + .pagination + .as_ref() + .map(|d| bcr::models::pagination::Pagination::from(d)), + sorts: Some( + req + .sorts + .iter() + .map(|d| beteran_common_rust::models::pagination::Sort::from(d)) + .collect(), + ), + }, + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + message + .respond( + bpr::ss::domain::site::ListSitesResponse { + error: None, + result: Some(bpr::ss::domain::site::list_sites_response::Result { + sites: list + .iter() + .map(|d| bpr::models::domain::Site::from(d)) + .collect(), + }), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::ss::domain::site::ListSitesResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn create_site(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::ss::domain::site::SUBJECT_CREATE_SITE, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let req = bpr::ss::domain::site::CreateSiteRequest::decode(message.data.as_slice()) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let client = match req.client { + Some(c) => c, + None => { + return Err(bcr::error::rpc::Error::InvalidParams( + bcr::error::rpc::InvalidParams { + message: "invalid client information".to_string(), + detail: bcr::error::rpc::InvalidParamsDetail { + location: "request".to_string(), + param: "client".to_string(), + value: "".to_string(), + error_type: bcr::error::rpc::InvalidParamsType::Required, + message: "".to_string(), + }, + }, + )); + } + }; + + let conn = self.pool.get().map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let s = self + .site_repository + .insert( + &conn, + &repositories::site::models::NewSite { + url: req.url, + show: req.show, + can_use: req.can_use, + }, + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + message + .respond( + bpr::ss::domain::site::CreateSiteResponse { + error: None, + result: Some(bpr::ss::domain::site::create_site_response::Result { + site: Some(bpr::models::domain::Site::from(&s)), + }), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::ss::domain::site::CreateSiteResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_site(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::ss::domain::site::SUBJECT_UPDATE_SITE, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let req = bpr::ss::domain::site::UpdateSiteRequest::decode(message.data.as_slice()) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let id = uuid::Uuid::from_str(req.id.as_str()).map_err(|e| { + bcr::error::rpc::Error::InvalidParams(bcr::error::rpc::InvalidParams { + message: "invalid id param".to_string(), + detail: bcr::error::rpc::InvalidParamsDetail { + location: "request".to_string(), + param: "id".to_string(), + value: req.id.clone(), + error_type: bcr::error::rpc::InvalidParamsType::Required, + message: e.to_string(), + }, + }) + })?; + + let client = match req.client { + Some(c) => c, + None => { + return Err(bcr::error::rpc::Error::InvalidParams( + bcr::error::rpc::InvalidParams { + message: "invalid client information".to_string(), + detail: bcr::error::rpc::InvalidParamsDetail { + location: "request".to_string(), + param: "client".to_string(), + value: "".to_string(), + error_type: bcr::error::rpc::InvalidParamsType::Required, + message: "".to_string(), + }, + }, + )); + } + }; + + let conn = self.pool.get().map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let _affected = self + .site_repository + .update( + &conn, + id, + &repositories::site::models::ModifySite { + url: req.url, + show: req.show, + can_use: req.can_use, + }, + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let s = self.site_repository.select(&conn, id).map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + message + .respond( + bpr::ss::domain::site::UpdateSiteResponse { + error: None, + result: Some(bpr::ss::domain::site::update_site_response::Result { + site: s.map(|d| bpr::models::domain::Site::from(&d)), + }), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::ss::domain::site::UpdateSiteResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn delete_site(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::ss::domain::site::SUBJECT_DELETE_SITE, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let req = bpr::ss::domain::site::DeleteSiteRequest::decode(message.data.as_slice()) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let id = uuid::Uuid::from_str(req.id.as_str()).map_err(|e| { + bcr::error::rpc::Error::InvalidParams(bcr::error::rpc::InvalidParams { + message: "invalid id param".to_string(), + detail: bcr::error::rpc::InvalidParamsDetail { + location: "request".to_string(), + param: "id".to_string(), + value: req.id.clone(), + error_type: bcr::error::rpc::InvalidParamsType::Required, + message: e.to_string(), + }, + }) + })?; + + let client = match req.client { + Some(c) => c, + None => { + return Err(bcr::error::rpc::Error::InvalidParams( + bcr::error::rpc::InvalidParams { + message: "invalid client information".to_string(), + detail: bcr::error::rpc::InvalidParamsDetail { + location: "request".to_string(), + param: "client".to_string(), + value: "".to_string(), + error_type: bcr::error::rpc::InvalidParamsType::Required, + message: "".to_string(), + }, + }, + )); + } + }; + + let conn = self.pool.get().map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let _affected = self.site_repository.delete(&conn, id).map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + message + .respond( + bpr::ss::domain::site::DeleteSiteResponse { + error: None, + result: Some(bpr::ss::domain::site::delete_site_response::Result {}), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::ss::domain::site::DeleteSiteResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } +}