bazzar/crates/account_manager/src/rpc.rs

use channels::accounts::rpc::Accounts;
use channels::accounts::{all, find_by_identity, me, register};
use channels::AsyncClient;
use config::SharedAppConfig;
use tarpc::context;

use crate::actions;
use crate::db::Database;

#[derive(Debug, Copy, Clone, serde::Serialize, thiserror::Error)]
#[serde(rename_all = "kebab-case", tag = "account")]
pub enum Error {
    #[error("Unable to send or receive msg from database")]
    DbCritical,
    #[error("Failed to load account data")]
    Account,
    #[error("Failed to load account addresses")]
    Addresses,
    #[error("Unable to save record")]
    Saving,
    #[error("Unable to hash password")]
    Hashing,
}

#[derive(Clone)]
struct AccountsServer {
    db: Database,
    config: SharedAppConfig,
    mqtt_client: AsyncClient,
}

#[tarpc::server]
impl Accounts for AccountsServer {
    async fn me(self, _: context::Context, input: me::Input) -> me::Output {
        let res = actions::me(input.account_id, self.db).await;
        tracing::info!("ME result: {:?}", res);
        res
    }

    async fn register_account(
        self,
        _: context::Context,
        input: register::Input,
    ) -> register::Output {
        use channels::accounts::Error;

        let res = actions::create_account(input, &self.db, self.config).await;
        tracing::info!("REGISTER result: {:?}", res);
        match res {
            Ok(account) => {
                self.mqtt_client.emit_account_created(&account).await;
                Ok(register::Details { account })
            }
            Err(_e) => Err(Error::Account),
        }
    }

    async fn all(self, _: context::Context, input: all::Input) -> all::Output {
        let res = actions::all(input, self.db).await;
        tracing::info!("ME result: {:?}", res);
        res
    }

    async fn find_by_identity(
        self,
        _: context::Context,
        input: find_by_identity::Input,
    ) -> find_by_identity::Output {
        let res = actions::find_by_identity(input, self.db).await;
        tracing::info!("ME result: {:?}", res);
        res
    }
}

pub async fn start(config: SharedAppConfig, db: Database, mqtt_client: AsyncClient) {
    let port = { config.lock().account_manager().rpc_port };

    channels::rpc::start("accounts", port, || {
        AccountsServer {
            db: db.clone(),
            config: config.clone(),
            mqtt_client: mqtt_client.clone(),
        }
        .serve()
    })
    .await;
}