events are added

.devcontainer/Dockerfile

@@ -1,9 +1,11 @@
-# See here for image contents: https://github.com/microsoft/vscode-dev-containers/tree/v0.245.0/containers/rust/.devcontainer/base.Dockerfile
+# See here for image contents: https://github.com/microsoft/vscode-dev-containers/tree/v0.245.0/containers/debian/.devcontainer/base.Dockerfile
 
-# [Choice] Debian OS version (use bullseye on local arm64/Apple Silicon): buster, bullseye
+# [Choice] Debian version (use bullseye on local arm64/Apple Silicon): bullseye, buster
 ARG VARIANT="buster"
-FROM mcr.microsoft.com/vscode/devcontainers/rust:0-${VARIANT}
+FROM mcr.microsoft.com/vscode/devcontainers/base:0-${VARIANT}
 
-# [Optional] Uncomment this section to install additional packages.
+# ** [Optional] Uncomment this section to install additional packages. **
 # RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
 #     && apt-get -y install --no-install-recommends <your-package-list-here>
+RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
+  && apt-get -y install --no-install-recommends libpq-dev

.devcontainer/devcontainer.json

@@ -7,7 +7,7 @@
 		"args": {
 			// Use the VARIANT arg to pick a Debian OS version: buster, bullseye
 			// Use bullseye when on local on arm64/Apple Silicon.
-			"VARIANT": "buster"
+			"VARIANT": "bullseye"
 		}
 	},
 	"runArgs": [
@@ -40,6 +40,10 @@
 			},
 			// Add the IDs of extensions you want installed when the container is created.
 			"extensions": [
+				"donjayamanne.githistory",
+				"eamodio.gitlens",
+				"mhutchie.git-graph",
+				"ms-azuretools.vscode-docker",
 				"mutantdino.resourcemonitor",
 				"rust-lang.rust-analyzer",
 				"serayuzgur.crates",
@@ -56,6 +60,7 @@
 	// Comment out to connect as root instead. More info: https://aka.ms/vscode-remote/containers/non-root.
 	"remoteUser": "vscode",
 	"features": {
+		"git": "latest",
 		"rust": "latest"
 	}
 }

Cargo.toml

@@ -29,7 +29,7 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 tokio-cron-scheduler = { version = "*" }
 uuid = { version = "0", features = ["serde", "v4", "v5"] }
 
-beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.78-snapshot" }
+beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.79-snapshot" }
-beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.63-snapshot" }
+beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.64-snapshot" }
 
 [build-dependencies]

src/events/member_bank_deposit/event.rs

@@ -0,0 +1,208 @@
+use crate::api;
+use crate::core;
+use crate::repositories;
+use crate::synchronizations;
+use beteran_common_rust as bcr;
+use beteran_protobuf_rust as bpr;
+use diesel::{
+  r2d2::{ConnectionManager, Pool},
+  PgConnection,
+};
+use prost::Message;
+use std::str::FromStr;
+
+pub struct EventHandler {
+  connection_broker: nats::asynk::Connection,
+  queue_broker: String,
+  pool: Pool<ConnectionManager<PgConnection>>,
+  api_config: core::config::ApiConfig,
+  member_repository: repositories::member::repository::Repository,
+  member_account_api: api::member_account::api::Api,
+  member_account_synchronizer: synchronizations::member_account::synchronizer::Synchronizer,
+}
+
+impl std::fmt::Debug for EventHandler {
+  fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+    f.debug_struct("EventHandler of members").finish()
+  }
+}
+
+impl EventHandler {
+  ///
+  pub fn new(
+    connection_broker: nats::asynk::Connection,
+    queue_broker: String,
+    pool: Pool<ConnectionManager<PgConnection>>,
+    member_account_synchronizer: synchronizations::member_account::synchronizer::Synchronizer,
+    api_config: core::config::ApiConfig,
+  ) -> EventHandler {
+    EventHandler {
+      connection_broker,
+      queue_broker,
+      pool,
+      member_account_synchronizer,
+      api_config: api_config.clone(),
+      member_repository: repositories::member::repository::Repository::new(),
+      member_account_api: api::member_account::api::Api::new(api_config.clone()),
+    }
+  }
+
+  pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box<dyn std::error::Error>> {
+    futures::try_join!(self.event_after_update_member_bank_deposit_for_state()).map(|_| ())
+  }
+
+  async fn event_after_update_member_bank_deposit_for_state(
+    &self,
+  ) -> Result<(), Box<dyn std::error::Error>> {
+    let s = self
+      .connection_broker
+      .queue_subscribe(
+        bpr::ss::member_bank_deposit::EVENT_SUBJECT_AFTER_UPDATE_MEMBER_BANK_DEPOSIT_FOR_STATE,
+        self.queue_broker.as_str(),
+      )
+      .await?;
+
+    while let Some(message) = s.next().await {
+      if let Err(e) = async {
+        let eve = bpr::ss::member_bank_deposit::AfterUpdateMemberBankDepositForState::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 eve.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 event = match eve.event {
+          Some(r) => r,
+          None => {
+            return Err(bcr::error::rpc::Error::InvalidParams(
+              bcr::error::rpc::InvalidParams {
+                message: "invalid event information".to_string(),
+                detail: bcr::error::rpc::InvalidParamsDetail {
+                  location: "event".to_string(),
+                  param: "event".to_string(),
+                  value: "".to_string(),
+                  error_type: bcr::error::rpc::InvalidParamsType::Required,
+                  message: "".to_string(),
+                },
+              },
+            ));
+          }
+        };
+        let member_bank_deposit = match event.member_bank_deposit {
+          Some(c) => c,
+          None => {
+            return Err(bcr::error::rpc::Error::InvalidParams(
+              bcr::error::rpc::InvalidParams {
+                message: "invalid event information".to_string(),
+                detail: bcr::error::rpc::InvalidParamsDetail {
+                  location: "event".to_string(),
+                  param: "member_bank_deposit".to_string(),
+                  value: "".to_string(),
+                  error_type: bcr::error::rpc::InvalidParamsType::Required,
+                  message: "".to_string(),
+                },
+              },
+            ));
+          }
+        };
+
+        let member = match member_bank_deposit.member {
+          Some(c) => c,
+          None => {
+            return Err(bcr::error::rpc::Error::InvalidParams(
+              bcr::error::rpc::InvalidParams {
+                message: "invalid event information".to_string(),
+                detail: bcr::error::rpc::InvalidParamsDetail {
+                  location: "event".to_string(),
+                  param: "member_bank_deposit.member".to_string(),
+                  value: "".to_string(),
+                  error_type: bcr::error::rpc::InvalidParamsType::Required,
+                  message: "".to_string(),
+                },
+              },
+            ));
+          }
+        };
+        let member_id = uuid::Uuid::from_str(member.id.as_str()).map_err(|e| {
+          bcr::error::rpc::Error::InvalidParams(bcr::error::rpc::InvalidParams {
+            message: "invalid member.id param".to_string(),
+            detail: bcr::error::rpc::InvalidParamsDetail {
+              location: "request".to_string(),
+              param: "member.id".to_string(),
+              value: member.id.clone(),
+              error_type: bcr::error::rpc::InvalidParamsType::Required,
+              message: e.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,
+          })
+        })?;
+
+        match beteran_protobuf_rust::models::member_bank_deposit::MemberBankDepositState::from_i32(
+          member_bank_deposit.state,
+        ) {
+          Some(s) => {
+            if s == beteran_protobuf_rust::models::member_bank_deposit::MemberBankDepositState::Complete {
+              self.member_account_api.create_deposit(api::member_account::models::CreateDepositRequest{
+                username: member.username.clone(),
+                cash_type: None,
+                amount: member_bank_deposit.amount as i64,
+                request_key: None,
+              }).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,
+                })
+              })?;
+
+              self.member_account_synchronizer.balance_for_user(member_id, &member.username).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,
+                })
+              })?;
+            }
+          }
+          None => {}
+        };
+
+        Ok::<(), bcr::error::rpc::Error>(())
+      }
+      .await
+      {
+        println!("error: {}", e);
+      }
+    }
+
+    Ok(())
+  }
+}

src/events/member_bank_deposit/mod.rs

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

src/events/member_bank_withdraw/event.rs

@@ -0,0 +1,206 @@
+use crate::api;
+use crate::core;
+use crate::repositories;
+use crate::synchronizations;
+use beteran_common_rust as bcr;
+use beteran_protobuf_rust as bpr;
+use diesel::{
+  r2d2::{ConnectionManager, Pool},
+  PgConnection,
+};
+use prost::Message;
+use std::str::FromStr;
+
+pub struct EventHandler {
+  connection_broker: nats::asynk::Connection,
+  queue_broker: String,
+  pool: Pool<ConnectionManager<PgConnection>>,
+  api_config: core::config::ApiConfig,
+  member_repository: repositories::member::repository::Repository,
+  member_account_api: api::member_account::api::Api,
+  member_account_synchronizer: synchronizations::member_account::synchronizer::Synchronizer,
+}
+
+impl std::fmt::Debug for EventHandler {
+  fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+    f.debug_struct("EventHandler of members").finish()
+  }
+}
+
+impl EventHandler {
+  ///
+  pub fn new(
+    connection_broker: nats::asynk::Connection,
+    queue_broker: String,
+    pool: Pool<ConnectionManager<PgConnection>>,
+    member_account_synchronizer: synchronizations::member_account::synchronizer::Synchronizer,
+    api_config: core::config::ApiConfig,
+  ) -> EventHandler {
+    EventHandler {
+      connection_broker,
+      queue_broker,
+      pool,
+      member_account_synchronizer,
+      api_config: api_config.clone(),
+      member_repository: repositories::member::repository::Repository::new(),
+      member_account_api: api::member_account::api::Api::new(api_config.clone()),
+    }
+  }
+
+  pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box<dyn std::error::Error>> {
+    futures::try_join!(self.event_after_update_member_bank_withdraw_for_state()).map(|_| ())
+  }
+
+  async fn event_after_update_member_bank_withdraw_for_state(
+    &self,
+  ) -> Result<(), Box<dyn std::error::Error>> {
+    let s = self
+      .connection_broker
+      .queue_subscribe(
+        bpr::ss::member_bank_withdraw::EVENT_SUBJECT_AFTER_UPDATE_MEMBER_BANK_WITHDRAW_FOR_STATE,
+        self.queue_broker.as_str(),
+      )
+      .await?;
+
+    while let Some(message) = s.next().await {
+      if let Err(e) = async {
+        let eve = bpr::ss::member_bank_withdraw::AfterUpdateMemberBankWithdrawForState::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 eve.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 event = match eve.event {
+          Some(r) => r,
+          None => {
+            return Err(bcr::error::rpc::Error::InvalidParams(
+              bcr::error::rpc::InvalidParams {
+                message: "invalid event information".to_string(),
+                detail: bcr::error::rpc::InvalidParamsDetail {
+                  location: "event".to_string(),
+                  param: "event".to_string(),
+                  value: "".to_string(),
+                  error_type: bcr::error::rpc::InvalidParamsType::Required,
+                  message: "".to_string(),
+                },
+              },
+            ));
+          }
+        };
+        let member_bank_withdraw = match event.member_bank_withdraw {
+          Some(c) => c,
+          None => {
+            return Err(bcr::error::rpc::Error::InvalidParams(
+              bcr::error::rpc::InvalidParams {
+                message: "invalid event information".to_string(),
+                detail: bcr::error::rpc::InvalidParamsDetail {
+                  location: "event".to_string(),
+                  param: "member_bank_withdraw".to_string(),
+                  value: "".to_string(),
+                  error_type: bcr::error::rpc::InvalidParamsType::Required,
+                  message: "".to_string(),
+                },
+              },
+            ));
+          }
+        };
+
+        let member = match member_bank_withdraw.member {
+          Some(c) => c,
+          None => {
+            return Err(bcr::error::rpc::Error::InvalidParams(
+              bcr::error::rpc::InvalidParams {
+                message: "invalid event information".to_string(),
+                detail: bcr::error::rpc::InvalidParamsDetail {
+                  location: "event".to_string(),
+                  param: "member_bank_withdraw.member".to_string(),
+                  value: "".to_string(),
+                  error_type: bcr::error::rpc::InvalidParamsType::Required,
+                  message: "".to_string(),
+                },
+              },
+            ));
+          }
+        };
+        let member_id = uuid::Uuid::from_str(member.id.as_str()).map_err(|e| {
+          bcr::error::rpc::Error::InvalidParams(bcr::error::rpc::InvalidParams {
+            message: "invalid member.id param".to_string(),
+            detail: bcr::error::rpc::InvalidParamsDetail {
+              location: "request".to_string(),
+              param: "member.id".to_string(),
+              value: member.id.clone(),
+              error_type: bcr::error::rpc::InvalidParamsType::Required,
+              message: e.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,
+          })
+        })?;
+
+        match beteran_protobuf_rust::models::member_bank_withdraw::MemberBankWithdrawState::from_i32(
+          member_bank_withdraw.state,
+        ) {
+          Some(s) => {
+            if s == beteran_protobuf_rust::models::member_bank_withdraw::MemberBankWithdrawState::Complete {
+              self.member_account_api.create_withdraw(api::member_account::models::CreateWithdrawRequest{
+                username: member.username.clone(),
+                request_key: None,
+              }).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,
+                })
+              })?;
+
+              self.member_account_synchronizer.balance_for_user(member_id, &member.username).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,
+                })
+              })?;
+            }
+          }
+          None => {}
+        };
+
+        Ok::<(), bcr::error::rpc::Error>(())
+      }
+      .await
+      {
+        println!("error: {}", e);
+      }
+    }
+
+    Ok(())
+  }
+}

src/events/member_bank_withdraw/mod.rs

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

src/events/mod.rs

@@ -1 +1,3 @@
 pub mod member;
+pub mod member_bank_withdraw;
+pub mod member_bank_deposit;

src/schedulers/balance/scheduler.rs

@@ -49,7 +49,7 @@ impl Scheduler {
       Box::pin(async move {
         self
           .member_account_synchronizer
-          .balance_for_user()
+          .balance_for_user_all()
           .await
           .expect("member_account_synchronizer.balance_for_user");
       })

src/synchronizations/member_account/synchronizer.rs

@@ -56,7 +56,7 @@ impl Synchronizer {
     }
   }
 
-  pub async fn balance_for_user(&self) -> Result<(), Box<dyn std::error::Error>> {
+  pub async fn balance_for_user_all(&self) -> Result<(), Box<dyn std::error::Error>> {
     let start_at = (chrono::Utc::now()).timestamp();
 
     if let Err(e) = async {
@@ -142,6 +142,68 @@ impl Synchronizer {
     Ok(())
   }
 
+  pub async fn balance_for_user(
+    &self,
+    member_id: uuid::Uuid,
+    usename: &str,
+  ) -> Result<(), Box<dyn std::error::Error>> {
+    let start_at = (chrono::Utc::now()).timestamp();
+
+    if let Err(e) = async {
+      let conn = self.pool.get().expect("conn");
+
+      let m = match self
+        .member_repository
+        .select_by_member_id(&conn, member_id)
+        .expect("member_repository.select_by_member_id")
+      {
+        Some(m) => m,
+        None => {
+          return Ok(());
+        }
+      };
+
+      let req = api::member_account::models::GetBalanceForUserRequest {
+        username: usename.to_string(),
+      };
+      let res = self.member_account_api.get_balance_for_user(req).await?;
+
+      let modify_member = repositories::member::models::ModifyMemberForBalance {
+        balance: res.balance,
+        balance_bota: res.balance_bota,
+        balance_sum: res.balance_sum,
+        companies: res.companies,
+      };
+
+      self
+        .member_repository
+        .update_balance(&conn, m.id, &modify_member)
+        .expect("member_repository.update_balance");
+
+      Ok::<(), api::core::models::Error>(())
+    }
+    .await
+    {
+      let conn = self.pool.get().expect("conn");
+
+      self
+        .synchronization_history_repository
+        .insert(
+          &conn,
+          &repositories::synchronization_history::models::NewSynchronizationHistory {
+            item: repositories::synchronization::models::ITEM_BALANCE_USER.to_string(),
+            start_at,
+            complete_at: (chrono::Utc::now()).timestamp(),
+            code: e.code,
+            message: e.msg,
+          },
+        )
+        .expect("synchronization_history insert");
+    }
+
+    Ok(())
+  }
+
   pub async fn balance_for_partner(&self) -> Result<(), Box<dyn std::error::Error>> {
     let start_at = (chrono::Utc::now()).timestamp();
 
@@ -219,6 +281,4 @@ impl Synchronizer {
 
     Ok(())
   }
-
-
 }