//! //! 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_composition: compositions::site::composition::Composition, member_level_repository: repositories::member_level::repository::Repository, } 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_composition: compositions::site::composition::Composition::new(), member_level_repository: repositories::member_level::repository::Repository::new(), } } pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box> { futures::try_join!( self.list_member_levels(), self.create_member_level(), self.update_member_level(), self.delete_member_level(), ) .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 create_member_level(&self) -> Result<(), Box> { let s = self .connection_broker .queue_subscribe( bpr::ss::member::member_level::SUBJECT_CREATE_MEMBER_LEVEL, self.queue_broker.as_str(), ) .await?; while let Some(message) = s.next().await { if let Err(e) = async { let req = bpr::ss::member::member_level::CreateMemberLevelRequest::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 .member_level_repository .insert( &conn, &repositories::member_level::models::NewMemberLevel { name: req.name, sort_order: req.sort_order as i32, }, ) .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::member::member_level::CreateMemberLevelResponse { error: None, result: Some( bpr::ss::member::member_level::create_member_level_response::Result { member_level: Some(bpr::models::member::MemberLevel::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::member::member_level::CreateMemberLevelResponse { error: Some(bpr::protobuf::rpc::Error::from(e)), result: None, } .encode_to_vec(), ) .await?; } } Ok(()) } async fn list_member_levels(&self) -> Result<(), Box> { let s = self .connection_broker .queue_subscribe( bpr::ss::member::member_level::SUBJECT_LIST_MEMBER_LEVELS, self.queue_broker.as_str(), ) .await?; while let Some(message) = s.next().await { if let Err(e) = async { let req = bpr::ss::member::member_level::ListMemberLevelsRequest::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 find_all = repositories::member_level::models::FindAll { name_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(), ), }; let count = self .member_level_repository .select_all_count(&conn, &find_all) .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 .member_level_repository .select_all(&conn, &find_all) .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::member::member_level::ListMemberLevelsResponse { error: None, result: Some( bpr::ss::member::member_level::list_member_levels_response::Result { member_levels: list .iter() .map(|d| bpr::models::member::MemberLevel::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::member::member_level::ListMemberLevelsResponse { error: Some(bpr::protobuf::rpc::Error::from(e)), result: None, } .encode_to_vec(), ) .await?; } } Ok(()) } async fn update_member_level(&self) -> Result<(), Box> { let s = self .connection_broker .queue_subscribe( bpr::ss::member::member_level::SUBJECT_UPDATE_MEMBER_LEVEL, self.queue_broker.as_str(), ) .await?; while let Some(message) = s.next().await { if let Err(e) = async { let req = bpr::ss::member::member_level::UpdateMemberLevelRequest::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 .member_level_repository .update( &conn, id, &repositories::member_level::models::ModifyMemberLevel { name: req.name, sort_order: req.sort_order as i32, }, ) .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 .member_level_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::member::member_level::UpdateMemberLevelResponse { error: None, result: Some( bpr::ss::member::member_level::update_member_level_response::Result { member_level: Some(bpr::models::member::MemberLevel::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::member::member_level::UpdateMemberLevelResponse { error: Some(bpr::protobuf::rpc::Error::from(e)), result: None, } .encode_to_vec(), ) .await?; } } Ok(()) } async fn delete_member_level(&self) -> Result<(), Box> { let s = self .connection_broker .queue_subscribe( bpr::ss::member::member_level::SUBJECT_DELETE_MEMBER_LEVEL, self.queue_broker.as_str(), ) .await?; while let Some(message) = s.next().await { if let Err(e) = async { let req = bpr::ss::member::member_level::DeleteMemberLevelRequest::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 .member_level_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::member::member_level::DeleteMemberLevelResponse { error: None, result: Some(bpr::ss::member::member_level::delete_member_level_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::member::member_level::DeleteMemberLevelResponse { error: Some(bpr::protobuf::rpc::Error::from(e)), result: None, } .encode_to_vec(), ) .await?; } } Ok(()) } }