rustfmt.toml

@@ -1,3 +1,3 @@
 reorder_impl_items = true
 use_small_heuristics = "Max"
-imports_granularity = "Module"
+imports_granularity = "Crate"

src/bin/brain.rs

@@ -6,8 +6,8 @@ use detee_shared::vm_proto::brain_vm_daemon_server::BrainVmDaemonServer;
 use dotenv::dotenv;
 use surreal_brain::constants::{BRAIN_GRPC_ADDR, CERT_KEY_PATH, CERT_PATH};
 use surreal_brain::db;
-use surreal_brain::grpc::general::GeneralCliServer;
-use surreal_brain::grpc::vm::{VmCliServer, VmDaemonServer};
+use surreal_brain::grpc::BrainVmCliForReal;
+use surreal_brain::grpc::{BrainGeneralCliForReal, BrainVmDaemonForReal};
 use tonic::transport::{Identity, Server, ServerTlsConfig};
 
 #[tokio::main]
@@ -26,9 +26,10 @@ async fn main() {
 
     let addr = BRAIN_GRPC_ADDR.parse().unwrap();
 
-    let snp_daemon_server = BrainVmDaemonServer::new(VmDaemonServer::new(db_arc.clone()));
-    let snp_cli_server = BrainVmCliServer::new(VmCliServer::new(db_arc.clone()));
-    let general_service_server = BrainGeneralCliServer::new(GeneralCliServer::new(db_arc.clone()));
+    let snp_daemon_server = BrainVmDaemonServer::new(BrainVmDaemonForReal::new(db_arc.clone()));
+    let snp_cli_server = BrainVmCliServer::new(BrainVmCliForReal::new(db_arc.clone()));
+    let general_service_server =
+        BrainGeneralCliServer::new(BrainGeneralCliForReal::new(db_arc.clone()));
 
     let cert = std::fs::read_to_string(CERT_PATH).unwrap();
     let key = std::fs::read_to_string(CERT_KEY_PATH).unwrap();

src/db.rs

@@ -1,16 +1,134 @@
-use std::str::FromStr;
-use std::time::Duration;
+use std::{str::FromStr, time::Duration};
+
+pub use crate::constants::{
+    ACCOUNT, ACTIVE_APP, ACTIVE_VM, DB_SCHEMA_FILE, DELETED_VM, ID_ALPHABET, NEW_VM_REQ,
+    UPDATE_VM_REQ, VM_CONTRACT, VM_NODE,
+};
 
-use super::Error;
-use crate::constants::{ACCOUNT, ACTIVE_VM, DELETED_VM, NEW_VM_REQ, VM_NODE};
-use crate::db::general::Report;
 use crate::old_brain;
 use serde::{Deserialize, Serialize};
-use surrealdb::engine::remote::ws::Client;
-use surrealdb::sql::Datetime;
-use surrealdb::{Notification, RecordId, Surreal};
+use surrealdb::{
+    engine::remote::ws::{Client, Ws},
+    opt::auth::Root,
+    sql::Datetime,
+    Notification, RecordId, Surreal,
+};
+use tokio::sync::mpsc::Sender;
 use tokio_stream::StreamExt as _;
 
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("Internal DB error: {0}")]
+    DataBase(#[from] surrealdb::Error),
+    #[error("Daemon channel got closed: {0}")]
+    DaemonConnection(#[from] tokio::sync::mpsc::error::SendError<DaemonNotification>),
+    #[error(transparent)]
+    StdIo(#[from] std::io::Error),
+    #[error(transparent)]
+    TimeOut(#[from] tokio::time::error::Elapsed),
+}
+
+pub async fn db_connection(
+    db_address: &str,
+    username: &str,
+    password: &str,
+    ns: &str,
+    db: &str,
+) -> Result<Surreal<Client>, Error> {
+    let db_connection: Surreal<Client> = Surreal::init();
+    db_connection.connect::<Ws>(db_address).await?;
+    // Sign in to the server
+    db_connection.signin(Root { username, password }).await?;
+    db_connection.use_ns(ns).use_db(db).await?;
+    Ok(db_connection)
+}
+
+pub async fn upsert_record<SomeRecord: Serialize + 'static>(
+    db: &Surreal<Client>,
+    table: &str,
+    id: &str,
+    my_record: SomeRecord,
+) -> Result<(), Error> {
+    #[derive(Deserialize)]
+    struct Wrapper {}
+    let _: Option<Wrapper> = db.create((table, id)).content(my_record).await?;
+    Ok(())
+}
+
+pub async fn migration0(
+    db: &Surreal<Client>,
+    old_data: &old_brain::BrainData,
+) -> Result<(), Error> {
+    let accounts: Vec<Account> = old_data.into();
+    let vm_nodes: Vec<VmNode> = old_data.into();
+    let app_nodes: Vec<AppNode> = old_data.into();
+    let vm_contracts: Vec<ActiveVm> = old_data.into();
+
+    let schema = std::fs::read_to_string(DB_SCHEMA_FILE)?;
+    db.query(schema).await?;
+
+    println!("Inserting accounts...");
+    let _: Vec<Account> = db.insert(()).content(accounts).await?;
+    println!("Inserting vm nodes...");
+    let _: Vec<VmNode> = db.insert(()).content(vm_nodes).await?;
+    println!("Inserting app nodes...");
+    let _: Vec<AppNode> = db.insert(()).content(app_nodes).await?;
+    println!("Inserting vm contracts...");
+    let _: Vec<ActiveVm> = db.insert("vm_contract").relation(vm_contracts).await?;
+
+    Ok(())
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Account {
+    pub id: RecordId,
+    pub balance: u64,
+    pub tmp_locked: u64,
+    pub escrow: u64,
+    pub email: String,
+}
+
+impl Account {
+    pub async fn get(db: &Surreal<Client>, address: &str) -> Result<Self, Error> {
+        let id = (ACCOUNT, address);
+        let account: Option<Self> = db.select(id).await?;
+        let account = match account {
+            Some(account) => account,
+            None => {
+                Self { id: id.into(), balance: 0, tmp_locked: 0, escrow: 0, email: String::new() }
+            }
+        };
+        Ok(account)
+    }
+
+    pub async fn airdrop(db: &Surreal<Client>, account: &str, tokens: u64) -> Result<(), Error> {
+        let tokens = tokens.saturating_mul(1_000_000_000);
+        let _ = db
+            .query(format!("upsert account:{account} SET balance = (balance || 0) + {tokens};"))
+            .await?;
+        Ok(())
+    }
+}
+
+impl Account {
+    pub async fn is_banned_by_node(
+        db: &Surreal<Client>,
+        user: &str,
+        node: &str,
+    ) -> Result<bool, Error> {
+        let ban: Option<Self> = db
+            .query(format!(
+                "(select operator->ban[0] as ban
+                    from vm_node:{node}
+                    where operator->ban->account contains account:{user}
+                ).ban;"
+            ))
+            .await?
+            .take(0)?;
+        Ok(ban.is_some())
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct VmNode {
     pub id: RecordId,
@@ -118,25 +236,25 @@ impl VmNodeWithReports {
     }
 }
 
-pub enum VmDaemonNotification {
+pub enum DaemonNotification {
     Create(NewVmReq),
     Update(UpdateVmReq),
     Delete(DeletedVm),
 }
 
-impl From<NewVmReq> for VmDaemonNotification {
+impl From<NewVmReq> for DaemonNotification {
     fn from(value: NewVmReq) -> Self {
         Self::Create(value)
     }
 }
 
-impl From<UpdateVmReq> for VmDaemonNotification {
+impl From<UpdateVmReq> for DaemonNotification {
     fn from(value: UpdateVmReq) -> Self {
         Self::Update(value)
     }
 }
 
-impl From<DeletedVm> for VmDaemonNotification {
+impl From<DeletedVm> for DaemonNotification {
     fn from(value: DeletedVm) -> Self {
         Self::Delete(value)
     }
@@ -348,6 +466,41 @@ pub struct UpdateVmReq {
     pub locked_nano: u64,
 }
 
+pub async fn listen_for_node<
+    T: Into<DaemonNotification> + std::marker::Unpin + for<'de> Deserialize<'de>,
+>(
+    db: &Surreal<Client>,
+    node: &str,
+    tx: Sender<DaemonNotification>,
+) -> Result<(), Error> {
+    let table_name = match std::any::type_name::<T>() {
+        "surreal_brain::db::NewVmReq" => NEW_VM_REQ.to_string(),
+        "surreal_brain::db::UpdateVmReq" => UPDATE_VM_REQ.to_string(),
+        "surreal_brain::db::DeletedVm" => DELETED_VM.to_string(),
+        wat => {
+            log::error!("listen_for_node: T has type {wat}");
+            String::from("wat")
+        }
+    };
+    let mut resp =
+        db.query(format!("live select * from {table_name} where out = vm_node:{node};")).await?;
+    let mut live_stream = resp.stream::<Notification<T>>(0)?;
+    while let Some(result) = live_stream.next().await {
+        match result {
+            Ok(notification) => {
+                if notification.action == surrealdb::Action::Create {
+                    tx.send(notification.data.into()).await?
+                }
+            }
+            Err(e) => {
+                log::warn!("listen_for_deletion DB stream failed for {node}: {e}");
+                return Err(Error::from(e));
+            }
+        }
+    }
+    Ok(())
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct DeletedVm {
     pub id: RecordId,
@@ -535,6 +688,225 @@ impl ActiveVmWithNode {
     }
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AppNode {
+    pub id: RecordId,
+    pub operator: RecordId,
+    pub country: String,
+    pub region: String,
+    pub city: String,
+    pub ip: String,
+    pub avail_mem_mb: u32,
+    pub avail_vcpus: u32,
+    pub avail_storage_gbs: u32,
+    pub avail_ports: u32,
+    pub max_ports_per_app: u32,
+    pub price: u64,
+    pub offline_minutes: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AppNodeWithReports {
+    pub id: RecordId,
+    pub operator: RecordId,
+    pub country: String,
+    pub region: String,
+    pub city: String,
+    pub ip: String,
+    pub avail_mem_mb: u32,
+    pub avail_vcpus: u32,
+    pub avail_storage_gbs: u32,
+    pub avail_ports: u32,
+    pub max_ports_per_app: u32,
+    pub price: u64,
+    pub offline_minutes: u64,
+    pub reports: Vec<Report>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ActiveApp {
+    id: RecordId,
+    #[serde(rename = "in")]
+    admin: RecordId,
+    #[serde(rename = "out")]
+    app_node: RecordId,
+    app_name: String,
+    mapped_ports: Vec<(u64, u64)>,
+    host_ipv4: String,
+    vcpus: u64,
+    memory_mb: u64,
+    disk_size_gb: u64,
+    created_at: Datetime,
+    price_per_unit: u64,
+    locked_nano: u64,
+    collected_at: Datetime,
+    mr_enclave: String,
+    package_url: String,
+    hratls_pubkey: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ActiveAppWithNode {
+    pub id: RecordId,
+    #[serde(rename = "in")]
+    pub admin: RecordId,
+    #[serde(rename = "out")]
+    pub app_node: AppNode,
+    pub app_name: String,
+    pub mapped_ports: Vec<(u64, u64)>,
+    pub host_ipv4: String,
+    pub vcpus: u64,
+    pub memory_mb: u64,
+    pub disk_size_gb: u64,
+    pub created_at: Datetime,
+    pub price_per_unit: u64,
+    pub locked_nano: u64,
+    pub collected_at: Datetime,
+    pub mr_enclave: String,
+    pub package_url: String,
+    pub hratls_pubkey: String,
+}
+
+impl ActiveAppWithNode {
+    pub async fn get_by_uuid(db: &Surreal<Client>, uuid: &str) -> Result<Option<Self>, Error> {
+        let contract: Option<Self> =
+            db.query(format!("select * from {ACTIVE_APP}:{uuid} fetch out;")).await?.take(0)?;
+        Ok(contract)
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Ban {
+    id: RecordId,
+    #[serde(rename = "in")]
+    from_account: RecordId,
+    #[serde(rename = "out")]
+    to_account: RecordId,
+    created_at: Datetime,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Kick {
+    id: RecordId,
+    #[serde(rename = "in")]
+    from_account: RecordId,
+    #[serde(rename = "out")]
+    to_account: RecordId,
+    created_at: Datetime,
+    reason: String,
+    contract: RecordId,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Report {
+    #[serde(rename = "in")]
+    from_account: RecordId,
+    #[serde(rename = "out")]
+    to_node: RecordId,
+    created_at: Datetime,
+    pub reason: String,
+    pub contract_id: String,
+}
+
+impl Report {
+    // TODO: test this functionality and remove this comment
+    pub async fn create(
+        db: &Surreal<Client>,
+        from_account: RecordId,
+        to_node: RecordId,
+        reason: String,
+        contract_id: String,
+    ) -> Result<(), Error> {
+        let _: Vec<Self> = db
+            .insert("report")
+            .relation(Report {
+                from_account,
+                to_node,
+                created_at: Datetime::default(),
+                reason,
+                contract_id,
+            })
+            .await?;
+        Ok(())
+    }
+}
+
+/// This is the operator obtained from the DB,
+/// however the relation is defined using OperatorRelation
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Operator {
+    pub account: RecordId,
+    pub app_nodes: u64,
+    pub vm_nodes: u64,
+    pub email: String,
+    pub escrow: u64,
+    pub reports: u64,
+}
+
+impl Operator {
+    pub async fn list(db: &Surreal<Client>) -> Result<Vec<Self>, Error> {
+        let mut result = db
+            .query(
+                "array::distinct(array::flatten( [
+                    (select operator from vm_node group by operator).operator,
+                    (select operator from app_node group by operator).operator
+                ]));"
+                    .to_string(),
+            )
+            .await?;
+        let operator_accounts: Vec<RecordId> = result.take(0)?;
+        let mut operators: Vec<Self> = Vec::new();
+        for account in operator_accounts.iter() {
+            if let Some(operator) = Self::inspect(db, &account.key().to_string()).await? {
+                operators.push(operator);
+            }
+        }
+        Ok(operators)
+    }
+
+    pub async fn inspect(db: &Surreal<Client>, account: &str) -> Result<Option<Self>, Error> {
+        let mut result = db
+            .query(format!(
+                "$vm_nodes = (select id from vm_node where operator = account:{account}).id;
+                $app_nodes = (select id from app_node where operator = account:{account}).id;
+                select *,
+                    id as account,
+                    email,
+                    escrow,
+                    $vm_nodes.len() as vm_nodes,
+                    $app_nodes.len() as app_nodes,
+                    (select id from report where $vm_nodes contains out).len() +
+                    (select id from report where $app_nodes contains out).len()
+                    as reports
+                from account where id = account:{account};"
+            ))
+            .await?;
+        let operator: Option<Self> = result.take(2)?;
+        Ok(operator)
+    }
+
+    pub async fn inspect_nodes(
+        db: &Surreal<Client>,
+        account: &str,
+    ) -> Result<(Option<Self>, Vec<VmNodeWithReports>, Vec<AppNodeWithReports>), Error> {
+        let operator = Self::inspect(db, account).await?;
+        let mut result = db
+            .query(format!(
+                "select *, operator, <-report.* as reports from vm_node
+                 where operator = account:{account};"
+            ))
+            .query(format!(
+                "select *, operator, <-report.* as reports from app_node
+                 where operator = account:{account};"
+            ))
+            .await?;
+        let vm_nodes: Vec<VmNodeWithReports> = result.take(0)?;
+        let app_nodes: Vec<AppNodeWithReports> = result.take(1)?;
+
+        Ok((operator, vm_nodes, app_nodes))
+    }
+}
+
 // TODO: delete all of these From implementation after migration 0 gets executed
 
 impl From<&old_brain::BrainData> for Vec<VmNode> {
@@ -593,3 +965,48 @@ impl From<&old_brain::BrainData> for Vec<ActiveVm> {
         contracts
     }
 }
+
+impl From<&old_brain::BrainData> for Vec<AppNode> {
+    fn from(old_data: &old_brain::BrainData) -> Self {
+        let mut nodes = Vec::new();
+        for old_node in old_data.app_nodes.iter() {
+            nodes.push(AppNode {
+                id: RecordId::from(("app_node", old_node.node_pubkey.clone())),
+                operator: RecordId::from((ACCOUNT, old_node.operator_wallet.clone())),
+                country: old_node.country.clone(),
+                region: old_node.region.clone(),
+                city: old_node.city.clone(),
+                ip: old_node.ip.clone(),
+                avail_mem_mb: old_node.avail_mem_mb,
+                avail_vcpus: old_node.avail_vcpus,
+                avail_storage_gbs: old_node.avail_storage_mb,
+                avail_ports: old_node.avail_no_of_port,
+                max_ports_per_app: old_node.max_ports_per_app,
+                price: old_node.price,
+                offline_minutes: old_node.offline_minutes,
+            });
+        }
+        nodes
+    }
+}
+
+impl From<&old_brain::BrainData> for Vec<Account> {
+    fn from(old_data: &old_brain::BrainData) -> Self {
+        let mut accounts = Vec::new();
+        for old_account in old_data.accounts.iter() {
+            let mut a = Account {
+                id: RecordId::from(("account", old_account.key())),
+                balance: old_account.value().balance,
+                tmp_locked: old_account.value().tmp_locked,
+                escrow: 0,
+                email: String::new(),
+            };
+            if let Some(operator) = old_data.operators.get(old_account.key()) {
+                a.escrow = operator.escrow;
+                a.email = operator.email.clone();
+            }
+            accounts.push(a);
+        }
+        accounts
+    }
+}

src/db/app.rs

@@ -1,120 +0,0 @@
-use crate::constants::{ACCOUNT, ACTIVE_APP};
-use crate::db::general::Report;
-
-use super::Error;
-use crate::old_brain;
-use serde::{Deserialize, Serialize};
-use surrealdb::engine::remote::ws::Client;
-use surrealdb::sql::Datetime;
-use surrealdb::{RecordId, Surreal};
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct AppNode {
-    pub id: RecordId,
-    pub operator: RecordId,
-    pub country: String,
-    pub region: String,
-    pub city: String,
-    pub ip: String,
-    pub avail_mem_mb: u32,
-    pub avail_vcpus: u32,
-    pub avail_storage_gbs: u32,
-    pub avail_ports: u32,
-    pub max_ports_per_app: u32,
-    pub price: u64,
-    pub offline_minutes: u64,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct AppNodeWithReports {
-    pub id: RecordId,
-    pub operator: RecordId,
-    pub country: String,
-    pub region: String,
-    pub city: String,
-    pub ip: String,
-    pub avail_mem_mb: u32,
-    pub avail_vcpus: u32,
-    pub avail_storage_gbs: u32,
-    pub avail_ports: u32,
-    pub max_ports_per_app: u32,
-    pub price: u64,
-    pub offline_minutes: u64,
-    pub reports: Vec<Report>,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct ActiveApp {
-    id: RecordId,
-    #[serde(rename = "in")]
-    admin: RecordId,
-    #[serde(rename = "out")]
-    app_node: RecordId,
-    app_name: String,
-    mapped_ports: Vec<(u64, u64)>,
-    host_ipv4: String,
-    vcpus: u64,
-    memory_mb: u64,
-    disk_size_gb: u64,
-    created_at: Datetime,
-    price_per_unit: u64,
-    locked_nano: u64,
-    collected_at: Datetime,
-    mr_enclave: String,
-    package_url: String,
-    hratls_pubkey: String,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct ActiveAppWithNode {
-    pub id: RecordId,
-    #[serde(rename = "in")]
-    pub admin: RecordId,
-    #[serde(rename = "out")]
-    pub app_node: AppNode,
-    pub app_name: String,
-    pub mapped_ports: Vec<(u64, u64)>,
-    pub host_ipv4: String,
-    pub vcpus: u64,
-    pub memory_mb: u64,
-    pub disk_size_gb: u64,
-    pub created_at: Datetime,
-    pub price_per_unit: u64,
-    pub locked_nano: u64,
-    pub collected_at: Datetime,
-    pub mr_enclave: String,
-    pub package_url: String,
-    pub hratls_pubkey: String,
-}
-
-impl ActiveAppWithNode {
-    pub async fn get_by_uuid(db: &Surreal<Client>, uuid: &str) -> Result<Option<Self>, Error> {
-        let contract: Option<Self> =
-            db.query(format!("select * from {ACTIVE_APP}:{uuid} fetch out;")).await?.take(0)?;
-        Ok(contract)
-    }
-}
-
-impl From<&old_brain::BrainData> for Vec<AppNode> {
-    fn from(old_data: &old_brain::BrainData) -> Self {
-        let mut nodes = Vec::new();
-        for old_node in old_data.app_nodes.iter() {
-            nodes.push(AppNode {
-                id: RecordId::from(("app_node", old_node.node_pubkey.clone())),
-                operator: RecordId::from((ACCOUNT, old_node.operator_wallet.clone())),
-                country: old_node.country.clone(),
-                region: old_node.region.clone(),
-                city: old_node.city.clone(),
-                ip: old_node.ip.clone(),
-                avail_mem_mb: old_node.avail_mem_mb,
-                avail_vcpus: old_node.avail_vcpus,
-                avail_storage_gbs: old_node.avail_storage_mb,
-                avail_ports: old_node.avail_no_of_port,
-                max_ports_per_app: old_node.max_ports_per_app,
-                price: old_node.price,
-                offline_minutes: old_node.offline_minutes,
-            });
-        }
-        nodes
-    }
-}

src/db/general.rs

@@ -1,212 +0,0 @@
-use crate::constants::ACCOUNT;
-use crate::db::prelude::*;
-
-use super::Error;
-use crate::old_brain;
-use serde::{Deserialize, Serialize};
-use surrealdb::engine::remote::ws::Client;
-use surrealdb::sql::Datetime;
-use surrealdb::{RecordId, Surreal};
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Account {
-    pub id: RecordId,
-    pub balance: u64,
-    pub tmp_locked: u64,
-    pub escrow: u64,
-    pub email: String,
-}
-
-impl Account {
-    pub async fn get(db: &Surreal<Client>, address: &str) -> Result<Self, Error> {
-        let id = (ACCOUNT, address);
-        let account: Option<Self> = db.select(id).await?;
-        let account = match account {
-            Some(account) => account,
-            None => {
-                Self { id: id.into(), balance: 0, tmp_locked: 0, escrow: 0, email: String::new() }
-            }
-        };
-        Ok(account)
-    }
-
-    pub async fn airdrop(db: &Surreal<Client>, account: &str, tokens: u64) -> Result<(), Error> {
-        let tokens = tokens.saturating_mul(1_000_000_000);
-        let _ = db
-            .query(format!("upsert account:{account} SET balance = (balance || 0) + {tokens};"))
-            .await?;
-        Ok(())
-    }
-}
-
-impl Account {
-    pub async fn is_banned_by_node(
-        db: &Surreal<Client>,
-        user: &str,
-        node: &str,
-    ) -> Result<bool, Error> {
-        let ban: Option<Self> = db
-            .query(format!(
-                "(select operator->ban[0] as ban
-                    from vm_node:{node}
-                    where operator->ban->account contains account:{user}
-                ).ban;"
-            ))
-            .await?
-            .take(0)?;
-        Ok(ban.is_some())
-    }
-}
-
-impl From<&old_brain::BrainData> for Vec<Account> {
-    fn from(old_data: &old_brain::BrainData) -> Self {
-        let mut accounts = Vec::new();
-        for old_account in old_data.accounts.iter() {
-            let mut a = Account {
-                id: RecordId::from(("account", old_account.key())),
-                balance: old_account.value().balance,
-                tmp_locked: old_account.value().tmp_locked,
-                escrow: 0,
-                email: String::new(),
-            };
-            if let Some(operator) = old_data.operators.get(old_account.key()) {
-                a.escrow = operator.escrow;
-                a.email = operator.email.clone();
-            }
-            accounts.push(a);
-        }
-        accounts
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Ban {
-    id: RecordId,
-    #[serde(rename = "in")]
-    from_account: RecordId,
-    #[serde(rename = "out")]
-    to_account: RecordId,
-    created_at: Datetime,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Kick {
-    id: RecordId,
-    #[serde(rename = "in")]
-    from_account: RecordId,
-    #[serde(rename = "out")]
-    to_account: RecordId,
-    created_at: Datetime,
-    reason: String,
-    contract: RecordId,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Report {
-    #[serde(rename = "in")]
-    from_account: RecordId,
-    #[serde(rename = "out")]
-    to_node: RecordId,
-    created_at: Datetime,
-    pub reason: String,
-    pub contract_id: String,
-}
-
-impl Report {
-    // TODO: test this functionality and remove this comment
-    pub async fn create(
-        db: &Surreal<Client>,
-        from_account: RecordId,
-        to_node: RecordId,
-        reason: String,
-        contract_id: String,
-    ) -> Result<(), Error> {
-        let _: Vec<Self> = db
-            .insert("report")
-            .relation(Report {
-                from_account,
-                to_node,
-                created_at: Datetime::default(),
-                reason,
-                contract_id,
-            })
-            .await?;
-        Ok(())
-    }
-}
-
-/// This is the operator obtained from the DB,
-/// however the relation is defined using OperatorRelation
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Operator {
-    pub account: RecordId,
-    pub app_nodes: u64,
-    pub vm_nodes: u64,
-    pub email: String,
-    pub escrow: u64,
-    pub reports: u64,
-}
-
-impl Operator {
-    pub async fn list(db: &Surreal<Client>) -> Result<Vec<Self>, Error> {
-        let mut result = db
-            .query(
-                "array::distinct(array::flatten( [
-                    (select operator from vm_node group by operator).operator,
-                    (select operator from app_node group by operator).operator
-                ]));"
-                    .to_string(),
-            )
-            .await?;
-        let operator_accounts: Vec<RecordId> = result.take(0)?;
-        let mut operators: Vec<Self> = Vec::new();
-        for account in operator_accounts.iter() {
-            if let Some(operator) = Self::inspect(db, &account.key().to_string()).await? {
-                operators.push(operator);
-            }
-        }
-        Ok(operators)
-    }
-
-    pub async fn inspect(db: &Surreal<Client>, account: &str) -> Result<Option<Self>, Error> {
-        let mut result = db
-            .query(format!(
-                "$vm_nodes = (select id from vm_node where operator = account:{account}).id;
-                $app_nodes = (select id from app_node where operator = account:{account}).id;
-                select *,
-                    id as account,
-                    email,
-                    escrow,
-                    $vm_nodes.len() as vm_nodes,
-                    $app_nodes.len() as app_nodes,
-                    (select id from report where $vm_nodes contains out).len() +
-                    (select id from report where $app_nodes contains out).len()
-                    as reports
-                from account where id = account:{account};"
-            ))
-            .await?;
-        let operator: Option<Self> = result.take(2)?;
-        Ok(operator)
-    }
-
-    pub async fn inspect_nodes(
-        db: &Surreal<Client>,
-        account: &str,
-    ) -> Result<(Option<Self>, Vec<VmNodeWithReports>, Vec<AppNodeWithReports>), Error> {
-        let operator = Self::inspect(db, account).await?;
-        let mut result = db
-            .query(format!(
-                "select *, operator, <-report.* as reports from vm_node
-                 where operator = account:{account};"
-            ))
-            .query(format!(
-                "select *, operator, <-report.* as reports from app_node
-                 where operator = account:{account};"
-            ))
-            .await?;
-        let vm_nodes: Vec<VmNodeWithReports> = result.take(0)?;
-        let app_nodes: Vec<AppNodeWithReports> = result.take(1)?;
-
-        Ok((operator, vm_nodes, app_nodes))
-    }
-}

src/db/mod.rs

@@ -1,118 +0,0 @@
-pub mod app;
-pub mod general;
-pub mod vm;
-
-use crate::constants::{DELETED_VM, NEW_VM_REQ, UPDATE_VM_REQ};
-use crate::old_brain;
-use prelude::*;
-use serde::{Deserialize, Serialize};
-use surrealdb::engine::remote::ws::{Client, Ws};
-use surrealdb::opt::auth::Root;
-use surrealdb::{Notification, Surreal};
-use tokio::sync::mpsc::Sender;
-use tokio_stream::StreamExt;
-
-#[derive(thiserror::Error, Debug)]
-pub enum Error {
-    #[error("Internal DB error: {0}")]
-    DataBase(#[from] surrealdb::Error),
-    #[error("Daemon channel got closed: {0}")]
-    VmDaemonConnection(#[from] tokio::sync::mpsc::error::SendError<VmDaemonNotification>),
-    #[error(transparent)]
-    StdIo(#[from] std::io::Error),
-    #[error(transparent)]
-    TimeOut(#[from] tokio::time::error::Elapsed),
-}
-
-pub mod prelude {
-    pub use super::app::*;
-    pub use super::general::*;
-    pub use super::vm::*;
-    pub use super::*;
-}
-
-pub async fn db_connection(
-    db_address: &str,
-    username: &str,
-    password: &str,
-    ns: &str,
-    db: &str,
-) -> Result<Surreal<Client>, Error> {
-    let db_connection: Surreal<Client> = Surreal::init();
-    db_connection.connect::<Ws>(db_address).await?;
-    // Sign in to the server
-    db_connection.signin(Root { username, password }).await?;
-    db_connection.use_ns(ns).use_db(db).await?;
-    Ok(db_connection)
-}
-
-pub async fn migration0(
-    db: &Surreal<Client>,
-    old_data: &old_brain::BrainData,
-) -> Result<(), Error> {
-    let accounts: Vec<Account> = old_data.into();
-    let vm_nodes: Vec<VmNode> = old_data.into();
-    let app_nodes: Vec<AppNode> = old_data.into();
-    let vm_contracts: Vec<ActiveVm> = old_data.into();
-
-    let schema = std::fs::read_to_string(crate::constants::DB_SCHEMA_FILE)?;
-    db.query(schema).await?;
-
-    println!("Inserting accounts...");
-    let _: Vec<Account> = db.insert(()).content(accounts).await?;
-    println!("Inserting vm nodes...");
-    let _: Vec<VmNode> = db.insert(()).content(vm_nodes).await?;
-    println!("Inserting app nodes...");
-    let _: Vec<AppNode> = db.insert(()).content(app_nodes).await?;
-    println!("Inserting vm contracts...");
-    let _: Vec<ActiveVm> = db.insert("vm_contract").relation(vm_contracts).await?;
-
-    Ok(())
-}
-
-pub async fn upsert_record<SomeRecord: Serialize + 'static>(
-    db: &Surreal<Client>,
-    table: &str,
-    id: &str,
-    my_record: SomeRecord,
-) -> Result<(), Error> {
-    #[derive(Deserialize)]
-    struct Wrapper {}
-    let _: Option<Wrapper> = db.create((table, id)).content(my_record).await?;
-    Ok(())
-}
-
-pub async fn listen_for_vm_node<
-    T: Into<vm::VmDaemonNotification> + std::marker::Unpin + for<'de> Deserialize<'de>,
->(
-    db: &Surreal<Client>,
-    node: &str,
-    tx: Sender<vm::VmDaemonNotification>,
-) -> Result<(), Error> {
-    let table_name = match std::any::type_name::<T>() {
-        "surreal_brain::db::NewVmReq" => NEW_VM_REQ.to_string(),
-        "surreal_brain::db::UpdateVmReq" => UPDATE_VM_REQ.to_string(),
-        "surreal_brain::db::DeletedVm" => DELETED_VM.to_string(),
-        wat => {
-            log::error!("listen_for_node: T has type {wat}");
-            String::from("wat")
-        }
-    };
-    let mut resp =
-        db.query(format!("live select * from {table_name} where out = vm_node:{node};")).await?;
-    let mut live_stream = resp.stream::<Notification<T>>(0)?;
-    while let Some(result) = live_stream.next().await {
-        match result {
-            Ok(notification) => {
-                if notification.action == surrealdb::Action::Create {
-                    tx.send(notification.data.into()).await?
-                }
-            }
-            Err(e) => {
-                log::warn!("listen_for_deletion DB stream failed for {node}: {e}");
-                return Err(Error::from(e));
-            }
-        }
-    }
-    Ok(())
-}

src/grpc.rs

@@ -0,0 +1,907 @@
+#![allow(dead_code)]
+use crate::constants::{ACCOUNT, ADMIN_ACCOUNTS, VM_NODE};
+use crate::db;
+use detee_shared::app_proto::{AppContract, AppNodeListResp};
+use detee_shared::{
+    common_proto::{Empty, Pubkey},
+    general_proto::{
+        brain_general_cli_server::BrainGeneralCli, Account, AccountBalance, AirdropReq, BanUserReq,
+        InspectOperatorResp, KickReq, KickResp, ListOperatorsResp, RegOperatorReq, ReportNodeReq,
+        SlashReq,
+    },
+    vm_proto::{
+        brain_vm_cli_server::BrainVmCli, brain_vm_daemon_server::BrainVmDaemon, ListVmContractsReq,
+        *,
+    },
+};
+use nanoid::nanoid;
+use surrealdb::{engine::remote::ws::Client, Surreal};
+
+use log::info;
+use std::pin::Pin;
+use std::sync::Arc;
+use surrealdb::RecordId;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
+use tokio_stream::{Stream, StreamExt};
+use tonic::{Request, Response, Status, Streaming};
+
+pub struct BrainGeneralCliForReal {
+    pub db: Arc<Surreal<Client>>,
+}
+
+impl BrainGeneralCliForReal {
+    pub fn new(db: Arc<Surreal<Client>>) -> Self {
+        Self { db }
+    }
+}
+
+impl From<db::Account> for AccountBalance {
+    fn from(account: db::Account) -> Self {
+        AccountBalance { balance: account.balance, tmp_locked: account.tmp_locked }
+    }
+}
+
+impl From<NewVmReq> for db::NewVmReq {
+    fn from(new_vm_req: NewVmReq) -> Self {
+        Self {
+            id: RecordId::from((db::NEW_VM_REQ, nanoid!(40, &db::ID_ALPHABET))),
+            admin: RecordId::from((db::ACCOUNT, new_vm_req.admin_pubkey)),
+            vm_node: RecordId::from((db::VM_NODE, new_vm_req.node_pubkey)),
+            hostname: new_vm_req.hostname,
+            extra_ports: new_vm_req.extra_ports,
+            public_ipv4: new_vm_req.public_ipv4,
+            public_ipv6: new_vm_req.public_ipv6,
+            disk_size_gb: new_vm_req.disk_size_gb,
+            vcpus: new_vm_req.vcpus,
+            memory_mb: new_vm_req.memory_mb,
+            kernel_url: new_vm_req.kernel_url,
+            kernel_sha: new_vm_req.kernel_sha,
+            dtrfs_url: new_vm_req.dtrfs_url,
+            dtrfs_sha: new_vm_req.dtrfs_sha,
+            price_per_unit: new_vm_req.price_per_unit,
+            locked_nano: new_vm_req.locked_nano,
+            created_at: surrealdb::sql::Datetime::default(),
+            error: String::new(),
+        }
+    }
+}
+
+impl From<db::NewVmReq> for NewVmReq {
+    fn from(new_vm_req: db::NewVmReq) -> Self {
+        Self {
+            uuid: new_vm_req.id.key().to_string(),
+            hostname: new_vm_req.hostname,
+            admin_pubkey: new_vm_req.admin.key().to_string(),
+            node_pubkey: new_vm_req.vm_node.key().to_string(),
+            extra_ports: new_vm_req.extra_ports,
+            public_ipv4: new_vm_req.public_ipv4,
+            public_ipv6: new_vm_req.public_ipv6,
+            disk_size_gb: new_vm_req.disk_size_gb,
+            vcpus: new_vm_req.vcpus,
+            memory_mb: new_vm_req.memory_mb,
+            kernel_url: new_vm_req.kernel_url,
+            kernel_sha: new_vm_req.kernel_sha,
+            dtrfs_url: new_vm_req.dtrfs_url,
+            dtrfs_sha: new_vm_req.dtrfs_sha,
+            price_per_unit: new_vm_req.price_per_unit,
+            locked_nano: new_vm_req.locked_nano,
+        }
+    }
+}
+
+impl From<db::NewVmResp> for NewVmResp {
+    fn from(resp: db::NewVmResp) -> Self {
+        match resp {
+            // TODO: This will require a small architecture change to pass MeasurementArgs from
+            // Daemon to CLI
+            db::NewVmResp::Args(uuid, args) => {
+                NewVmResp { uuid, error: String::new(), args: Some(args) }
+            }
+            db::NewVmResp::Error(uuid, error) => NewVmResp { uuid, error, args: None },
+        }
+    }
+}
+
+impl From<db::UpdateVmReq> for UpdateVmReq {
+    fn from(update_vm_req: db::UpdateVmReq) -> Self {
+        Self {
+            uuid: update_vm_req.id.key().to_string(),
+            // daemon does not care about VM hostname
+            hostname: String::new(),
+            admin_pubkey: update_vm_req.admin.key().to_string(),
+            disk_size_gb: update_vm_req.disk_size_gb,
+            vcpus: update_vm_req.vcpus,
+            memory_mb: update_vm_req.memory_mb,
+            kernel_url: update_vm_req.kernel_url,
+            kernel_sha: update_vm_req.kernel_sha,
+            dtrfs_url: update_vm_req.dtrfs_url,
+            dtrfs_sha: update_vm_req.dtrfs_sha,
+        }
+    }
+}
+
+impl From<db::DeletedVm> for DeleteVmReq {
+    fn from(delete_vm_req: db::DeletedVm) -> Self {
+        Self {
+            uuid: delete_vm_req.id.key().to_string(),
+            admin_pubkey: delete_vm_req.admin.key().to_string(),
+        }
+    }
+}
+
+impl From<db::DaemonNotification> for BrainVmMessage {
+    fn from(notification: db::DaemonNotification) -> Self {
+        match notification {
+            db::DaemonNotification::Create(new_vm_req) => {
+                BrainVmMessage { msg: Some(brain_vm_message::Msg::NewVmReq(new_vm_req.into())) }
+            }
+            db::DaemonNotification::Update(update_vm_req) => BrainVmMessage {
+                msg: Some(brain_vm_message::Msg::UpdateVmReq(update_vm_req.into())),
+            },
+            db::DaemonNotification::Delete(deleted_vm) => {
+                BrainVmMessage { msg: Some(brain_vm_message::Msg::DeleteVm(deleted_vm.into())) }
+            }
+        }
+    }
+}
+
+impl From<db::ActiveVmWithNode> for VmContract {
+    fn from(db_c: db::ActiveVmWithNode) -> Self {
+        let mut exposed_ports = Vec::new();
+        for port in db_c.mapped_ports.iter() {
+            exposed_ports.push(port.0);
+        }
+        VmContract {
+            uuid: db_c.id.key().to_string(),
+            hostname: db_c.hostname.clone(),
+            admin_pubkey: db_c.admin.key().to_string(),
+            node_pubkey: db_c.vm_node.id.key().to_string(),
+            node_ip: db_c.vm_node.ip.clone(),
+            location: format!(
+                "{}, {}, {}",
+                db_c.vm_node.city, db_c.vm_node.region, db_c.vm_node.country
+            ),
+            memory_mb: db_c.memory_mb,
+            vcpus: db_c.vcpus,
+            disk_size_gb: db_c.disk_size_gb,
+            mapped_ports: db_c
+                .mapped_ports
+                .iter()
+                .map(|(h, g)| MappedPort { host_port: *h, guest_port: *g })
+                .collect(),
+            vm_public_ipv6: db_c.public_ipv6.clone(),
+            vm_public_ipv4: db_c.public_ipv4.clone(),
+            locked_nano: db_c.locked_nano,
+            dtrfs_sha: db_c.dtrfs_sha.clone(),
+            kernel_sha: db_c.kernel_sha.clone(),
+            nano_per_minute: db_c.price_per_minute(),
+            created_at: db_c.created_at.to_rfc3339(),
+            // TODO: remove updated_at from the proto
+            // This will get moved to VM history (users will be able to
+            // query old contracts, which also shows updates of existing contracts).
+            updated_at: db_c.created_at.to_rfc3339(),
+            collected_at: db_c.collected_at.to_rfc3339(),
+        }
+    }
+}
+
+impl From<db::Error> for tonic::Status {
+    fn from(e: db::Error) -> Self {
+        Self::internal(format!("Internal error: {e}"))
+    }
+}
+
+impl From<db::Operator> for ListOperatorsResp {
+    fn from(db_o: db::Operator) -> Self {
+        ListOperatorsResp {
+            pubkey: db_o.account.key().to_string(),
+            escrow: db_o.escrow,
+            email: db_o.email,
+            app_nodes: db_o.app_nodes,
+            vm_nodes: db_o.vm_nodes,
+            reports: db_o.reports,
+        }
+    }
+}
+
+impl From<db::VmNodeWithReports> for VmNodeListResp {
+    fn from(vm_node: db::VmNodeWithReports) -> Self {
+        Self {
+            operator: vm_node.operator.key().to_string(),
+            node_pubkey: vm_node.id.key().to_string(),
+            country: vm_node.country,
+            region: vm_node.region,
+            city: vm_node.city,
+            ip: vm_node.ip,
+            reports: vm_node.reports.iter().map(|n| n.reason.clone()).collect(),
+            price: vm_node.price,
+        }
+    }
+}
+
+impl From<db::AppNodeWithReports> for AppNodeListResp {
+    fn from(app_node: db::AppNodeWithReports) -> Self {
+        Self {
+            operator: app_node.operator.key().to_string(),
+            node_pubkey: app_node.id.key().to_string(),
+            country: app_node.country,
+            region: app_node.region,
+            city: app_node.city,
+            ip: app_node.ip,
+            reports: app_node.reports.iter().map(|n| n.reason.clone()).collect(),
+            price: app_node.price,
+        }
+    }
+}
+
+impl From<VmNodeResources> for db::VmNodeResources {
+    fn from(res: VmNodeResources) -> Self {
+        Self {
+            avail_mem_mb: res.avail_memory_mb,
+            avail_vcpus: res.avail_vcpus,
+            avail_storage_gbs: res.avail_storage_gb,
+            avail_ipv4: res.avail_ipv4,
+            avail_ipv6: res.avail_ipv6,
+            avail_ports: res.avail_ports,
+            max_ports_per_vm: res.max_ports_per_vm,
+        }
+    }
+}
+
+pub struct BrainVmDaemonForReal {
+    pub db: Arc<Surreal<Client>>,
+}
+
+impl BrainVmDaemonForReal {
+    pub fn new(db: Arc<Surreal<Client>>) -> Self {
+        Self { db }
+    }
+}
+
+#[tonic::async_trait]
+impl BrainVmDaemon for BrainVmDaemonForReal {
+    type RegisterVmNodeStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
+    async fn register_vm_node(
+        &self,
+        req: Request<RegisterVmNodeReq>,
+    ) -> Result<Response<Self::RegisterVmNodeStream>, Status> {
+        let req = check_sig_from_req(req)?;
+        info!("Starting registration process for {:?}", req);
+        db::VmNode {
+            id: surrealdb::RecordId::from((VM_NODE, req.node_pubkey.clone())),
+            operator: surrealdb::RecordId::from((ACCOUNT, req.operator_wallet)),
+            country: req.country,
+            region: req.region,
+            city: req.city,
+            ip: req.main_ip,
+            price: req.price,
+            avail_mem_mb: 0,
+            avail_vcpus: 0,
+            avail_storage_gbs: 0,
+            avail_ipv4: 0,
+            avail_ipv6: 0,
+            avail_ports: 0,
+            max_ports_per_vm: 0,
+            offline_minutes: 0,
+        }
+        .register(&self.db)
+        .await?;
+
+        info!("Sending existing contracts to {}", req.node_pubkey);
+        let contracts = db::ActiveVmWithNode::list_by_node(&self.db, &req.node_pubkey).await?;
+        let (tx, rx) = mpsc::channel(6);
+        tokio::spawn(async move {
+            for contract in contracts {
+                let _ = tx.send(Ok(contract.into())).await;
+            }
+        });
+        let output_stream = ReceiverStream::new(rx);
+        Ok(Response::new(Box::pin(output_stream) as Self::RegisterVmNodeStream))
+    }
+
+    type BrainMessagesStream = Pin<Box<dyn Stream<Item = Result<BrainVmMessage, Status>> + Send>>;
+    async fn brain_messages(
+        &self,
+        req: Request<DaemonStreamAuth>,
+    ) -> Result<Response<Self::BrainMessagesStream>, Status> {
+        let auth = req.into_inner();
+        let pubkey = auth.pubkey.clone();
+        check_sig_from_parts(
+            &pubkey,
+            &auth.timestamp,
+            &format!("{:?}", auth.contracts),
+            &auth.signature,
+        )?;
+        info!("Daemon {} connected to receive brain messages", pubkey);
+
+        let (tx, rx) = mpsc::channel(6);
+        {
+            let db = self.db.clone();
+            let pubkey = pubkey.clone();
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                match db::listen_for_node::<db::DeletedVm>(&db, &pubkey, tx).await {
+                    Ok(()) => log::info!("db::VmContract::listen_for_node ended for {pubkey}"),
+                    Err(e) => {
+                        log::warn!("db::VmContract::listen_for_node errored for {pubkey}: {e}")
+                    }
+                };
+            });
+        }
+        {
+            let db = self.db.clone();
+            let pubkey = pubkey.clone();
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                let _ = db::listen_for_node::<db::NewVmReq>(&db, &pubkey, tx.clone()).await;
+            });
+        }
+        {
+            let db = self.db.clone();
+            let pubkey = pubkey.clone();
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                let _ = db::listen_for_node::<db::UpdateVmReq>(&db, &pubkey, tx.clone()).await;
+            });
+        }
+
+        let output_stream = ReceiverStream::new(rx).map(|msg| Ok(msg.into()));
+        Ok(Response::new(Box::pin(output_stream) as Self::BrainMessagesStream))
+    }
+
+    async fn daemon_messages(
+        &self,
+        req: Request<Streaming<VmDaemonMessage>>,
+    ) -> Result<Response<Empty>, Status> {
+        let mut req_stream = req.into_inner();
+        let pubkey: String;
+        if let Some(Ok(msg)) = req_stream.next().await {
+            log::debug!("demon_messages received the following auth message: {:?}", msg.msg);
+            if let Some(vm_daemon_message::Msg::Auth(auth)) = msg.msg {
+                pubkey = auth.pubkey.clone();
+                check_sig_from_parts(
+                    &pubkey,
+                    &auth.timestamp,
+                    &format!("{:?}", auth.contracts),
+                    &auth.signature,
+                )?;
+            } else {
+                return Err(Status::unauthenticated(
+                    "Could not authenticate the daemon: could not extract auth signature",
+                ));
+            }
+        } else {
+            return Err(Status::unauthenticated("Could not authenticate the daemon"));
+        }
+
+        while let Some(daemon_message) = req_stream.next().await {
+            match daemon_message {
+                Ok(msg) => match msg.msg {
+                    Some(vm_daemon_message::Msg::NewVmResp(new_vm_resp)) => {
+                        // TODO: move new_vm_req to active_vm
+                        // also handle failure properly
+                        if !new_vm_resp.error.is_empty() {
+                            db::NewVmReq::submit_error(
+                                &self.db,
+                                &new_vm_resp.uuid,
+                                new_vm_resp.error,
+                            )
+                            .await?;
+                        } else {
+                            db::upsert_record(
+                                &self.db,
+                                "measurement_args",
+                                &new_vm_resp.uuid,
+                                new_vm_resp.args.clone(),
+                            )
+                            .await?;
+                            if let Some(args) = new_vm_resp.args {
+                                db::ActiveVm::activate(&self.db, &new_vm_resp.uuid, args).await?;
+                            }
+                        }
+                    }
+                    Some(vm_daemon_message::Msg::UpdateVmResp(_update_vm_resp)) => {
+                        todo!();
+                        // self.data.submit_updatevm_resp(update_vm_resp).await;
+                    }
+                    Some(vm_daemon_message::Msg::VmNodeResources(node_resources)) => {
+                        let node_resources: db::VmNodeResources = node_resources.into();
+                        node_resources.merge(&self.db, &pubkey).await?;
+                    }
+                    _ => {}
+                },
+                Err(e) => {
+                    log::warn!("Daemon disconnected: {e:?}");
+                }
+            }
+        }
+        Ok(Response::new(Empty {}))
+    }
+}
+
+#[tonic::async_trait]
+impl BrainGeneralCli for BrainGeneralCliForReal {
+    type ListAccountsStream = Pin<Box<dyn Stream<Item = Result<Account, Status>> + Send>>;
+    type ListAllAppContractsStream =
+        Pin<Box<dyn Stream<Item = Result<AppContract, Status>> + Send>>;
+    type ListAllVmContractsStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
+    type ListOperatorsStream =
+        Pin<Box<dyn Stream<Item = Result<ListOperatorsResp, Status>> + Send>>;
+
+    async fn get_balance(&self, req: Request<Pubkey>) -> Result<Response<AccountBalance>, Status> {
+        let req = check_sig_from_req(req)?;
+        Ok(Response::new(db::Account::get(&self.db, &req.pubkey).await?.into()))
+    }
+
+    async fn report_node(&self, req: Request<ReportNodeReq>) -> Result<Response<Empty>, Status> {
+        let req = check_sig_from_req(req)?;
+        let (account, node, contract_id) =
+            match db::ActiveVmWithNode::get_by_uuid(&self.db, &req.contract).await? {
+                Some(vm_contract)
+                    if vm_contract.admin.key().to_string() == req.admin_pubkey
+                        && vm_contract.vm_node.id.key().to_string() == req.node_pubkey =>
+                {
+                    (vm_contract.admin, vm_contract.vm_node.id, vm_contract.id.to_string())
+                }
+                _ => match db::ActiveAppWithNode::get_by_uuid(&self.db, &req.contract).await? {
+                    Some(app_contract)
+                        if app_contract.admin.key().to_string() == req.admin_pubkey
+                            && app_contract.app_node.id.key().to_string() == req.node_pubkey =>
+                    {
+                        (app_contract.admin, app_contract.app_node.id, app_contract.id.to_string())
+                    }
+                    _ => {
+                        return Err(Status::unauthenticated("No contract found by this ID."));
+                    }
+                },
+            };
+        db::Report::create(&self.db, account, node, req.reason, contract_id).await?;
+        Ok(Response::new(Empty {}))
+    }
+
+    async fn list_operators(
+        &self,
+        req: Request<Empty>,
+    ) -> Result<Response<Self::ListOperatorsStream>, Status> {
+        let _ = check_sig_from_req(req)?;
+        let operators = db::Operator::list(&self.db).await?;
+        let (tx, rx) = mpsc::channel(6);
+        tokio::spawn(async move {
+            for op in operators {
+                let _ = tx.send(Ok(op.into())).await;
+            }
+        });
+        let output_stream = ReceiverStream::new(rx);
+        Ok(Response::new(Box::pin(output_stream) as Self::ListOperatorsStream))
+    }
+
+    async fn inspect_operator(
+        &self,
+        req: Request<Pubkey>,
+    ) -> Result<Response<InspectOperatorResp>, Status> {
+        match db::Operator::inspect_nodes(&self.db, &req.into_inner().pubkey).await? {
+            (Some(op), vm_nodes, app_nodes) => Ok(Response::new(InspectOperatorResp {
+                operator: Some(op.into()),
+                vm_nodes: vm_nodes.into_iter().map(|n| n.into()).collect(),
+                app_nodes: app_nodes.into_iter().map(|n| n.into()).collect(),
+            })),
+            (None, _, _) => Err(Status::not_found("The wallet you specified is not an operator")),
+        }
+    }
+
+    async fn register_operator(
+        &self,
+        _req: Request<RegOperatorReq>,
+    ) -> Result<Response<Empty>, Status> {
+        todo!();
+        // let req = check_sig_from_req(req)?;
+        // info!("Regitering new operator: {req:?}");
+        // match self.data.register_operator(req) {
+        //     Ok(()) => Ok(Response::new(Empty {})),
+        //     Err(e) => Err(Status::failed_precondition(e.to_string())),
+        // }
+    }
+
+    async fn kick_contract(&self, _req: Request<KickReq>) -> Result<Response<KickResp>, Status> {
+        todo!();
+        // let req = check_sig_from_req(req)?;
+        // match self.data.kick_contract(&req.operator_wallet, &req.contract_uuid, &req.reason).await {
+        //     Ok(nano_lp) => Ok(Response::new(KickResp { nano_lp })),
+        //     Err(e) => Err(Status::permission_denied(e.to_string())),
+        // }
+    }
+
+    async fn ban_user(&self, _req: Request<BanUserReq>) -> Result<Response<Empty>, Status> {
+        todo!();
+        // let req = check_sig_from_req(req)?;
+        // self.data.ban_user(&req.operator_wallet, &req.user_wallet);
+        // Ok(Response::new(Empty {}))
+    }
+
+    // admin commands
+
+    async fn airdrop(&self, req: Request<AirdropReq>) -> Result<Response<Empty>, Status> {
+        check_admin_key(&req)?;
+        let req = check_sig_from_req(req)?;
+        db::Account::airdrop(&self.db, &req.pubkey, req.tokens).await?;
+        Ok(Response::new(Empty {}))
+    }
+
+    async fn slash(&self, _req: Request<SlashReq>) -> Result<Response<Empty>, Status> {
+        todo!();
+        // check_admin_key(&req)?;
+        // let req = check_sig_from_req(req)?;
+        // self.data.slash_account(&req.pubkey, req.tokens);
+        // Ok(Response::new(Empty {}))
+    }
+
+    async fn list_accounts(
+        &self,
+        _req: Request<Empty>,
+    ) -> Result<Response<Self::ListAccountsStream>, Status> {
+        todo!();
+        // check_admin_key(&req)?;
+        // let _ = check_sig_from_req(req)?;
+        // let accounts = self.data.list_accounts();
+        // let (tx, rx) = mpsc::channel(6);
+        // tokio::spawn(async move {
+        //     for account in accounts {
+        //         let _ = tx.send(Ok(account.into())).await;
+        //     }
+        // });
+        // let output_stream = ReceiverStream::new(rx);
+        // Ok(Response::new(Box::pin(output_stream) as Self::ListAccountsStream))
+    }
+
+    async fn list_all_vm_contracts(
+        &self,
+        _req: Request<Empty>,
+    ) -> Result<Response<Self::ListAllVmContractsStream>, Status> {
+        todo!();
+        // check_admin_key(&req)?;
+        // let _ = check_sig_from_req(req)?;
+        // let contracts = self.data.list_all_contracts();
+        // let (tx, rx) = mpsc::channel(6);
+        // tokio::spawn(async move {
+        //     for contract in contracts {
+        //         let _ = tx.send(Ok(contract.into())).await;
+        //     }
+        // });
+        // let output_stream = ReceiverStream::new(rx);
+        // Ok(Response::new(Box::pin(output_stream) as Self::ListAllVmContractsStream))
+    }
+
+    async fn list_all_app_contracts(
+        &self,
+        _req: tonic::Request<Empty>,
+    ) -> Result<tonic::Response<Self::ListAllAppContractsStream>, Status> {
+        todo!();
+        // check_admin_key(&req)?;
+        // let _ = check_sig_from_req(req)?;
+        // let contracts = self.data.list_all_app_contracts();
+        // let (tx, rx) = mpsc::channel(6);
+        // tokio::spawn(async move {
+        //     for contract in contracts {
+        //         let _ = tx.send(Ok(contract.into())).await;
+        //     }
+        // });
+        // let output_stream = ReceiverStream::new(rx);
+        // Ok(Response::new(Box::pin(output_stream)))
+    }
+}
+
+pub struct BrainVmCliForReal {
+    pub db: Arc<Surreal<Client>>,
+}
+
+impl BrainVmCliForReal {
+    pub fn new(db: Arc<Surreal<Client>>) -> Self {
+        Self { db }
+    }
+}
+
+#[tonic::async_trait]
+impl BrainVmCli for BrainVmCliForReal {
+    type ListVmContractsStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
+    type ListVmNodesStream = Pin<Box<dyn Stream<Item = Result<VmNodeListResp, Status>> + Send>>;
+
+    async fn new_vm(&self, req: Request<NewVmReq>) -> Result<Response<NewVmResp>, Status> {
+        let req = check_sig_from_req(req)?;
+        info!("New VM requested via CLI: {req:?}");
+        if db::Account::is_banned_by_node(&self.db, &req.admin_pubkey, &req.node_pubkey).await? {
+            return Err(Status::permission_denied("This operator banned you. What did you do?"));
+        }
+
+        let new_vm_req: db::NewVmReq = req.into();
+        let id = new_vm_req.id.key().to_string();
+
+        let db = self.db.clone();
+
+        let (oneshot_tx, oneshot_rx) = tokio::sync::oneshot::channel();
+        tokio::spawn(async move {
+            let _ = oneshot_tx.send(db::NewVmResp::listen(&db, &id).await);
+        });
+        new_vm_req.submit(&self.db).await?;
+
+        match oneshot_rx.await {
+            Ok(Err(db::Error::TimeOut(_))) => Err(Status::deadline_exceeded("Request failed due to timeout. Please try again later or contact the DeTEE devs team.")),
+            Ok(new_vm_resp) => Ok(Response::new(new_vm_resp?.into())),
+            Err(e) => {
+                log::error!("Something weird happened. Reached error {e:?}");
+                Err(Status::unknown(
+                    "Request failed due to unknown error. Please try again or contact the DeTEE devs team.",
+                ))
+            }
+        }
+    }
+
+    async fn update_vm(&self, req: Request<UpdateVmReq>) -> Result<Response<UpdateVmResp>, Status> {
+        let req = check_sig_from_req(req)?;
+        info!("Update VM requested via CLI: {req:?}");
+        todo!();
+        // let (oneshot_tx, oneshot_rx) = tokio::sync::oneshot::channel();
+        // self.data.submit_updatevm_req(req, oneshot_tx).await;
+        // match oneshot_rx.await {
+        //     Ok(response) => {
+        //         info!("Sending UpdateVMResp: {response:?}");
+        //         Ok(Response::new(response))
+        //     }
+        //     Err(e) => Err(Status::unknown(format!(
+        //         "Update VM request failed due to error: {e}"
+        //     ))),
+        // }
+    }
+
+    async fn extend_vm(&self, req: Request<ExtendVmReq>) -> Result<Response<Empty>, Status> {
+        let _req = check_sig_from_req(req)?;
+        todo!();
+        // match self
+        //     .data
+        //     .extend_vm_contract_time(&req.uuid, &req.admin_pubkey, req.locked_nano)
+        // {
+        //     Ok(()) => Ok(Response::new(Empty {})),
+        //     Err(e) => Err(Status::unknown(format!("Could not extend contract: {e}"))),
+        // }
+    }
+
+    async fn delete_vm(&self, req: Request<DeleteVmReq>) -> Result<Response<Empty>, Status> {
+        let _req = check_sig_from_req(req)?;
+        todo!();
+        // match self.data.delete_vm(req).await {
+        //     Ok(()) => Ok(Response::new(Empty {})),
+        //     Err(e) => Err(Status::not_found(e.to_string())),
+        // }
+    }
+
+    async fn list_vm_contracts(
+        &self,
+        req: Request<ListVmContractsReq>,
+    ) -> Result<Response<Self::ListVmContractsStream>, Status> {
+        let req = check_sig_from_req(req)?;
+        info!(
+            "CLI {} requested ListVMVmContractsStream. As operator: {}",
+            req.wallet, req.as_operator
+        );
+        let mut contracts = Vec::new();
+        if !req.uuid.is_empty() {
+            if let Some(specific_contract) =
+                db::ActiveVmWithNode::get_by_uuid(&self.db, &req.uuid).await?
+            {
+                if specific_contract.admin.key().to_string() == req.wallet {
+                    contracts.push(specific_contract);
+                }
+                // TODO: allow operator to inspect contracts
+            }
+        } else if req.as_operator {
+            contracts
+                .append(&mut db::ActiveVmWithNode::list_by_operator(&self.db, &req.wallet).await?);
+        } else {
+            contracts
+                .append(&mut db::ActiveVmWithNode::list_by_admin(&self.db, &req.wallet).await?);
+        }
+        let (tx, rx) = mpsc::channel(6);
+        tokio::spawn(async move {
+            for contract in contracts {
+                let _ = tx.send(Ok(contract.into())).await;
+            }
+        });
+        let output_stream = ReceiverStream::new(rx);
+        Ok(Response::new(Box::pin(output_stream) as Self::ListVmContractsStream))
+    }
+
+    async fn list_vm_nodes(
+        &self,
+        req: Request<VmNodeFilters>,
+    ) -> Result<Response<Self::ListVmNodesStream>, tonic::Status> {
+        let req = check_sig_from_req(req)?;
+        info!("CLI requested ListVmNodesStream: {req:?}");
+        let nodes = db::VmNodeWithReports::find_by_filters(&self.db, req).await?;
+        let (tx, rx) = mpsc::channel(6);
+        tokio::spawn(async move {
+            for node in nodes {
+                let _ = tx.send(Ok(node.into())).await;
+            }
+        });
+        let output_stream = ReceiverStream::new(rx);
+        Ok(Response::new(Box::pin(output_stream) as Self::ListVmNodesStream))
+    }
+
+    async fn get_one_vm_node(
+        &self,
+        req: Request<VmNodeFilters>,
+    ) -> Result<Response<VmNodeListResp>, Status> {
+        let req = check_sig_from_req(req)?;
+        info!("Unknown CLI requested ListVmNodesStream: {req:?}");
+        // TODO: optimize this query so that it gets only one node
+        let nodes = db::VmNodeWithReports::find_by_filters(&self.db, req).await?;
+        if let Some(node) = nodes.into_iter().next() {
+            return Ok(Response::new(node.into()));
+        }
+        Err(Status::not_found("Could not find any node based on your search criteria"))
+    }
+}
+
+trait PubkeyGetter {
+    fn get_pubkey(&self) -> Option<String>;
+}
+
+macro_rules! impl_pubkey_getter {
+    ($t:ty, $field:ident) => {
+        impl PubkeyGetter for $t {
+            fn get_pubkey(&self) -> Option<String> {
+                Some(self.$field.clone())
+            }
+        }
+    };
+    ($t:ty) => {
+        impl PubkeyGetter for $t {
+            fn get_pubkey(&self) -> Option<String> {
+                None
+            }
+        }
+    };
+}
+
+impl_pubkey_getter!(Pubkey, pubkey);
+impl_pubkey_getter!(NewVmReq, admin_pubkey);
+impl_pubkey_getter!(DeleteVmReq, admin_pubkey);
+impl_pubkey_getter!(UpdateVmReq, admin_pubkey);
+impl_pubkey_getter!(ExtendVmReq, admin_pubkey);
+impl_pubkey_getter!(ReportNodeReq, admin_pubkey);
+impl_pubkey_getter!(ListVmContractsReq, wallet);
+impl_pubkey_getter!(RegisterVmNodeReq, node_pubkey);
+impl_pubkey_getter!(RegOperatorReq, pubkey);
+impl_pubkey_getter!(KickReq, operator_wallet);
+impl_pubkey_getter!(BanUserReq, operator_wallet);
+
+impl_pubkey_getter!(VmNodeFilters);
+impl_pubkey_getter!(Empty);
+impl_pubkey_getter!(AirdropReq);
+impl_pubkey_getter!(SlashReq);
+
+// impl_pubkey_getter!(NewAppReq, admin_pubkey);
+// impl_pubkey_getter!(DelAppReq, admin_pubkey);
+// impl_pubkey_getter!(ListAppContractsReq, admin_pubkey);
+//
+// impl_pubkey_getter!(RegisterAppNodeReq);
+// impl_pubkey_getter!(AppNodeFilters);
+
+fn check_sig_from_req<T: std::fmt::Debug + PubkeyGetter>(req: Request<T>) -> Result<T, Status> {
+    let time = match req.metadata().get("timestamp") {
+        Some(t) => t.clone(),
+        None => return Err(Status::unauthenticated("Timestamp not found in metadata.")),
+    };
+    let time = time
+        .to_str()
+        .map_err(|_| Status::unauthenticated("Timestamp in metadata is not a string"))?;
+
+    let now = chrono::Utc::now();
+    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
+        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
+    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
+    if !(-4..=4).contains(&seconds_elapsed) {
+        return Err(Status::unauthenticated(format!(
+            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
+            parsed_time, now
+        )));
+    }
+
+    let signature = match req.metadata().get("request-signature") {
+        Some(t) => t,
+        None => return Err(Status::unauthenticated("signature not found in metadata.")),
+    };
+    let signature = bs58::decode(signature)
+        .into_vec()
+        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
+    let signature = ed25519_dalek::Signature::from_bytes(
+        signature
+            .as_slice()
+            .try_into()
+            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
+    );
+
+    let pubkey_value = match req.metadata().get("pubkey") {
+        Some(p) => p.clone(),
+        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
+    };
+    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
+        &bs58::decode(&pubkey_value)
+            .into_vec()
+            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
+            .try_into()
+            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
+    )
+    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
+
+    let req = req.into_inner();
+    let message = format!("{time}{req:?}");
+    use ed25519_dalek::Verifier;
+    pubkey
+        .verify(message.as_bytes(), &signature)
+        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
+    if let Some(req_pubkey) = req.get_pubkey() {
+        if *pubkey_value.to_str().unwrap() != req_pubkey {
+            return Err(Status::unauthenticated(
+                "pubkey of signature does not match pubkey of request",
+            ));
+        }
+    }
+    Ok(req)
+}
+
+fn check_sig_from_parts(pubkey: &str, time: &str, msg: &str, sig: &str) -> Result<(), Status> {
+    let now = chrono::Utc::now();
+    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
+        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
+    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
+    if !(-4..=4).contains(&seconds_elapsed) {
+        return Err(Status::unauthenticated(format!(
+            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
+            parsed_time, now
+        )));
+    }
+
+    let signature = bs58::decode(sig)
+        .into_vec()
+        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
+    let signature = ed25519_dalek::Signature::from_bytes(
+        signature
+            .as_slice()
+            .try_into()
+            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
+    );
+
+    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
+        &bs58::decode(&pubkey)
+            .into_vec()
+            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
+            .try_into()
+            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
+    )
+    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
+
+    let msg = time.to_string() + msg;
+    use ed25519_dalek::Verifier;
+    pubkey
+        .verify(msg.as_bytes(), &signature)
+        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
+
+    Ok(())
+}
+
+fn check_admin_key<T>(req: &Request<T>) -> Result<(), Status> {
+    let pubkey = match req.metadata().get("pubkey") {
+        Some(p) => p.clone(),
+        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
+    };
+    let pubkey = pubkey
+        .to_str()
+        .map_err(|_| Status::unauthenticated("could not parse pubkey metadata to str"))?;
+
+    if !ADMIN_ACCOUNTS.contains(&pubkey) {
+        return Err(Status::unauthenticated("This operation is reserved to admin accounts"));
+    }
+
+    Ok(())
+}

src/grpc/app.rs

@@ -1 +0,0 @@
-

src/grpc/general.rs

@@ -1,200 +0,0 @@
-use crate::db;
-use crate::grpc::{check_admin_key, check_sig_from_req};
-use detee_shared::app_proto::AppContract;
-use detee_shared::common_proto::{Empty, Pubkey};
-use detee_shared::general_proto::brain_general_cli_server::BrainGeneralCli;
-use detee_shared::general_proto::{
-    Account, AccountBalance, AirdropReq, BanUserReq, InspectOperatorResp, KickReq, KickResp,
-    ListOperatorsResp, RegOperatorReq, ReportNodeReq, SlashReq,
-};
-use detee_shared::vm_proto::VmContract;
-use surrealdb::engine::remote::ws::Client;
-use surrealdb::Surreal;
-
-use std::pin::Pin;
-use std::sync::Arc;
-use tokio::sync::mpsc;
-use tokio_stream::wrappers::ReceiverStream;
-use tokio_stream::Stream;
-use tonic::{Request, Response, Status};
-
-pub struct GeneralCliServer {
-    pub db: Arc<Surreal<Client>>,
-}
-
-impl GeneralCliServer {
-    pub fn new(db: Arc<Surreal<Client>>) -> Self {
-        Self { db }
-    }
-}
-
-#[tonic::async_trait]
-impl BrainGeneralCli for GeneralCliServer {
-    type ListAccountsStream = Pin<Box<dyn Stream<Item = Result<Account, Status>> + Send>>;
-    type ListAllAppContractsStream =
-        Pin<Box<dyn Stream<Item = Result<AppContract, Status>> + Send>>;
-    type ListAllVmContractsStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
-    type ListOperatorsStream =
-        Pin<Box<dyn Stream<Item = Result<ListOperatorsResp, Status>> + Send>>;
-
-    async fn get_balance(&self, req: Request<Pubkey>) -> Result<Response<AccountBalance>, Status> {
-        let req = check_sig_from_req(req)?;
-        Ok(Response::new(db::general::Account::get(&self.db, &req.pubkey).await?.into()))
-    }
-
-    async fn report_node(&self, req: Request<ReportNodeReq>) -> Result<Response<Empty>, Status> {
-        let req = check_sig_from_req(req)?;
-        let (account, node, contract_id) =
-            match db::vm::ActiveVmWithNode::get_by_uuid(&self.db, &req.contract).await? {
-                Some(vm_contract)
-                    if vm_contract.admin.key().to_string() == req.admin_pubkey
-                        && vm_contract.vm_node.id.key().to_string() == req.node_pubkey =>
-                {
-                    (vm_contract.admin, vm_contract.vm_node.id, vm_contract.id.to_string())
-                }
-                _ => match db::app::ActiveAppWithNode::get_by_uuid(&self.db, &req.contract).await? {
-                    Some(app_contract)
-                        if app_contract.admin.key().to_string() == req.admin_pubkey
-                            && app_contract.app_node.id.key().to_string() == req.node_pubkey =>
-                    {
-                        (app_contract.admin, app_contract.app_node.id, app_contract.id.to_string())
-                    }
-                    _ => {
-                        return Err(Status::unauthenticated("No contract found by this ID."));
-                    }
-                },
-            };
-        db::general::Report::create(&self.db, account, node, req.reason, contract_id).await?;
-        Ok(Response::new(Empty {}))
-    }
-
-    async fn list_operators(
-        &self,
-        req: Request<Empty>,
-    ) -> Result<Response<Self::ListOperatorsStream>, Status> {
-        let _ = check_sig_from_req(req)?;
-        let operators = db::general::Operator::list(&self.db).await?;
-        let (tx, rx) = mpsc::channel(6);
-        tokio::spawn(async move {
-            for op in operators {
-                let _ = tx.send(Ok(op.into())).await;
-            }
-        });
-        let output_stream = ReceiverStream::new(rx);
-        Ok(Response::new(Box::pin(output_stream) as Self::ListOperatorsStream))
-    }
-
-    async fn inspect_operator(
-        &self,
-        req: Request<Pubkey>,
-    ) -> Result<Response<InspectOperatorResp>, Status> {
-        match db::general::Operator::inspect_nodes(&self.db, &req.into_inner().pubkey).await? {
-            (Some(op), vm_nodes, app_nodes) => Ok(Response::new(InspectOperatorResp {
-                operator: Some(op.into()),
-                vm_nodes: vm_nodes.into_iter().map(|n| n.into()).collect(),
-                app_nodes: app_nodes.into_iter().map(|n| n.into()).collect(),
-            })),
-            (None, _, _) => Err(Status::not_found("The wallet you specified is not an operator")),
-        }
-    }
-
-    async fn register_operator(
-        &self,
-        _req: Request<RegOperatorReq>,
-    ) -> Result<Response<Empty>, Status> {
-        todo!();
-        // let req = check_sig_from_req(req)?;
-        // info!("Regitering new operator: {req:?}");
-        // match self.data.register_operator(req) {
-        //     Ok(()) => Ok(Response::new(Empty {})),
-        //     Err(e) => Err(Status::failed_precondition(e.to_string())),
-        // }
-    }
-
-    async fn kick_contract(&self, _req: Request<KickReq>) -> Result<Response<KickResp>, Status> {
-        todo!();
-        // let req = check_sig_from_req(req)?;
-        // match self.data.kick_contract(&req.operator_wallet, &req.contract_uuid, &req.reason).await {
-        //     Ok(nano_lp) => Ok(Response::new(KickResp { nano_lp })),
-        //     Err(e) => Err(Status::permission_denied(e.to_string())),
-        // }
-    }
-
-    async fn ban_user(&self, _req: Request<BanUserReq>) -> Result<Response<Empty>, Status> {
-        todo!();
-        // let req = check_sig_from_req(req)?;
-        // self.data.ban_user(&req.operator_wallet, &req.user_wallet);
-        // Ok(Response::new(Empty {}))
-    }
-
-    // admin commands
-
-    async fn airdrop(&self, req: Request<AirdropReq>) -> Result<Response<Empty>, Status> {
-        check_admin_key(&req)?;
-        let req = check_sig_from_req(req)?;
-        db::general::Account::airdrop(&self.db, &req.pubkey, req.tokens).await?;
-        Ok(Response::new(Empty {}))
-    }
-
-    async fn slash(&self, _req: Request<SlashReq>) -> Result<Response<Empty>, Status> {
-        todo!();
-        // check_admin_key(&req)?;
-        // let req = check_sig_from_req(req)?;
-        // self.data.slash_account(&req.pubkey, req.tokens);
-        // Ok(Response::new(Empty {}))
-    }
-
-    async fn list_accounts(
-        &self,
-        _req: Request<Empty>,
-    ) -> Result<Response<Self::ListAccountsStream>, Status> {
-        todo!();
-        // check_admin_key(&req)?;
-        // let _ = check_sig_from_req(req)?;
-        // let accounts = self.data.list_accounts();
-        // let (tx, rx) = mpsc::channel(6);
-        // tokio::spawn(async move {
-        //     for account in accounts {
-        //         let _ = tx.send(Ok(account.into())).await;
-        //     }
-        // });
-        // let output_stream = ReceiverStream::new(rx);
-        // Ok(Response::new(Box::pin(output_stream) as Self::ListAccountsStream))
-    }
-
-    async fn list_all_vm_contracts(
-        &self,
-        _req: Request<Empty>,
-    ) -> Result<Response<Self::ListAllVmContractsStream>, Status> {
-        todo!();
-        // check_admin_key(&req)?;
-        // let _ = check_sig_from_req(req)?;
-        // let contracts = self.data.list_all_contracts();
-        // let (tx, rx) = mpsc::channel(6);
-        // tokio::spawn(async move {
-        //     for contract in contracts {
-        //         let _ = tx.send(Ok(contract.into())).await;
-        //     }
-        // });
-        // let output_stream = ReceiverStream::new(rx);
-        // Ok(Response::new(Box::pin(output_stream) as Self::ListAllVmContractsStream))
-    }
-
-    async fn list_all_app_contracts(
-        &self,
-        _req: tonic::Request<Empty>,
-    ) -> Result<tonic::Response<Self::ListAllAppContractsStream>, Status> {
-        todo!();
-        // check_admin_key(&req)?;
-        // let _ = check_sig_from_req(req)?;
-        // let contracts = self.data.list_all_app_contracts();
-        // let (tx, rx) = mpsc::channel(6);
-        // tokio::spawn(async move {
-        //     for contract in contracts {
-        //         let _ = tx.send(Ok(contract.into())).await;
-        //     }
-        // });
-        // let output_stream = ReceiverStream::new(rx);
-        // Ok(Response::new(Box::pin(output_stream)))
-    }
-}

src/grpc/mod.rs

@@ -1,176 +0,0 @@
-pub mod app;
-pub mod general;
-pub mod types;
-pub mod vm;
-
-use crate::constants::ADMIN_ACCOUNTS;
-use detee_shared::common_proto::{Empty, Pubkey};
-use detee_shared::general_proto::{
-    AirdropReq, BanUserReq, KickReq, RegOperatorReq, ReportNodeReq, SlashReq,
-};
-use detee_shared::vm_proto::{ListVmContractsReq, *};
-use tonic::{Request, Status};
-
-pub trait PubkeyGetter {
-    fn get_pubkey(&self) -> Option<String>;
-}
-
-macro_rules! impl_pubkey_getter {
-    ($t:ty, $field:ident) => {
-        impl PubkeyGetter for $t {
-            fn get_pubkey(&self) -> Option<String> {
-                Some(self.$field.clone())
-            }
-        }
-    };
-    ($t:ty) => {
-        impl PubkeyGetter for $t {
-            fn get_pubkey(&self) -> Option<String> {
-                None
-            }
-        }
-    };
-}
-
-impl_pubkey_getter!(Pubkey, pubkey);
-impl_pubkey_getter!(NewVmReq, admin_pubkey);
-impl_pubkey_getter!(DeleteVmReq, admin_pubkey);
-impl_pubkey_getter!(UpdateVmReq, admin_pubkey);
-impl_pubkey_getter!(ExtendVmReq, admin_pubkey);
-impl_pubkey_getter!(ReportNodeReq, admin_pubkey);
-impl_pubkey_getter!(ListVmContractsReq, wallet);
-impl_pubkey_getter!(RegisterVmNodeReq, node_pubkey);
-impl_pubkey_getter!(RegOperatorReq, pubkey);
-impl_pubkey_getter!(KickReq, operator_wallet);
-impl_pubkey_getter!(BanUserReq, operator_wallet);
-
-impl_pubkey_getter!(VmNodeFilters);
-impl_pubkey_getter!(Empty);
-impl_pubkey_getter!(AirdropReq);
-impl_pubkey_getter!(SlashReq);
-
-// impl_pubkey_getter!(NewAppReq, admin_pubkey);
-// impl_pubkey_getter!(DelAppReq, admin_pubkey);
-// impl_pubkey_getter!(ListAppContractsReq, admin_pubkey);
-//
-// impl_pubkey_getter!(RegisterAppNodeReq);
-// impl_pubkey_getter!(AppNodeFilters);
-
-pub fn check_sig_from_req<T: std::fmt::Debug + PubkeyGetter>(req: Request<T>) -> Result<T, Status> {
-    let time = match req.metadata().get("timestamp") {
-        Some(t) => t.clone(),
-        None => return Err(Status::unauthenticated("Timestamp not found in metadata.")),
-    };
-    let time = time
-        .to_str()
-        .map_err(|_| Status::unauthenticated("Timestamp in metadata is not a string"))?;
-
-    let now = chrono::Utc::now();
-    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
-        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
-    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
-    if !(-4..=4).contains(&seconds_elapsed) {
-        return Err(Status::unauthenticated(format!(
-            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
-            parsed_time, now
-        )));
-    }
-
-    let signature = match req.metadata().get("request-signature") {
-        Some(t) => t,
-        None => return Err(Status::unauthenticated("signature not found in metadata.")),
-    };
-    let signature = bs58::decode(signature)
-        .into_vec()
-        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
-    let signature = ed25519_dalek::Signature::from_bytes(
-        signature
-            .as_slice()
-            .try_into()
-            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
-    );
-
-    let pubkey_value = match req.metadata().get("pubkey") {
-        Some(p) => p.clone(),
-        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
-    };
-    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
-        &bs58::decode(&pubkey_value)
-            .into_vec()
-            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
-            .try_into()
-            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
-    )
-    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
-
-    let req = req.into_inner();
-    let message = format!("{time}{req:?}");
-    use ed25519_dalek::Verifier;
-    pubkey
-        .verify(message.as_bytes(), &signature)
-        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
-    if let Some(req_pubkey) = req.get_pubkey() {
-        if *pubkey_value.to_str().unwrap() != req_pubkey {
-            return Err(Status::unauthenticated(
-                "pubkey of signature does not match pubkey of request",
-            ));
-        }
-    }
-    Ok(req)
-}
-
-pub fn check_sig_from_parts(pubkey: &str, time: &str, msg: &str, sig: &str) -> Result<(), Status> {
-    let now = chrono::Utc::now();
-    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
-        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
-    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
-    if !(-4..=4).contains(&seconds_elapsed) {
-        return Err(Status::unauthenticated(format!(
-            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
-            parsed_time, now
-        )));
-    }
-
-    let signature = bs58::decode(sig)
-        .into_vec()
-        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
-    let signature = ed25519_dalek::Signature::from_bytes(
-        signature
-            .as_slice()
-            .try_into()
-            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
-    );
-
-    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
-        &bs58::decode(&pubkey)
-            .into_vec()
-            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
-            .try_into()
-            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
-    )
-    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
-
-    let msg = time.to_string() + msg;
-    use ed25519_dalek::Verifier;
-    pubkey
-        .verify(msg.as_bytes(), &signature)
-        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
-
-    Ok(())
-}
-
-pub fn check_admin_key<T>(req: &Request<T>) -> Result<(), Status> {
-    let pubkey = match req.metadata().get("pubkey") {
-        Some(p) => p.clone(),
-        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
-    };
-    let pubkey = pubkey
-        .to_str()
-        .map_err(|_| Status::unauthenticated("could not parse pubkey metadata to str"))?;
-
-    if !ADMIN_ACCOUNTS.contains(&pubkey) {
-        return Err(Status::unauthenticated("This operation is reserved to admin accounts"));
-    }
-
-    Ok(())
-}

src/grpc/types.rs

@@ -1,221 +0,0 @@
-use crate::constants::{ACCOUNT, ID_ALPHABET, NEW_VM_REQ, VM_NODE};
-use crate::db::prelude as db;
-use detee_shared::app_proto::AppNodeListResp;
-use detee_shared::general_proto::{AccountBalance, ListOperatorsResp};
-use detee_shared::vm_proto::*;
-use nanoid::nanoid;
-
-use surrealdb::RecordId;
-
-impl From<db::Account> for AccountBalance {
-    fn from(account: db::Account) -> Self {
-        AccountBalance { balance: account.balance, tmp_locked: account.tmp_locked }
-    }
-}
-
-impl From<NewVmReq> for db::NewVmReq {
-    fn from(new_vm_req: NewVmReq) -> Self {
-        Self {
-            id: RecordId::from((NEW_VM_REQ, nanoid!(40, &ID_ALPHABET))),
-            admin: RecordId::from((ACCOUNT, new_vm_req.admin_pubkey)),
-            vm_node: RecordId::from((VM_NODE, new_vm_req.node_pubkey)),
-            hostname: new_vm_req.hostname,
-            extra_ports: new_vm_req.extra_ports,
-            public_ipv4: new_vm_req.public_ipv4,
-            public_ipv6: new_vm_req.public_ipv6,
-            disk_size_gb: new_vm_req.disk_size_gb,
-            vcpus: new_vm_req.vcpus,
-            memory_mb: new_vm_req.memory_mb,
-            kernel_url: new_vm_req.kernel_url,
-            kernel_sha: new_vm_req.kernel_sha,
-            dtrfs_url: new_vm_req.dtrfs_url,
-            dtrfs_sha: new_vm_req.dtrfs_sha,
-            price_per_unit: new_vm_req.price_per_unit,
-            locked_nano: new_vm_req.locked_nano,
-            created_at: surrealdb::sql::Datetime::default(),
-            error: String::new(),
-        }
-    }
-}
-
-impl From<db::NewVmReq> for NewVmReq {
-    fn from(new_vm_req: db::NewVmReq) -> Self {
-        Self {
-            uuid: new_vm_req.id.key().to_string(),
-            hostname: new_vm_req.hostname,
-            admin_pubkey: new_vm_req.admin.key().to_string(),
-            node_pubkey: new_vm_req.vm_node.key().to_string(),
-            extra_ports: new_vm_req.extra_ports,
-            public_ipv4: new_vm_req.public_ipv4,
-            public_ipv6: new_vm_req.public_ipv6,
-            disk_size_gb: new_vm_req.disk_size_gb,
-            vcpus: new_vm_req.vcpus,
-            memory_mb: new_vm_req.memory_mb,
-            kernel_url: new_vm_req.kernel_url,
-            kernel_sha: new_vm_req.kernel_sha,
-            dtrfs_url: new_vm_req.dtrfs_url,
-            dtrfs_sha: new_vm_req.dtrfs_sha,
-            price_per_unit: new_vm_req.price_per_unit,
-            locked_nano: new_vm_req.locked_nano,
-        }
-    }
-}
-
-impl From<db::NewVmResp> for NewVmResp {
-    fn from(resp: db::NewVmResp) -> Self {
-        match resp {
-            // TODO: This will require a small architecture change to pass MeasurementArgs from
-            // Daemon to CLI
-            db::NewVmResp::Args(uuid, args) => {
-                NewVmResp { uuid, error: String::new(), args: Some(args) }
-            }
-            db::NewVmResp::Error(uuid, error) => NewVmResp { uuid, error, args: None },
-        }
-    }
-}
-
-impl From<db::UpdateVmReq> for UpdateVmReq {
-    fn from(update_vm_req: db::UpdateVmReq) -> Self {
-        Self {
-            uuid: update_vm_req.id.key().to_string(),
-            // daemon does not care about VM hostname
-            hostname: String::new(),
-            admin_pubkey: update_vm_req.admin.key().to_string(),
-            disk_size_gb: update_vm_req.disk_size_gb,
-            vcpus: update_vm_req.vcpus,
-            memory_mb: update_vm_req.memory_mb,
-            kernel_url: update_vm_req.kernel_url,
-            kernel_sha: update_vm_req.kernel_sha,
-            dtrfs_url: update_vm_req.dtrfs_url,
-            dtrfs_sha: update_vm_req.dtrfs_sha,
-        }
-    }
-}
-
-impl From<db::DeletedVm> for DeleteVmReq {
-    fn from(delete_vm_req: db::DeletedVm) -> Self {
-        Self {
-            uuid: delete_vm_req.id.key().to_string(),
-            admin_pubkey: delete_vm_req.admin.key().to_string(),
-        }
-    }
-}
-
-impl From<db::VmDaemonNotification> for BrainVmMessage {
-    fn from(notification: db::VmDaemonNotification) -> Self {
-        match notification {
-            db::VmDaemonNotification::Create(new_vm_req) => {
-                BrainVmMessage { msg: Some(brain_vm_message::Msg::NewVmReq(new_vm_req.into())) }
-            }
-            db::VmDaemonNotification::Update(update_vm_req) => BrainVmMessage {
-                msg: Some(brain_vm_message::Msg::UpdateVmReq(update_vm_req.into())),
-            },
-            db::VmDaemonNotification::Delete(deleted_vm) => {
-                BrainVmMessage { msg: Some(brain_vm_message::Msg::DeleteVm(deleted_vm.into())) }
-            }
-        }
-    }
-}
-
-impl From<db::ActiveVmWithNode> for VmContract {
-    fn from(db_c: db::ActiveVmWithNode) -> Self {
-        let mut exposed_ports = Vec::new();
-        for port in db_c.mapped_ports.iter() {
-            exposed_ports.push(port.0);
-        }
-        VmContract {
-            uuid: db_c.id.key().to_string(),
-            hostname: db_c.hostname.clone(),
-            admin_pubkey: db_c.admin.key().to_string(),
-            node_pubkey: db_c.vm_node.id.key().to_string(),
-            node_ip: db_c.vm_node.ip.clone(),
-            location: format!(
-                "{}, {}, {}",
-                db_c.vm_node.city, db_c.vm_node.region, db_c.vm_node.country
-            ),
-            memory_mb: db_c.memory_mb,
-            vcpus: db_c.vcpus,
-            disk_size_gb: db_c.disk_size_gb,
-            mapped_ports: db_c
-                .mapped_ports
-                .iter()
-                .map(|(h, g)| MappedPort { host_port: *h, guest_port: *g })
-                .collect(),
-            vm_public_ipv6: db_c.public_ipv6.clone(),
-            vm_public_ipv4: db_c.public_ipv4.clone(),
-            locked_nano: db_c.locked_nano,
-            dtrfs_sha: db_c.dtrfs_sha.clone(),
-            kernel_sha: db_c.kernel_sha.clone(),
-            nano_per_minute: db_c.price_per_minute(),
-            created_at: db_c.created_at.to_rfc3339(),
-            // TODO: remove updated_at from the proto
-            // This will get moved to VM history (users will be able to
-            // query old contracts, which also shows updates of existing contracts).
-            updated_at: db_c.created_at.to_rfc3339(),
-            collected_at: db_c.collected_at.to_rfc3339(),
-        }
-    }
-}
-
-impl From<db::Error> for tonic::Status {
-    fn from(e: db::Error) -> Self {
-        Self::internal(format!("Internal error: {e}"))
-    }
-}
-
-impl From<db::Operator> for ListOperatorsResp {
-    fn from(db_o: db::Operator) -> Self {
-        ListOperatorsResp {
-            pubkey: db_o.account.key().to_string(),
-            escrow: db_o.escrow,
-            email: db_o.email,
-            app_nodes: db_o.app_nodes,
-            vm_nodes: db_o.vm_nodes,
-            reports: db_o.reports,
-        }
-    }
-}
-
-impl From<db::VmNodeWithReports> for VmNodeListResp {
-    fn from(vm_node: db::VmNodeWithReports) -> Self {
-        Self {
-            operator: vm_node.operator.key().to_string(),
-            node_pubkey: vm_node.id.key().to_string(),
-            country: vm_node.country,
-            region: vm_node.region,
-            city: vm_node.city,
-            ip: vm_node.ip,
-            reports: vm_node.reports.iter().map(|n| n.reason.clone()).collect(),
-            price: vm_node.price,
-        }
-    }
-}
-
-impl From<db::AppNodeWithReports> for AppNodeListResp {
-    fn from(app_node: db::AppNodeWithReports) -> Self {
-        Self {
-            operator: app_node.operator.key().to_string(),
-            node_pubkey: app_node.id.key().to_string(),
-            country: app_node.country,
-            region: app_node.region,
-            city: app_node.city,
-            ip: app_node.ip,
-            reports: app_node.reports.iter().map(|n| n.reason.clone()).collect(),
-            price: app_node.price,
-        }
-    }
-}
-
-impl From<VmNodeResources> for db::VmNodeResources {
-    fn from(res: VmNodeResources) -> Self {
-        Self {
-            avail_mem_mb: res.avail_memory_mb,
-            avail_vcpus: res.avail_vcpus,
-            avail_storage_gbs: res.avail_storage_gb,
-            avail_ipv4: res.avail_ipv4,
-            avail_ipv6: res.avail_ipv6,
-            avail_ports: res.avail_ports,
-            max_ports_per_vm: res.max_ports_per_vm,
-        }
-    }
-}

src/grpc/vm.rs

@@ -1,341 +0,0 @@
-#![allow(dead_code)]
-use crate::constants::{ACCOUNT, VM_NODE};
-use crate::db::prelude as db;
-use crate::grpc::{check_sig_from_parts, check_sig_from_req};
-use detee_shared::common_proto::Empty;
-use detee_shared::vm_proto::brain_vm_cli_server::BrainVmCli;
-use detee_shared::vm_proto::brain_vm_daemon_server::BrainVmDaemon;
-use detee_shared::vm_proto::{ListVmContractsReq, *};
-use surrealdb::engine::remote::ws::Client;
-use surrealdb::Surreal;
-
-use log::info;
-use std::pin::Pin;
-use std::sync::Arc;
-use tokio::sync::mpsc;
-use tokio_stream::wrappers::ReceiverStream;
-use tokio_stream::{Stream, StreamExt};
-use tonic::{Request, Response, Status, Streaming};
-
-pub struct VmDaemonServer {
-    pub db: Arc<Surreal<Client>>,
-}
-
-impl VmDaemonServer {
-    pub fn new(db: Arc<Surreal<Client>>) -> Self {
-        Self { db }
-    }
-}
-
-#[tonic::async_trait]
-impl BrainVmDaemon for VmDaemonServer {
-    type BrainMessagesStream = Pin<Box<dyn Stream<Item = Result<BrainVmMessage, Status>> + Send>>;
-    type RegisterVmNodeStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
-
-    async fn register_vm_node(
-        &self,
-        req: Request<RegisterVmNodeReq>,
-    ) -> Result<Response<Self::RegisterVmNodeStream>, Status> {
-        let req = check_sig_from_req(req)?;
-        info!("Starting registration process for {:?}", req);
-        db::VmNode {
-            id: surrealdb::RecordId::from((VM_NODE, req.node_pubkey.clone())),
-            operator: surrealdb::RecordId::from((ACCOUNT, req.operator_wallet)),
-            country: req.country,
-            region: req.region,
-            city: req.city,
-            ip: req.main_ip,
-            price: req.price,
-            avail_mem_mb: 0,
-            avail_vcpus: 0,
-            avail_storage_gbs: 0,
-            avail_ipv4: 0,
-            avail_ipv6: 0,
-            avail_ports: 0,
-            max_ports_per_vm: 0,
-            offline_minutes: 0,
-        }
-        .register(&self.db)
-        .await?;
-
-        info!("Sending existing contracts to {}", req.node_pubkey);
-        let contracts = db::ActiveVmWithNode::list_by_node(&self.db, &req.node_pubkey).await?;
-        let (tx, rx) = mpsc::channel(6);
-        tokio::spawn(async move {
-            for contract in contracts {
-                let _ = tx.send(Ok(contract.into())).await;
-            }
-        });
-        let output_stream = ReceiverStream::new(rx);
-        Ok(Response::new(Box::pin(output_stream) as Self::RegisterVmNodeStream))
-    }
-
-    async fn brain_messages(
-        &self,
-        req: Request<DaemonStreamAuth>,
-    ) -> Result<Response<Self::BrainMessagesStream>, Status> {
-        let auth = req.into_inner();
-        let pubkey = auth.pubkey.clone();
-        check_sig_from_parts(
-            &pubkey,
-            &auth.timestamp,
-            &format!("{:?}", auth.contracts),
-            &auth.signature,
-        )?;
-        info!("Daemon {} connected to receive brain messages", pubkey);
-
-        let (tx, rx) = mpsc::channel(6);
-        {
-            let db = self.db.clone();
-            let pubkey = pubkey.clone();
-            let tx = tx.clone();
-            tokio::spawn(async move {
-                match db::listen_for_vm_node::<db::DeletedVm>(&db, &pubkey, tx).await {
-                    Ok(()) => log::info!("db::VmContract::listen_for_node ended for {pubkey}"),
-                    Err(e) => {
-                        log::warn!("db::VmContract::listen_for_node errored for {pubkey}: {e}")
-                    }
-                };
-            });
-        }
-        {
-            let db = self.db.clone();
-            let pubkey = pubkey.clone();
-            let tx = tx.clone();
-            tokio::spawn(async move {
-                let _ = db::listen_for_vm_node::<db::NewVmReq>(&db, &pubkey, tx.clone()).await;
-            });
-        }
-        {
-            let db = self.db.clone();
-            let pubkey = pubkey.clone();
-            let tx = tx.clone();
-            tokio::spawn(async move {
-                let _ = db::listen_for_vm_node::<db::UpdateVmReq>(&db, &pubkey, tx.clone()).await;
-            });
-        }
-
-        let output_stream = ReceiverStream::new(rx).map(|msg| Ok(msg.into()));
-        Ok(Response::new(Box::pin(output_stream) as Self::BrainMessagesStream))
-    }
-
-    async fn daemon_messages(
-        &self,
-        req: Request<Streaming<VmDaemonMessage>>,
-    ) -> Result<Response<Empty>, Status> {
-        let mut req_stream = req.into_inner();
-        let pubkey: String;
-        if let Some(Ok(msg)) = req_stream.next().await {
-            log::debug!("demon_messages received the following auth message: {:?}", msg.msg);
-            if let Some(vm_daemon_message::Msg::Auth(auth)) = msg.msg {
-                pubkey = auth.pubkey.clone();
-                check_sig_from_parts(
-                    &pubkey,
-                    &auth.timestamp,
-                    &format!("{:?}", auth.contracts),
-                    &auth.signature,
-                )?;
-            } else {
-                return Err(Status::unauthenticated(
-                    "Could not authenticate the daemon: could not extract auth signature",
-                ));
-            }
-        } else {
-            return Err(Status::unauthenticated("Could not authenticate the daemon"));
-        }
-
-        while let Some(daemon_message) = req_stream.next().await {
-            match daemon_message {
-                Ok(msg) => match msg.msg {
-                    Some(vm_daemon_message::Msg::NewVmResp(new_vm_resp)) => {
-                        // TODO: move new_vm_req to active_vm
-                        // also handle failure properly
-                        if !new_vm_resp.error.is_empty() {
-                            db::NewVmReq::submit_error(
-                                &self.db,
-                                &new_vm_resp.uuid,
-                                new_vm_resp.error,
-                            )
-                            .await?;
-                        } else {
-                            db::upsert_record(
-                                &self.db,
-                                "measurement_args",
-                                &new_vm_resp.uuid,
-                                new_vm_resp.args.clone(),
-                            )
-                            .await?;
-                            if let Some(args) = new_vm_resp.args {
-                                db::ActiveVm::activate(&self.db, &new_vm_resp.uuid, args).await?;
-                            }
-                        }
-                    }
-                    Some(vm_daemon_message::Msg::UpdateVmResp(_update_vm_resp)) => {
-                        todo!();
-                        // self.data.submit_updatevm_resp(update_vm_resp).await;
-                    }
-                    Some(vm_daemon_message::Msg::VmNodeResources(node_resources)) => {
-                        let node_resources: db::VmNodeResources = node_resources.into();
-                        node_resources.merge(&self.db, &pubkey).await?;
-                    }
-                    _ => {}
-                },
-                Err(e) => {
-                    log::warn!("Daemon disconnected: {e:?}");
-                }
-            }
-        }
-        Ok(Response::new(Empty {}))
-    }
-}
-
-pub struct VmCliServer {
-    pub db: Arc<Surreal<Client>>,
-}
-
-impl VmCliServer {
-    pub fn new(db: Arc<Surreal<Client>>) -> Self {
-        Self { db }
-    }
-}
-
-#[tonic::async_trait]
-impl BrainVmCli for VmCliServer {
-    type ListVmContractsStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
-    type ListVmNodesStream = Pin<Box<dyn Stream<Item = Result<VmNodeListResp, Status>> + Send>>;
-
-    async fn new_vm(&self, req: Request<NewVmReq>) -> Result<Response<NewVmResp>, Status> {
-        let req = check_sig_from_req(req)?;
-        info!("New VM requested via CLI: {req:?}");
-        if db::general::Account::is_banned_by_node(&self.db, &req.admin_pubkey, &req.node_pubkey).await? {
-            return Err(Status::permission_denied("This operator banned you. What did you do?"));
-        }
-
-        let new_vm_req: db::NewVmReq = req.into();
-        let id = new_vm_req.id.key().to_string();
-
-        let db = self.db.clone();
-
-        let (oneshot_tx, oneshot_rx) = tokio::sync::oneshot::channel();
-        tokio::spawn(async move {
-            let _ = oneshot_tx.send(db::NewVmResp::listen(&db, &id).await);
-        });
-        new_vm_req.submit(&self.db).await?;
-
-        match oneshot_rx.await {
-            Ok(Err(db::Error::TimeOut(_))) => Err(Status::deadline_exceeded("Request failed due to timeout. Please try again later or contact the DeTEE devs team.")),
-            Ok(new_vm_resp) => Ok(Response::new(new_vm_resp?.into())),
-            Err(e) => {
-                log::error!("Something weird happened. Reached error {e:?}");
-                Err(Status::unknown(
-                    "Request failed due to unknown error. Please try again or contact the DeTEE devs team.",
-                ))
-            }
-        }
-    }
-
-    async fn update_vm(&self, req: Request<UpdateVmReq>) -> Result<Response<UpdateVmResp>, Status> {
-        let req = check_sig_from_req(req)?;
-        info!("Update VM requested via CLI: {req:?}");
-        todo!();
-        // let (oneshot_tx, oneshot_rx) = tokio::sync::oneshot::channel();
-        // self.data.submit_updatevm_req(req, oneshot_tx).await;
-        // match oneshot_rx.await {
-        //     Ok(response) => {
-        //         info!("Sending UpdateVMResp: {response:?}");
-        //         Ok(Response::new(response))
-        //     }
-        //     Err(e) => Err(Status::unknown(format!(
-        //         "Update VM request failed due to error: {e}"
-        //     ))),
-        // }
-    }
-
-    async fn extend_vm(&self, req: Request<ExtendVmReq>) -> Result<Response<Empty>, Status> {
-        let _req = check_sig_from_req(req)?;
-        todo!();
-        // match self
-        //     .data
-        //     .extend_vm_contract_time(&req.uuid, &req.admin_pubkey, req.locked_nano)
-        // {
-        //     Ok(()) => Ok(Response::new(Empty {})),
-        //     Err(e) => Err(Status::unknown(format!("Could not extend contract: {e}"))),
-        // }
-    }
-
-    async fn delete_vm(&self, req: Request<DeleteVmReq>) -> Result<Response<Empty>, Status> {
-        let _req = check_sig_from_req(req)?;
-        todo!();
-        // match self.data.delete_vm(req).await {
-        //     Ok(()) => Ok(Response::new(Empty {})),
-        //     Err(e) => Err(Status::not_found(e.to_string())),
-        // }
-    }
-
-    async fn list_vm_contracts(
-        &self,
-        req: Request<ListVmContractsReq>,
-    ) -> Result<Response<Self::ListVmContractsStream>, Status> {
-        let req = check_sig_from_req(req)?;
-        info!(
-            "CLI {} requested ListVMVmContractsStream. As operator: {}",
-            req.wallet, req.as_operator
-        );
-        let mut contracts = Vec::new();
-        if !req.uuid.is_empty() {
-            if let Some(specific_contract) =
-                db::ActiveVmWithNode::get_by_uuid(&self.db, &req.uuid).await?
-            {
-                if specific_contract.admin.key().to_string() == req.wallet {
-                    contracts.push(specific_contract);
-                }
-                // TODO: allow operator to inspect contracts
-            }
-        } else if req.as_operator {
-            contracts
-                .append(&mut db::ActiveVmWithNode::list_by_operator(&self.db, &req.wallet).await?);
-        } else {
-            contracts
-                .append(&mut db::ActiveVmWithNode::list_by_admin(&self.db, &req.wallet).await?);
-        }
-        let (tx, rx) = mpsc::channel(6);
-        tokio::spawn(async move {
-            for contract in contracts {
-                let _ = tx.send(Ok(contract.into())).await;
-            }
-        });
-        let output_stream = ReceiverStream::new(rx);
-        Ok(Response::new(Box::pin(output_stream) as Self::ListVmContractsStream))
-    }
-
-    async fn list_vm_nodes(
-        &self,
-        req: Request<VmNodeFilters>,
-    ) -> Result<Response<Self::ListVmNodesStream>, tonic::Status> {
-        let req = check_sig_from_req(req)?;
-        info!("CLI requested ListVmNodesStream: {req:?}");
-        let nodes = db::VmNodeWithReports::find_by_filters(&self.db, req).await?;
-        let (tx, rx) = mpsc::channel(6);
-        tokio::spawn(async move {
-            for node in nodes {
-                let _ = tx.send(Ok(node.into())).await;
-            }
-        });
-        let output_stream = ReceiverStream::new(rx);
-        Ok(Response::new(Box::pin(output_stream) as Self::ListVmNodesStream))
-    }
-
-    async fn get_one_vm_node(
-        &self,
-        req: Request<VmNodeFilters>,
-    ) -> Result<Response<VmNodeListResp>, Status> {
-        let req = check_sig_from_req(req)?;
-        info!("Unknown CLI requested ListVmNodesStream: {req:?}");
-        // TODO: optimize this query so that it gets only one node
-        let nodes = db::VmNodeWithReports::find_by_filters(&self.db, req).await?;
-        if let Some(node) = nodes.into_iter().next() {
-            return Ok(Response::new(node.into()));
-        }
-        Err(Status::not_found("Could not find any node based on your search criteria"))
-    }
-}

tests/common/prepare_test_env.rs

@@ -1,19 +1,17 @@
-use detee_shared::general_proto::brain_general_cli_server::BrainGeneralCliServer;
 use detee_shared::vm_proto::brain_vm_cli_server::BrainVmCliServer;
-use detee_shared::vm_proto::brain_vm_daemon_server::BrainVmDaemonServer;
+use detee_shared::{
+    general_proto::brain_general_cli_server::BrainGeneralCliServer,
+    vm_proto::brain_vm_daemon_server::BrainVmDaemonServer,
+};
 use dotenv::dotenv;
 use hyper_util::rt::TokioIo;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use surreal_brain::grpc::{
-    general::GeneralCliServer,
-    vm::{VmCliServer, VmDaemonServer},
-};
+use surreal_brain::grpc::{BrainGeneralCliForReal, BrainVmCliForReal, BrainVmDaemonForReal};
 use surrealdb::engine::remote::ws::Client;
 use surrealdb::Surreal;
 use tokio::io::DuplexStream;
-use tokio::net::TcpListener;
-use tokio::sync::OnceCell;
+use tokio::{net::TcpListener, sync::OnceCell};
 use tonic::transport::{Channel, Endpoint, Server, Uri};
 use tower::service_fn;
 
@@ -60,9 +58,9 @@ pub async fn run_service_in_background() -> SocketAddr {
         let db_arc = Arc::new(db);
 
         Server::builder()
-            .add_service(BrainGeneralCliServer::new(GeneralCliServer::new(db_arc.clone())))
-            .add_service(BrainVmCliServer::new(VmCliServer::new(db_arc.clone())))
-            .add_service(BrainVmDaemonServer::new(VmDaemonServer::new(db_arc.clone())))
+            .add_service(BrainGeneralCliServer::new(BrainGeneralCliForReal::new(db_arc.clone())))
+            .add_service(BrainVmCliServer::new(BrainVmCliForReal::new(db_arc.clone())))
+            .add_service(BrainVmDaemonServer::new(BrainVmDaemonForReal::new(db_arc.clone())))
             .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
             .await
             .unwrap();
@@ -90,9 +88,9 @@ pub async fn run_service_for_stream_server() -> DuplexStream {
         let db_arc = Arc::new(db);
 
         tonic::transport::Server::builder()
-            .add_service(BrainGeneralCliServer::new(GeneralCliServer::new(db_arc.clone())))
-            .add_service(BrainVmCliServer::new(VmCliServer::new(db_arc.clone())))
-            .add_service(BrainVmDaemonServer::new(VmDaemonServer::new(db_arc.clone())))
+            .add_service(BrainGeneralCliServer::new(BrainGeneralCliForReal::new(db_arc.clone())))
+            .add_service(BrainVmCliServer::new(BrainVmCliForReal::new(db_arc.clone())))
+            .add_service(BrainVmDaemonServer::new(BrainVmDaemonForReal::new(db_arc.clone())))
             .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
             .await
     });

tests/common/test_utils.rs

@@ -1,6 +1,7 @@
 use anyhow::Result;
 use detee_shared::vm_proto as snp_proto;
-use ed25519_dalek::{Signer, SigningKey};
+use ed25519_dalek::Signer;
+use ed25519_dalek::SigningKey;
 use tonic::metadata::AsciiMetadataValue;
 use tonic::Request;
 

tests/common/vm_cli_utils.rs

@@ -2,7 +2,7 @@ use super::test_utils::Key;
 use detee_shared::vm_proto;
 use detee_shared::vm_proto::brain_vm_cli_client::BrainVmCliClient;
 use surreal_brain::constants::{ACTIVE_VM, NEW_VM_REQ};
-use surreal_brain::db::prelude as db;
+use surreal_brain::db;
 use surrealdb::engine::remote::ws::Client;
 use surrealdb::Surreal;
 use tonic::transport::Channel;

tests/grpc_test.rs

@@ -1,17 +1,19 @@
-use common::prepare_test_env::{
-    prepare_test_db, run_service_for_stream, run_service_in_background,
+use common::{
+    prepare_test_env::{prepare_test_db, run_service_for_stream, run_service_in_background},
+    test_utils::Key,
+    vm_cli_utils::create_new_vm,
+    vm_daemon_utils::mock_vm_daemon,
 };
-use common::test_utils::Key;
-use common::vm_cli_utils::create_new_vm;
-use common::vm_daemon_utils::mock_vm_daemon;
-use detee_shared::common_proto::{Empty, Pubkey};
-use detee_shared::general_proto::brain_general_cli_client::BrainGeneralCliClient;
+use detee_shared::common_proto::Empty;
 use detee_shared::general_proto::ReportNodeReq;
 use detee_shared::vm_proto::brain_vm_cli_client::BrainVmCliClient;
 use detee_shared::vm_proto::ListVmContractsReq;
+use detee_shared::{
+    common_proto::Pubkey, general_proto::brain_general_cli_client::BrainGeneralCliClient,
+};
 use futures::StreamExt;
 use surreal_brain::constants::VM_NODE;
-use surreal_brain::db::vm::VmNodeWithReports;
+use surreal_brain::db::VmNodeWithReports;
 
 mod common;
 

tests/grpc_vm_daemon_test.rs

@@ -1,8 +1,8 @@
-use common::prepare_test_env::{
-    prepare_test_db, run_service_for_stream, run_service_in_background,
+use common::{
+    prepare_test_env::{prepare_test_db, run_service_for_stream, run_service_in_background},
+    test_utils::Key,
+    vm_daemon_utils::{mock_vm_daemon, register_vm_node},
 };
-use common::test_utils::Key;
-use common::vm_daemon_utils::{mock_vm_daemon, register_vm_node};
 use detee_shared::vm_proto;
 use detee_shared::vm_proto::brain_vm_cli_client::BrainVmCliClient;
 use detee_shared::vm_proto::brain_vm_daemon_client::BrainVmDaemonClient;