use beteran_common_rust as bcr;
use beteran_protobuf_rust as bpr;
use prost::Message;

///
pub struct Service {
  connection_broker: nats::asynk::Connection,
  queue_broker: String,
}

impl std::fmt::Debug for Service {
  fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    f.debug_struct("beteran-backend-server-edge::identity::service::Service")
      .finish()
  }
}

impl Service {
  ///
  pub fn new(connection_broker: nats::asynk::Connection, queue_broker: String) -> Service {
    Service {
      connection_broker,
      queue_broker,
    }
  }

  pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box<dyn std::error::Error>> {
    futures::try_join!(
      self.create_member(),
      self.list_members(),
      self.get_member(),
      self.get_member_by_username(),
      self.update_member(),
      self.update_member_for_password(),
      self.update_member_for_state(),
      self.delete_member(),
    )
    .map(|_| ())
  }

  fn get_client_in_header(
    &self,
    message: &nats::asynk::Message,
  ) -> Result<bpr::models::core::network::Client, bcr::error::rpc::Error> {
    match &message.headers {
      Some(headers) => {
        let client = match headers.get(bpr::c2se::core::network::HEADER_CLIENT) {
          Some(c) => {
            let msg = base64::decode(c).map_err(|e| {
              bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse {
                message: format!("invalid header: {}", e),
              })
            })?;
            bpr::models::core::network::Client::decode(msg.as_slice()).map_err(|e| {
              bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse {
                message: format!("invalid header: {}", e),
              })
            })?
          }
          None => {
            return Err(bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse {
              message: "invalid client information".to_string(),
            }));
          }
        };

        Ok(client)
      }
      None => Err(bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse {
        message: "invalid header".to_string(),
      })),
    }
  }

  async fn create_member(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_CREATE_MEMBER,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req =
          bpr::c2se::member::CreateMemberRequest::decode(message.data.as_slice()).map_err(|e| {
            bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
              message: format!("invalid request: {}", e),
            })
          })?;

        let bank_account = match req.bank_account {
          Some(a) => Some(
            bpr::ss::member::create_member_request::request::BankAccount {
              bank_id: a.bank_id,
              name: a.name,
              account_number: a.account_number,
              exchange_password: a.exchange_password,
            },
          ),
          None => None,
        };

        let game_setting =
          req.game_setting.map(
            |a| bpr::ss::member::create_member_request::request::GameSetting {
              can_bet_casino: a.can_bet_casino,
              can_bet_slot: a.can_bet_slot,
              can_bet_powerball: a.can_bet_powerball,
              can_bet_powerladder: a.can_bet_powerladder,
              can_bet_eos: a.can_bet_eos,
              can_bet_bogglepowerball: a.can_bet_bogglepowerball,
            },
          );

        let settlement_setting = req.settlement_setting.map(|a| {
          bpr::ss::member::create_member_request::request::SettlementSetting {
            can_exchange: a.can_exchange,
            can_first_charge_comp: a.can_first_charge_comp,
            can_per_charge_comp: a.can_per_charge_comp,
            manual_payment_type_for_partner: a.manual_payment_type_for_partner,
            settlement_type: a.settlement_type,
            rate_casino: a.rate_casino,
            rate_casino_loosing: a.rate_casino_loosing,
            rate_casino_bacara: a.rate_casino_bacara,
            rate_casino_roulette: a.rate_casino_roulette,
            rate_casino_dragon_tiger: a.rate_casino_dragon_tiger,
            rate_slot: a.rate_slot,
            rate_slot_loosing: a.rate_slot_loosing,
            rate_powerball_single: a.rate_powerball_single,
            rate_powerball_combo: a.rate_powerball_combo,
            rate_powerladder_single: a.rate_powerladder_single,
            rate_powerladder_combo: a.rate_powerladder_combo,
            rate_eos_single: a.rate_eos_single,
            rate_eos_combo: a.rate_eos_combo,
            rate_bogglepowerball_single: a.rate_bogglepowerball_single,
            rate_bogglepowerball_combo: a.rate_bogglepowerball_combo,
          }
        });

        let ss_create_member_req = bpr::ss::member::CreateMemberRequest {
          client: Some(client),
          request: Some(bpr::ss::member::create_member_request::Request {
            site_id: req.site_id,
            member_level_id: req.member_level_id,
            mobile_phone_number: req.mobile_phone_number,
            member_class_id: req.member_class_id,
            referrer_member_username: req.referrer_member_username,
            username: req.username,
            password: req.password,
            nickname: req.nickname,
            state: bpr::models::member::MemberState::Normal as i32,
            bank_account,
            game_setting,
            settlement_setting,
          }),
        };

        let ss_create_member_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_CREATE_MEMBER,
            ss_create_member_req.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,
            })
          })?;

        let ss_create_member_res =
          bpr::ss::member::CreateMemberResponse::decode(ss_create_member_res_msg.data.as_slice())
            .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,
            })
          })?;

        if let Some(e) = ss_create_member_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(r) = ss_create_member_res.result {
          message
            .respond(
              bpr::c2se::member::CreateMemberResponse {
                error: None,
                result: Some(bpr::c2se::member::create_member_response::Result {
                  member: r.member,
                }),
              }
              .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::c2se::member::CreateMemberResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }

  async fn list_members(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_LIST_MEMBERS,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req =
          bpr::c2se::member::ListMembersRequest::decode(message.data.as_slice()).map_err(|e| {
            bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
              message: format!("invalid request: {}", e),
            })
          })?;

        let ss_list_members_req = bpr::ss::member::ListMembersRequest {
          client: Some(client),
          request: Some(bpr::ss::member::list_members_request::Request {
            pagination: req.pagination,
            sorts: req.sorts,
            search: match req.search {
              Some(s) => Some(bpr::ss::member::list_members_request::request::Search {
                site_id: s.site_id,
                member_class_id: s.member_class_id,
                member_level_id: s.member_level_id,
                parent_member_id: s.parent_member_id,
                username: s.username,
                usernames: s.usernames,
                username_like: s.username_like,
                nickname: s.nickname,
                nicknames: s.nicknames,
                nickname_like: s.nickname_like,
                mobile_phone_number_like: s.mobile_phone_number_like,
                last_signined_ip: s.last_signined_ip,
                state: s.state,
                deleted_at: s.deleted_at,
              }),
              None => None,
            },
          }),
        };

        let ss_list_members_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_LIST_MEMBERS,
            ss_list_members_req.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,
            })
          })?;

        let ss_list_members_res =
          bpr::ss::member::ListMembersResponse::decode(ss_list_members_res_msg.data.as_slice())
            .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,
              })
            })?;

        if let Some(e) = ss_list_members_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(r) = ss_list_members_res.result {
          message
            .respond(
              bpr::c2se::member::ListMembersResponse {
                error: None,
                result: Some(bpr::c2se::member::list_members_response::Result {
                  total_count: r.total_count,
                  members: r.members,
                }),
              }
              .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::c2se::member::ListMembersResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }

  async fn get_member(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_GET_MEMBER,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req =
          bpr::c2se::member::GetMemberRequest::decode(message.data.as_slice()).map_err(|e| {
            bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
              message: format!("invalid request: {}", e),
            })
          })?;

        let ss_get_member_req = bpr::ss::member::GetMemberRequest {
          client: Some(client),
          request: Some(bpr::ss::member::get_member_request::Request { id: req.id }),
        };

        let ss_get_member_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_GET_MEMBER,
            ss_get_member_req.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,
            })
          })?;

        let ss_get_member_res =
          bpr::ss::member::GetMemberResponse::decode(ss_get_member_res_msg.data.as_slice())
            .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,
              })
            })?;

        if let Some(e) = ss_get_member_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(r) = ss_get_member_res.result {
          message
            .respond(
              bpr::c2se::member::GetMemberResponse {
                error: None,
                result: Some(bpr::c2se::member::get_member_response::Result { member: r.member }),
              }
              .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::c2se::member::GetMemberResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }

  async fn get_member_by_username(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_GET_MEMBER_BY_USERNAME,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req = bpr::c2se::member::GetMemberByUsernameRequest::decode(message.data.as_slice())
          .map_err(|e| {
            bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
              message: format!("invalid request: {}", e),
            })
          })?;

        let ss_get_member_by_username_req = bpr::ss::member::GetMemberByUsernameRequest {
          client: Some(client),
          request: Some(bpr::ss::member::get_member_by_username_request::Request {
            username: req.username,
          }),
        };

        let ss_get_member_by_username_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_GET_MEMBER_BY_USERNAME,
            ss_get_member_by_username_req.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,
            })
          })?;

        let ss_get_member_by_username_res = bpr::ss::member::GetMemberByUsernameResponse::decode(
          ss_get_member_by_username_res_msg.data.as_slice(),
        )
        .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,
          })
        })?;

        if let Some(e) = ss_get_member_by_username_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(r) = ss_get_member_by_username_res.result {
          message
            .respond(
              bpr::c2se::member::GetMemberByUsernameResponse {
                error: None,
                result: Some(bpr::c2se::member::get_member_by_username_response::Result {
                  member: r.member,
                }),
              }
              .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::c2se::member::GetMemberByUsernameResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }

  async fn update_member(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_UPDATE_MEMBER,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req =
          bpr::c2se::member::UpdateMemberRequest::decode(message.data.as_slice()).map_err(|e| {
            bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
              message: format!("invalid request: {}", e),
            })
          })?;

        let bank_account = match req.bank_account {
          Some(a) => Some(
            bpr::ss::member::update_member_request::request::BankAccount {
              id: a.id,
              bank_id: a.bank_id,
              name: a.name,
              account_number: a.account_number,
              memo: a.memo,
            },
          ),
          None => None,
        };

        let game_setting = match req.game_setting {
          Some(a) => Some(
            bpr::ss::member::update_member_request::request::GameSetting {
              id: a.id,
              can_bet_casino: a.can_bet_casino,
              can_bet_slot: a.can_bet_slot,
              can_bet_powerball: a.can_bet_powerball,
              can_bet_powerladder: a.can_bet_powerladder,
              can_bet_eos: a.can_bet_eos,
              can_bet_bogglepowerball: a.can_bet_bogglepowerball,
            },
          ),
          None => None,
        };

        let settlement_setting = match req.settlement_setting {
          Some(a) => Some(
            bpr::ss::member::update_member_request::request::SettlementSetting {
              id: a.id,
              can_exchange: a.can_exchange,
              can_first_charge_comp: a.can_first_charge_comp,
              can_per_charge_comp: a.can_per_charge_comp,
              manual_payment_type_for_partner: a.manual_payment_type_for_partner,
              settlement_type: a.settlement_type,
              rate_casino: a.rate_casino,
              rate_casino_loosing: a.rate_casino_loosing,
              rate_casino_bacara: a.rate_casino_bacara,
              rate_casino_roulette: a.rate_casino_roulette,
              rate_casino_dragon_tiger: a.rate_casino_dragon_tiger,
              rate_slot: a.rate_slot,
              rate_slot_loosing: a.rate_slot_loosing,
              rate_powerball_single: a.rate_powerball_single,
              rate_powerball_combo: a.rate_powerball_combo,
              rate_powerladder_single: a.rate_powerladder_single,
              rate_powerladder_combo: a.rate_powerladder_combo,
              rate_eos_single: a.rate_eos_single,
              rate_eos_combo: a.rate_eos_combo,
              rate_bogglepowerball_single: a.rate_bogglepowerball_single,
              rate_bogglepowerball_combo: a.rate_bogglepowerball_combo,
            },
          ),
          None => None,
        };

        let ss_update_member_req = bpr::ss::member::UpdateMemberRequest {
          client: Some(client),
          request: Some(bpr::ss::member::update_member_request::Request {
            id: req.id,
            site_id: req.site_id,
            member_level_id: req.member_level_id,
            mobile_phone_number: req.mobile_phone_number,
            bank_account,
            game_setting,
            settlement_setting,
          }),
        };

        let ss_update_member_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_UPDATE_MEMBER,
            ss_update_member_req.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,
            })
          })?;

        let ss_update_member_res =
          bpr::ss::member::UpdateMemberResponse::decode(ss_update_member_res_msg.data.as_slice())
            .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,
            })
          })?;

        if let Some(e) = ss_update_member_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(r) = ss_update_member_res.result {
          message
            .respond(
              bpr::c2se::member::UpdateMemberResponse {
                error: None,
                result: Some(bpr::c2se::member::update_member_response::Result {
                  member: r.member,
                }),
              }
              .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::c2se::member::UpdateMemberResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }

  async fn update_member_for_password(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_UPDATE_MEMBER_FOR_PASSWORD,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req =
          bpr::c2se::member::UpdateMemberForPasswordRequest::decode(message.data.as_slice())
            .map_err(|e| {
              bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
                message: format!("invalid request: {}", e),
              })
            })?;

        let ss_update_member_req = bpr::ss::member::UpdateMemberForPasswordRequest {
          client: Some(client),
          request: Some(
            bpr::ss::member::update_member_for_password_request::Request {
              id: req.id,
              password: req.password,
            },
          ),
        };

        let ss_update_member_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_UPDATE_MEMBER_FOR_PASSWORD,
            ss_update_member_req.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,
            })
          })?;

        let ss_update_member_res = bpr::ss::member::UpdateMemberForPasswordResponse::decode(
          ss_update_member_res_msg.data.as_slice(),
        )
        .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,
          })
        })?;

        if let Some(e) = ss_update_member_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(r) = ss_update_member_res.result {
          message
            .respond(
              bpr::c2se::member::UpdateMemberForPasswordResponse {
                error: None,
                result: Some(bpr::c2se::member::update_member_for_password_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::c2se::member::UpdateMemberForPasswordResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }

  async fn update_member_for_state(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_UPDATE_MEMBER_FOR_STATE,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req = bpr::c2se::member::UpdateMemberForStateRequest::decode(message.data.as_slice())
          .map_err(|e| {
          bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
            message: format!("invalid request: {}", e),
          })
        })?;

        let ss_update_member_req = bpr::ss::member::UpdateMemberForStateRequest {
          client: Some(client),
          request: Some(bpr::ss::member::update_member_for_state_request::Request {
            id: req.id,
            state: req.state,
          }),
        };

        let ss_update_member_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_UPDATE_MEMBER_FOR_STATE,
            ss_update_member_req.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,
            })
          })?;

        let ss_update_member_res = bpr::ss::member::UpdateMemberForStateResponse::decode(
          ss_update_member_res_msg.data.as_slice(),
        )
        .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,
          })
        })?;

        if let Some(e) = ss_update_member_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(r) = ss_update_member_res.result {
          message
            .respond(
              bpr::c2se::member::UpdateMemberForStateResponse {
                error: None,
                result: Some(
                  bpr::c2se::member::update_member_for_state_response::Result { member: r.member },
                ),
              }
              .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::c2se::member::UpdateMemberForStateResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }

  async fn delete_member(&self) -> Result<(), Box<dyn std::error::Error>> {
    let s = self
      .connection_broker
      .queue_subscribe(
        bpr::c2se::backend::member::SUBJECT_DELETE_MEMBER,
        self.queue_broker.as_str(),
      )
      .await?;

    while let Some(message) = s.next().await {
      if let Err(e) = async {
        let client = self.get_client_in_header(&message)?;

        let req =
          bpr::c2se::member::DeleteMemberRequest::decode(message.data.as_slice()).map_err(|e| {
            bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
              message: format!("invalid request: {}", e),
            })
          })?;

        let ss_delete_member_req = bpr::ss::member::DeleteMemberRequest {
          client: Some(client),
          request: Some(bpr::ss::member::delete_member_request::Request { id: req.id }),
        };

        let ss_delete_member_res_msg = self
          .connection_broker
          .request(
            bpr::ss::member::SUBJECT_DELETE_MEMBER,
            ss_delete_member_req.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,
            })
          })?;

        let ss_delete_member_res =
          bpr::ss::member::DeleteMemberResponse::decode(ss_delete_member_res_msg.data.as_slice())
            .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,
            })
          })?;

        if let Some(e) = ss_delete_member_res.error {
          return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server {
            code: bpr::protobuf::rpc::Error::SERVER_00,
            message: format!("server {}", e),
            data: None,
          }));
        }

        if let Some(_r) = ss_delete_member_res.result {
          message
            .respond(
              bpr::c2se::member::DeleteMemberResponse {
                error: None,
                result: Some(bpr::c2se::member::delete_member_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::c2se::member::DeleteMemberResponse {
              error: Some(bpr::protobuf::rpc::Error::from(e)),
              result: None,
            }
            .encode_to_vec(),
          )
          .await?;
      }
    }

    Ok(())
  }
}