test listing operators and list vm contracts

test report node

Cargo.lock

@@ -3771,6 +3771,7 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
 name = "surreal-brain"
 version = "0.1.0"
 dependencies = [
+ "anyhow",
  "bs58",
  "chrono",
  "dashmap 6.1.0",

Cargo.toml

@@ -27,3 +27,8 @@ strip = true
 opt-level = 3
 panic = 'abort'
 codegen-units = 1
+
+[dev-dependencies]
+anyhow = "1.0.98"
+bs58 = "0.5.1"
+ed25519-dalek = "2.1.1"

interim_tables.surql

@@ -1,10 +1,11 @@
 DEFINE TABLE account SCHEMAFULL;
-DEFINE FIELD balance ON TABLE account TYPE int;
+DEFINE FIELD balance ON TABLE account TYPE int DEFAULT 0;
-DEFINE FIELD tmp_locked ON TABLE account TYPE int;
+DEFINE FIELD tmp_locked ON TABLE account TYPE int DEFAULT 0;
-DEFINE FIELD escrow ON TABLE account TYPE int;
+DEFINE FIELD escrow ON TABLE account TYPE int DEFAULT 0;
-DEFINE FIELD email ON TABLE account TYPE string;
+DEFINE FIELD email ON TABLE account TYPE string DEFAULT "";
 
 DEFINE TABLE vm_node SCHEMAFULL;
+DEFINE FIELD operator ON TABLE vm_node TYPE record<account>;
 DEFINE FIELD country ON TABLE vm_node TYPE string;
 DEFINE FIELD region ON TABLE vm_node TYPE string;
 DEFINE FIELD city ON TABLE vm_node TYPE string;
@@ -37,6 +38,7 @@ DEFINE FIELD locked_nano ON TABLE vm_contract TYPE int;
 DEFINE FIELD collected_at ON TABLE vm_contract TYPE datetime;
 
 DEFINE TABLE app_node SCHEMAFULL;
+DEFINE FIELD operator ON TABLE app_node TYPE record<account>;
 DEFINE FIELD country ON TABLE app_node TYPE string;
 DEFINE FIELD region ON TABLE app_node TYPE string;
 DEFINE FIELD city ON TABLE app_node TYPE string;
@@ -75,7 +77,5 @@ DEFINE FIELD reason ON TABLE kick TYPE string;
 DEFINE FIELD contract ON TABLE kick TYPE record<vm_contract|app_contract>;
 
 DEFINE TABLE report TYPE RELATION FROM account TO vm_node|app_node;
-DEFINE FIELD created_at ON TABLE ban TYPE datetime;
+DEFINE FIELD created_at ON TABLE report TYPE datetime;
-DEFINE FIELD reason ON TABLE ban TYPE string;
+DEFINE FIELD reason ON TABLE report TYPE string;
-
-DEFINE TABLE operator TYPE RELATION FROM account TO vm_node|app_node;

src/bin/brain.rs

@@ -1,21 +1,24 @@
 use detee_shared::general_proto::brain_general_cli_server::BrainGeneralCliServer;
 use detee_shared::vm_proto::brain_vm_cli_server::BrainVmCliServer;
-use surreal_brain::grpc::BrainGeneralCliMock;
+use surreal_brain::constants::{
-use surreal_brain::grpc::BrainVmCliMock;
+    BRAIN_GRPC_ADDR, CERT_KEY_PATH, CERT_PATH, DB_ADDRESS, DB_NAME, DB_NS,
+};
 use surreal_brain::db;
+use surreal_brain::grpc::BrainGeneralCliForReal;
+use surreal_brain::grpc::BrainVmCliForReal;
 use tonic::transport::{Identity, Server, ServerTlsConfig};
 
 #[tokio::main]
 async fn main() {
     env_logger::builder().filter_level(log::LevelFilter::Debug).init();
-    db::init().await.unwrap();
+    db::init(DB_ADDRESS, DB_NS, DB_NAME).await.unwrap();
-    let addr = "0.0.0.0:31337".parse().unwrap();
+    let addr = BRAIN_GRPC_ADDR.parse().unwrap();
 
-    let snp_cli_server = BrainVmCliServer::new(BrainVmCliMock {});
+    let snp_cli_server = BrainVmCliServer::new(BrainVmCliForReal {});
-    let general_service_server = BrainGeneralCliServer::new(BrainGeneralCliMock {});
+    let general_service_server = BrainGeneralCliServer::new(BrainGeneralCliForReal {});
 
-    let cert = std::fs::read_to_string("./tmp/brain-crt.pem").unwrap();
+    let cert = std::fs::read_to_string(CERT_PATH).unwrap();
-    let key = std::fs::read_to_string("./tmp/brain-key.pem").unwrap();
+    let key = std::fs::read_to_string(CERT_KEY_PATH).unwrap();
 
     let identity = Identity::from_pem(cert, key);
 

src/bin/migration0.rs

@@ -1,6 +1,8 @@
 // After deleting this migration, also delete old_brain structs
 // and dangling impls from the model
 use std::error::Error;
+use surreal_brain::constants::{DB_ADDRESS, DB_NAME, DB_NS};
+use surreal_brain::db::init;
 use surreal_brain::{db, old_brain};
 
 #[tokio::main]
@@ -8,6 +10,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
     let old_brain_data = old_brain::BrainData::load_from_disk()?;
     // println!("{}", serde_yaml::to_string(&old_brain_data)?);
 
+    init(DB_ADDRESS, DB_NS, DB_NAME).await?;
+
     let result = db::migration0(&old_brain_data).await?;
 
     println!("{result:?}");

src/constants.rs

@@ -0,0 +1,24 @@
+pub const BRAIN_GRPC_ADDR: &str = "0.0.0.0:31337";
+pub const CERT_PATH: &str = "./tmp/brain-crt.pem";
+pub const CERT_KEY_PATH: &str = "./tmp/brain-key.pem";
+
+pub const DB_ADDRESS: &str = "localhost:8000";
+pub const DB_NS: &str = "brain";
+pub const DB_NAME: &str = "migration";
+
+// TODO: read from .env
+pub const DB_USER: &str = "root";
+pub const DB_PASS: &str = "root";
+
+pub const ACCOUNT: &str = "account";
+pub const OPERATOR: &str = "operator";
+pub const VM_CONTRACT: &str = "vm_contract";
+pub const VM_NODE: &str = "vm_node";
+
+pub const ADMIN_ACCOUNTS: &[&str] = &[
+    "x52w7jARC5erhWWK65VZmjdGXzBK6ZDgfv1A283d8XK",
+    "FHuecMbeC1PfjkW2JKyoicJAuiU7khgQT16QUB3Q1XdL",
+    "H21Shi4iE7vgfjWEQNvzmpmBMJSaiZ17PYUcdNoAoKNc",
+];
+
+pub const OLD_BRAIN_DATA_PATH: &str = "./saved_data.yaml";

src/db.rs

@@ -1,3 +1,4 @@
+use crate::constants::{ACCOUNT, DB_PASS, DB_USER, VM_CONTRACT, VM_NODE};
 use crate::old_brain;
 use serde::{Deserialize, Serialize};
 use std::sync::LazyLock;
@@ -8,11 +9,7 @@ use surrealdb::{
     RecordId, Surreal,
 };
 
-static DB: LazyLock<Surreal<Client>> = LazyLock::new(Surreal::init);
+pub static DB: LazyLock<Surreal<Client>> = LazyLock::new(Surreal::init);
-const ACCOUNT: &str = "account";
-const OPERATOR: &str = "operator";
-const VM_CONTRACT: &str = "vm_contract";
-const VM_NODE: &str = "vm_node";
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
@@ -20,11 +17,11 @@ pub enum Error {
     DataBase(#[from] surrealdb::Error),
 }
 
-pub async fn init() -> surrealdb::Result<()> {
+pub async fn init(db_address: &str, ns: &str, db: &str) -> surrealdb::Result<()> {
-    DB.connect::<Ws>("localhost:8000").await?;
+    DB.connect::<Ws>(db_address).await?;
     // Sign in to the server
-    DB.signin(Root { username: "root", password: "root" }).await?;
+    DB.signin(Root { username: DB_USER, password: DB_PASS }).await?;
-    DB.use_ns("brain").use_db("migration").await?;
+    DB.use_ns(ns).use_db(db).await?;
     Ok(())
 }
 
@@ -33,9 +30,6 @@ pub async fn migration0(old_data: &old_brain::BrainData) -> surrealdb::Result<()
     let vm_nodes: Vec<VmNode> = old_data.into();
     let app_nodes: Vec<AppNode> = old_data.into();
     let vm_contracts: Vec<VmContract> = old_data.into();
-    let operators: Vec<OperatorRelation> = old_data.into();
-
-    init().await?;
 
     println!("Inserting accounts...");
     let _: Vec<Account> = DB.insert(()).content(accounts).await?;
@@ -45,8 +39,6 @@ pub async fn migration0(old_data: &old_brain::BrainData) -> surrealdb::Result<()
     let _: Vec<AppNode> = DB.insert(()).content(app_nodes).await?;
     println!("Inserting vm contracts...");
     let _: Vec<VmContract> = DB.insert("vm_contract").relation(vm_contracts).await?;
-    println!("Inserting operators...");
-    let _: Vec<OperatorRelation> = DB.insert(OPERATOR).relation(operators).await?;
 
     Ok(())
 }
@@ -85,6 +77,7 @@ impl Account {
 #[derive(Debug, Serialize, Deserialize)]
 pub struct VmNode {
     pub id: RecordId,
+    pub operator: RecordId,
     pub country: String,
     pub region: String,
     pub city: String,
@@ -100,6 +93,33 @@ pub struct VmNode {
     pub offline_minutes: u64,
 }
 
+impl VmNode {
+    pub async fn register(self) -> Result<(), Error> {
+        let _: Option<VmNode> = DB.upsert(self.id.clone()).content(self).await?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct VmNodeWithReports {
+    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_ipv4: u32,
+    pub avail_ipv6: u32,
+    pub avail_ports: u32,
+    pub max_ports_per_vm: u32,
+    pub price: u64,
+    pub offline_minutes: u64,
+    pub reports: Vec<Report>,
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct VmContract {
     pub id: RecordId,
@@ -142,6 +162,8 @@ impl VmContract {
     }
 }
 
+impl VmContract {}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct VmContractWithNode {
     pub id: RecordId,
@@ -181,6 +203,14 @@ impl VmContractWithNode {
         Ok(contracts)
     }
 
+    pub async fn list_by_node(admin: &str) -> Result<Vec<Self>, Error> {
+        let mut result = DB
+            .query(format!("select * from {VM_CONTRACT} where out = {VM_NODE}:{admin} fetch out;"))
+            .await?;
+        let contracts: Vec<Self> = result.take(0)?;
+        Ok(contracts)
+    }
+
     pub async fn list_by_operator(operator: &str) -> Result<Vec<Self>, Error> {
         let mut result = DB
             .query(format!(
@@ -221,18 +251,37 @@ impl VmContractWithNode {
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct AppNode {
-    id: RecordId,
+    pub id: RecordId,
-    country: String,
+    pub operator: RecordId,
-    region: String,
+    pub country: String,
-    city: String,
+    pub region: String,
-    ip: String,
+    pub city: String,
-    avail_mem_mb: u32,
+    pub ip: String,
-    avail_vcpus: u32,
+    pub avail_mem_mb: u32,
-    avail_storage_gbs: u32,
+    pub avail_vcpus: u32,
-    avail_ports: u32,
+    pub avail_storage_gbs: u32,
-    max_ports_per_app: u32,
+    pub avail_ports: u32,
-    price: u64,
+    pub max_ports_per_app: u32,
-    offline_minutes: u64,
+    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)]
@@ -288,12 +337,16 @@ pub struct Report {
     #[serde(rename = "out")]
     to_node: RecordId,
     created_at: Datetime,
-    reason: String,
+    pub reason: String,
 }
 
 impl Report {
     // TODO: test this functionality and remove this comment
-    pub async fn create(from_account: RecordId, to_node: RecordId, reason: String) -> Result<(), Error> {
+    pub async fn create(
+        from_account: RecordId,
+        to_node: RecordId,
+        reason: String,
+    ) -> Result<(), Error> {
         let _: Vec<Self> = DB
             .insert("report")
             .relation(Report { from_account, to_node, created_at: Datetime::default(), reason })
@@ -302,23 +355,6 @@ impl Report {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
-pub struct OperatorRelation {
-    #[serde(rename = "in")]
-    pub account: RecordId,
-    #[serde(rename = "out")]
-    pub node: RecordId,
-}
-
-impl OperatorRelation {
-    fn new(account: &str, vm_node: &str) -> Self {
-        Self {
-            account: RecordId::from(("account", account.to_string())),
-            node: RecordId::from(("vm_node", vm_node.to_string())),
-        }
-    }
-}
-
 /// This is the operator obtained from the DB,
 /// however the relation is defined using OperatorRelation
 #[derive(Debug, Serialize, Deserialize)]
@@ -335,21 +371,62 @@ impl Operator {
     pub async fn list() -> Result<Vec<Self>, Error> {
         let mut result = DB
             .query(format!(
-                "select *,
+                "array::distinct(array::flatten( [
-                    in as account,
+                    (select operator from vm_node group by operator).operator,
-                    <-account.email[0] as email,
+                    (select operator from app_node group by operator).operator
-                    <-account.escrow[0] as escrow,
+                ]));"
-                    count(->vm_node) as vm_nodes, 
-                    count(->app_node) as app_nodes,
-                    (select in from <-account->operator->vm_node<-report).len() +
-                        (select in from <-account->operator->app_node<-report).len()
-                    as reports
-                from operator group by in;"
             ))
             .await?;
-        let operators: Vec<Self> = result.take(0)?;
+        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(&account.key().to_string()).await? {
+                operators.push(operator);
+            }
+        }
         Ok(operators)
     }
+
+    pub async fn inspect(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(
+        account: &str,
+    ) -> Result<(Option<Self>, Vec<VmNodeWithReports>, Vec<AppNodeWithReports>), Error> {
+        let operator = Self::inspect(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
@@ -359,7 +436,8 @@ impl From<&old_brain::BrainData> for Vec<VmNode> {
         let mut nodes = Vec::new();
         for old_node in old_data.vm_nodes.iter() {
             nodes.push(VmNode {
-                id: RecordId::from(("vm_node", old_node.public_key.clone())),
+                id: RecordId::from((VM_NODE, old_node.public_key.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(),
@@ -388,7 +466,11 @@ impl From<&old_brain::BrainData> for Vec<VmContract> {
                 mapped_ports.push((*port, 8080 as u32));
             }
             contracts.push(VmContract {
-                id: RecordId::from((VM_CONTRACT, old_c.uuid.replace("-", ""))),
+                id: RecordId::from((
+                    VM_CONTRACT,
+                    old_c.node_pubkey.chars().take(20).collect::<String>()
+                        + &old_c.uuid.replace("-", "").chars().take(20).collect::<String>(),
+                )),
                 admin: RecordId::from((ACCOUNT, old_c.admin_pubkey.clone())),
                 vm_node: RecordId::from((VM_NODE, old_c.node_pubkey.clone())),
                 state: "active".to_string(),
@@ -418,6 +500,7 @@ impl From<&old_brain::BrainData> for Vec<AppNode> {
         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(),
@@ -455,18 +538,3 @@ impl From<&old_brain::BrainData> for Vec<Account> {
         accounts
     }
 }
-
-impl From<&old_brain::BrainData> for Vec<OperatorRelation> {
-    fn from(old_data: &old_brain::BrainData) -> Self {
-        let mut operator_entries = Vec::new();
-        for operator in old_data.operators.clone() {
-            for vm_node in operator.1.vm_nodes.iter() {
-                operator_entries.push(OperatorRelation::new(&operator.0, vm_node));
-            }
-            for app_node in operator.1.app_nodes.iter() {
-                operator_entries.push(OperatorRelation::new(&operator.0, app_node));
-            }
-        }
-        operator_entries
-    }
-}

src/grpc.rs

@@ -1,6 +1,7 @@
 #![allow(dead_code)]
+use crate::constants::{ACCOUNT, ADMIN_ACCOUNTS, VM_NODE};
 use crate::db;
-use detee_shared::app_proto::AppContract;
+use detee_shared::app_proto::{AppContract, AppNodeListResp};
 use detee_shared::{
     common_proto::{Empty, Pubkey},
     general_proto::{
@@ -8,7 +9,10 @@ use detee_shared::{
         InspectOperatorResp, KickReq, KickResp, ListOperatorsResp, RegOperatorReq, ReportNodeReq,
         SlashReq,
     },
-    vm_proto::{brain_vm_cli_server::BrainVmCli, ListVmContractsReq, *},
+    vm_proto::{
+        brain_vm_cli_server::BrainVmCli, brain_vm_daemon_server::BrainVmDaemon, ListVmContractsReq,
+        *,
+    },
 };
 
 use log::info;
@@ -18,9 +22,9 @@ use tokio_stream::wrappers::ReceiverStream;
 // use tokio::sync::mpsc;
 // use tokio_stream::{wrappers::ReceiverStream, Stream};
 use tokio_stream::Stream;
-use tonic::{Request, Response, Status};
+use tonic::{Request, Response, Status, Streaming};
 
-pub struct BrainGeneralCliMock {}
+pub struct BrainGeneralCliForReal {}
 
 impl From<db::Account> for AccountBalance {
     fn from(account: db::Account) -> Self {
@@ -84,8 +88,172 @@ impl From<db::Operator> for ListOperatorsResp {
     }
 }
 
+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,
+        }
+    }
+}
+
+struct BrainVmDaemonForReal {}
+
 #[tonic::async_trait]
-impl BrainGeneralCli for BrainGeneralCliMock {
+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()
+        .await?;
+
+        info!("Sending existing contracts to {}", req.node_pubkey);
+        let contracts = db::VmContractWithNode::list_by_node(&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> {
+        todo!();
+        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);
+        // A surreql query that listens live for changes:
+        // live select * from vm_contract where meta::id(id) in "3zRxiGRnf46vd3zAEmpa".."3zRxiGRnf46vd3zAEmpb";
+        //
+        // And the same from Rust:
+        //     let mut stream = db
+        //         .select("vm_contract")
+        //         .range("3zRxiGRnf46vd3zAEmpa".."3zRxiGRnf46vd3zAEmpb")
+        //         .live()
+        //         .await?;
+        //
+        //     fn handle(result: Result<Notification<VmContract>, surrealdb::Error>) {
+        //         println!("Received notification: {:?}", result);
+        //     }
+        //
+        //     while let Some(result) = stream.next().await {
+        //         handle(result);
+        //     }
+        //
+
+        // let (tx, rx) = mpsc::channel(6);
+        //self.data.add_daemon_tx(&pubkey, tx);
+        //let output_stream = ReceiverStream::new(rx).map(|msg| Ok(msg));
+        //Ok(Response::new(Box::pin(output_stream) as Self::BrainMessagesStream))
+    }
+
+    async fn daemon_messages(
+        &self,
+        _req: Request<Streaming<VmDaemonMessage>>,
+    ) -> Result<Response<Empty>, Status> {
+        todo!();
+        // 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"));
+        // }
+
+        // // info!("Received a message from daemon {pubkey}: {daemon_message:?}");
+        // 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)) => {
+        //                 self.data.submit_newvm_resp(new_vm_resp).await;
+        //             }
+        //             Some(vm_daemon_message::Msg::UpdateVmResp(update_vm_resp)) => {
+        //                 self.data.submit_updatevm_resp(update_vm_resp).await;
+        //             }
+        //             Some(vm_daemon_message::Msg::VmNodeResources(node_resources)) => {
+        //                 self.data.submit_node_resources(node_resources);
+        //             }
+        //             _ => {}
+        //         },
+        //         Err(e) => {
+        //             log::warn!("Daemon disconnected: {e:?}");
+        //             self.data.del_daemon_tx(&pubkey);
+        //         }
+        //     }
+        // }
+        // 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>>;
@@ -134,13 +302,16 @@ impl BrainGeneralCli for BrainGeneralCliMock {
 
     async fn inspect_operator(
         &self,
-        _req: Request<Pubkey>,
+        req: Request<Pubkey>,
     ) -> Result<Response<InspectOperatorResp>, Status> {
-        todo!();
+        match db::Operator::inspect_nodes(&req.into_inner().pubkey).await? {
-        // match self.data.inspect_operator(&req.into_inner().pubkey) {
+            (Some(op), vm_nodes, app_nodes) => Ok(Response::new(InspectOperatorResp {
-        //     Some(op) => Ok(Response::new(op.into())),
+                operator: Some(op.into()),
-        //     None => Err(Status::not_found("The wallet you specified is not an operator")),
+                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(
@@ -244,10 +415,10 @@ impl BrainGeneralCli for BrainGeneralCliMock {
     }
 }
 
-pub struct BrainVmCliMock {}
+pub struct BrainVmCliForReal {}
 
 #[tonic::async_trait]
-impl BrainVmCli for BrainVmCliMock {
+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>>;
 
@@ -499,11 +670,45 @@ fn check_sig_from_req<T: std::fmt::Debug + PubkeyGetter>(req: Request<T>) -> Res
     Ok(req)
 }
 
-const ADMIN_ACCOUNTS: &[&str] = &[
+fn check_sig_from_parts(pubkey: &str, time: &str, msg: &str, sig: &str) -> Result<(), Status> {
-    "x52w7jARC5erhWWK65VZmjdGXzBK6ZDgfv1A283d8XK",
+    let now = chrono::Utc::now();
-    "FHuecMbeC1PfjkW2JKyoicJAuiU7khgQT16QUB3Q1XdL",
+    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
-    "H21Shi4iE7vgfjWEQNvzmpmBMJSaiZ17PYUcdNoAoKNc",
+        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
-];
+    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
+    if seconds_elapsed > 4 || seconds_elapsed < -4 {
+        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") {

src/lib.rs

@@ -1,3 +1,4 @@
-pub mod grpc;
+pub mod constants;
 pub mod db;
+pub mod grpc;
 pub mod old_brain;

src/old_brain.rs

@@ -5,6 +5,8 @@ use dashmap::DashMap;
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
 
+use crate::constants::OLD_BRAIN_DATA_PATH;
+
 #[derive(Clone, Default, Serialize, Deserialize, Debug)]
 pub struct AccountData {
     pub balance: u64,
@@ -124,7 +126,7 @@ pub struct BrainData {
 
 impl BrainData {
     pub fn load_from_disk() -> Result<Self, Box<dyn std::error::Error>> {
-        let content = std::fs::read_to_string("./saved_data.yaml")?;
+        let content = std::fs::read_to_string(OLD_BRAIN_DATA_PATH)?;
         let data: Self = serde_yaml::from_str(&content)?;
         Ok(data)
     }

tests/common/mod.rs

@@ -0,0 +1,2 @@
+pub mod prepare_test_env;
+pub mod test_utils;

tests/common/prepare_test_env.rs

@@ -0,0 +1,54 @@
+use detee_shared::{
+    general_proto::brain_general_cli_server::BrainGeneralCliServer,
+    vm_proto::brain_vm_cli_server::BrainVmCliServer,
+};
+use surreal_brain::grpc::{BrainGeneralCliMock, BrainVmCliMock};
+use tokio::sync::OnceCell;
+use tonic::transport::Channel;
+
+pub const DB_URL: &str = "localhost:8000";
+pub const DB_NS: &str = "test_brain";
+pub const DB_NAME: &str = "test_migration_db";
+
+pub const GRPC_ADDR: &str = "127.0.0.1:31337";
+
+pub static TEST_STATE: OnceCell<Channel> = OnceCell::const_new();
+
+pub async fn prepare_test_db() {
+    surreal_brain::db::init(DB_URL, DB_NS, DB_NAME)
+        .await
+        .expect("Failed to initialize the database");
+
+    let old_brain_data = surreal_brain::old_brain::BrainData::load_from_disk().unwrap();
+    let _ = surreal_brain::db::DB.query(format!("REMOVE DATABASE {DB_NAME}")).await;
+    surreal_brain::db::migration0(&old_brain_data).await.unwrap();
+}
+
+pub async fn fake_grpc_server() {
+    tonic::transport::Server::builder()
+        .add_service(BrainGeneralCliServer::new(BrainGeneralCliMock {}))
+        .add_service(BrainVmCliServer::new(BrainVmCliMock {}))
+        .serve(GRPC_ADDR.parse().unwrap())
+        .await
+        .unwrap();
+}
+
+pub async fn fake_grpc_client() -> Channel {
+    let url = format!("http://{GRPC_ADDR}");
+    Channel::from_shared(url).unwrap().connect().await.unwrap()
+}
+
+pub async fn prepare_test_setup() {
+    TEST_STATE
+        .get_or_init(|| async {
+            prepare_test_db().await;
+
+            tokio::spawn(async {
+                fake_grpc_server().await;
+            });
+
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            fake_grpc_client().await
+        })
+        .await;
+}

tests/common/test_utils.rs

@@ -0,0 +1,41 @@
+use anyhow::Result;
+use ed25519_dalek::Signer;
+use ed25519_dalek::SigningKey;
+use tonic::metadata::AsciiMetadataValue;
+use tonic::Request;
+
+pub const WALLET_KEY_PATH: &str = "tests/fixtures/secret_detee_wallet_key";
+
+pub fn sign_request<T: std::fmt::Debug>(req: T) -> Result<Request<T>> {
+    let pubkey = get_pub_key()?;
+    let timestamp = chrono::Utc::now().to_rfc3339();
+    let signature = try_sign_message(&format!("{timestamp}{req:?}"))?;
+    let timestamp: AsciiMetadataValue = timestamp.parse()?;
+    let pubkey: AsciiMetadataValue = pubkey.parse()?;
+    let signature: AsciiMetadataValue = signature.parse()?;
+    let mut req = Request::new(req);
+    req.metadata_mut().insert("timestamp", timestamp);
+    req.metadata_mut().insert("pubkey", pubkey);
+    req.metadata_mut().insert("request-signature", signature);
+
+    Ok(req)
+}
+
+fn get_signing_key() -> Result<SigningKey> {
+    let key = bs58::decode(std::fs::read_to_string(WALLET_KEY_PATH)?.trim())
+        .into_vec()?
+        .try_into()
+        .map_err(|e: Vec<u8>| anyhow::anyhow!("Invalid key length: {}", e.len()))?;
+    let key = SigningKey::from_bytes(&key);
+    Ok(key)
+}
+
+pub fn get_pub_key() -> Result<String> {
+    let key = get_signing_key()?;
+    Ok(bs58::encode(key.verifying_key().to_bytes()).into_string())
+}
+
+pub fn try_sign_message(message: &str) -> Result<String> {
+    let key = get_signing_key()?;
+    Ok(bs58::encode(key.sign(message.as_bytes()).to_bytes()).into_string())
+}

tests/fixtures/secret_detee_wallet_key

@@ -0,0 +1 @@
+9RBoFzqSfMVjQmmCbnMhfNGxGEdRmTyb9eF4wDdRVX6f

tests/grpcs_test.rs

@@ -0,0 +1,123 @@
+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,
+};
+mod common;
+use common::prepare_test_env::{prepare_test_setup, TEST_STATE};
+use common::test_utils::{get_pub_key, sign_request};
+use tokio_stream::StreamExt;
+
+#[tokio::test]
+async fn test_general_balance() {
+    prepare_test_setup().await;
+    let grpc_channel = TEST_STATE.get().unwrap().clone();
+
+    let mut brain_general_cli_client = BrainGeneralCliClient::new(grpc_channel.clone());
+
+    let req_data = Pubkey { pubkey: get_pub_key().unwrap() };
+
+    let acc_bal = brain_general_cli_client
+        .get_balance(sign_request(req_data).unwrap())
+        .await
+        .unwrap()
+        .into_inner();
+
+    // verify it in db also
+
+    assert_eq!(acc_bal.balance, 0);
+    assert_eq!(acc_bal.tmp_locked, 0);
+}
+
+#[tokio::test]
+async fn test_report_node() {
+    prepare_test_setup().await;
+    let grpc_channel = TEST_STATE.get().unwrap().clone();
+
+    let mut brain_general_cli_client = BrainGeneralCliClient::new(grpc_channel.clone());
+
+    // TODO: create contract, node and operator in db and use it here
+    let req_data = ReportNodeReq {
+        admin_pubkey: get_pub_key().unwrap(),
+        node_pubkey: String::from("node_pubkey"),
+        contract: String::from("uuid"),
+        reason: String::from("reason"),
+    };
+
+    let report_error =
+        brain_general_cli_client.report_node(sign_request(req_data).unwrap()).await.err().unwrap();
+
+    assert_eq!(report_error.message(), "No contract found by this ID.");
+
+    // verify report in db
+}
+
+#[tokio::test]
+// TODO: register some operators before testing this
+async fn test_list_operators() {
+    prepare_test_setup().await;
+    let grpc_channel = TEST_STATE.get().unwrap().clone();
+
+    let mut brain_general_cli_client = BrainGeneralCliClient::new(grpc_channel.clone());
+
+    let mut grpc_stream = brain_general_cli_client
+        .list_operators(sign_request(Empty {}).unwrap())
+        .await
+        .unwrap()
+        .into_inner();
+
+    let mut operators = Vec::new();
+    while let Some(stream_update) = grpc_stream.next().await {
+        match stream_update {
+            Ok(op) => {
+                operators.push(op);
+            }
+            Err(e) => {
+                panic!("Received error instead of operators: {e:?}");
+            }
+        }
+    }
+
+    assert!(!operators.is_empty())
+
+    // verify report in db
+}
+
+#[tokio::test]
+// TODO: create vm for this user before testing this
+async fn test_list_vm_contracts() {
+    prepare_test_setup().await;
+    let grpc_channel = TEST_STATE.get().unwrap().clone();
+
+    let mut brain_general_cli_client = BrainVmCliClient::new(grpc_channel.clone());
+
+    let req_data = ListVmContractsReq {
+        wallet: get_pub_key().unwrap(),
+        uuid: String::from("uuid"),
+        as_operator: false,
+    };
+
+    let mut grpc_stream = brain_general_cli_client
+        .list_vm_contracts(sign_request(req_data).unwrap())
+        .await
+        .unwrap()
+        .into_inner();
+
+    let mut vm_contracts = Vec::new();
+    while let Some(stream_update) = grpc_stream.next().await {
+        match stream_update {
+            Ok(vm_c) => {
+                vm_contracts.push(vm_c);
+            }
+            Err(e) => {
+                panic!("Received error instead of vm_contracts: {e:?}");
+            }
+        }
+    }
+
+    assert!(vm_contracts.is_empty())
+
+    // verify report in db
+}