added capability to search nodes

compiles but not tested yet

.gitignore

@@ -1,2 +1,3 @@
 /target
+secrets
 tmp

Cargo.lock

@@ -1000,7 +1000,7 @@ dependencies = [
 [[package]]
 name = "detee-shared"
 version = "0.1.0"
-source = "git+ssh://git@gitea.detee.cloud/testnet/proto?branch=surreal_brain#fb38352e1b47837b14f32d8df5ae7f6b17202aae"
+source = "git+ssh://git@gitea.detee.cloud/testnet/proto?branch=surreal_brain#d0d4622c52efdf74ed6582fbac23a6159986ade3"
 dependencies = [
  "bincode 2.0.1",
  "prost",
@@ -3779,6 +3779,7 @@ dependencies = [
  "env_logger",
  "futures",
  "log",
+ "nanoid",
  "serde",
  "serde_json",
  "serde_yaml",

Cargo.toml

@@ -20,6 +20,7 @@ tokio-stream = "0.1.17"
 log = "0.4.27"
 env_logger = "0.11.8"
 thiserror = "2.0.12"
+nanoid = "0.4.0"
 
 [profile.release]
 lto = true

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;
@@ -19,24 +20,64 @@ DEFINE FIELD max_ports_per_vm ON TABLE vm_node TYPE int;
 DEFINE FIELD price ON TABLE vm_node TYPE int;
 DEFINE FIELD offline_minutes ON TABLE vm_node TYPE int;
 
-DEFINE TABLE vm_contract TYPE RELATION FROM account TO vm_node SCHEMAFULL;
+DEFINE TABLE new_vm_req TYPE RELATION FROM account TO vm_node SCHEMAFULL;
-DEFINE FIELD state ON TABLE vm_contract TYPE string;
+DEFINE FIELD hostname ON TABLE new_vm_req TYPE string;
-DEFINE FIELD hostname ON TABLE vm_contract TYPE string;
+DEFINE FIELD extra_ports ON TABLE new_vm_req TYPE array<int>;
-DEFINE FIELD mapped_ports ON TABLE vm_contract TYPE array<[int, int]>;
+DEFINE FIELD public_ipv4 ON TABLE new_vm_req TYPE bool;
-DEFINE FIELD public_ipv4 ON TABLE vm_contract TYPE string;
+DEFINE FIELD public_ipv6 ON TABLE new_vm_req TYPE bool;
-DEFINE FIELD public_ipv6 ON TABLE vm_contract TYPE string;
+DEFINE FIELD disk_size_gb ON TABLE new_vm_req TYPE int;
-DEFINE FIELD disk_size_gb ON TABLE vm_contract TYPE int;
+DEFINE FIELD vcpus ON TABLE new_vm_req TYPE int;
-DEFINE FIELD vcpus ON TABLE vm_contract TYPE int;
+DEFINE FIELD memory_mb ON TABLE new_vm_req TYPE int;
-DEFINE FIELD memory_mb ON TABLE vm_contract TYPE int;
+DEFINE FIELD dtrfs_sha ON TABLE new_vm_req TYPE string;
-DEFINE FIELD dtrfs_sha ON TABLE vm_contract TYPE string;
+DEFINE FIELD dtrfs_url ON TABLE new_vm_req TYPE string;
-DEFINE FIELD kernel_sha ON TABLE vm_contract TYPE string;
+DEFINE FIELD kernel_sha ON TABLE new_vm_req TYPE string;
-DEFINE FIELD created_at ON TABLE vm_contract TYPE datetime;
+DEFINE FIELD kernel_url ON TABLE new_vm_req TYPE string;
-DEFINE FIELD updated_at ON TABLE vm_contract TYPE datetime;
+DEFINE FIELD created_at ON TABLE new_vm_req TYPE datetime;
-DEFINE FIELD price_per_unit ON TABLE vm_contract TYPE int;
+DEFINE FIELD updated_at ON TABLE new_vm_req TYPE datetime;
-DEFINE FIELD locked_nano ON TABLE vm_contract TYPE int;
+DEFINE FIELD price_per_unit ON TABLE new_vm_req TYPE int;
-DEFINE FIELD collected_at ON TABLE vm_contract TYPE datetime;
+DEFINE FIELD locked_nano ON TABLE new_vm_req TYPE int;
+DEFINE FIELD error ON TABLE new_vm_req TYPE string;
+
+DEFINE TABLE active_vm TYPE RELATION FROM account TO vm_node SCHEMAFULL;
+DEFINE FIELD hostname ON TABLE active_vm TYPE string;
+DEFINE FIELD mapped_ports ON TABLE active_vm TYPE array<[int, int]>;
+DEFINE FIELD public_ipv4 ON TABLE active_vm TYPE string;
+DEFINE FIELD public_ipv6 ON TABLE active_vm TYPE string;
+DEFINE FIELD disk_size_gb ON TABLE active_vm TYPE int;
+DEFINE FIELD vcpus ON TABLE active_vm TYPE int;
+DEFINE FIELD memory_mb ON TABLE active_vm TYPE int;
+DEFINE FIELD dtrfs_sha ON TABLE active_vm TYPE string;
+DEFINE FIELD kernel_sha ON TABLE active_vm TYPE string;
+DEFINE FIELD created_at ON TABLE active_vm TYPE datetime;
+DEFINE FIELD price_per_unit ON TABLE active_vm TYPE int;
+DEFINE FIELD locked_nano ON TABLE active_vm TYPE int;
+DEFINE FIELD collected_at ON TABLE active_vm TYPE datetime;
+
+DEFINE TABLE update_vm_req TYPE RELATION FROM account TO vm_node SCHEMAFULL;
+DEFINE FIELD vcpus ON TABLE update_vm_req TYPE int;
+DEFINE FIELD memory_mb ON TABLE update_vm_req TYPE int;
+DEFINE FIELD disk_size_gb ON TABLE update_vm_req TYPE int;
+DEFINE FIELD dtrfs_sha ON TABLE update_vm_req TYPE string;
+DEFINE FIELD dtrfs_url ON TABLE update_vm_req TYPE string;
+DEFINE FIELD kernel_sha ON TABLE update_vm_req TYPE string;
+DEFINE FIELD kernel_url ON TABLE update_vm_req TYPE string;
+
+DEFINE TABLE deleted_vm TYPE RELATION FROM account TO vm_node SCHEMAFULL;
+DEFINE FIELD hostname ON TABLE deleted_vm TYPE string;
+DEFINE FIELD mapped_ports ON TABLE deleted_vm TYPE array<[int, int]>;
+DEFINE FIELD public_ipv4 ON TABLE deleted_vm TYPE string;
+DEFINE FIELD public_ipv6 ON TABLE deleted_vm TYPE string;
+DEFINE FIELD disk_size_gb ON TABLE deleted_vm TYPE int;
+DEFINE FIELD vcpus ON TABLE deleted_vm TYPE int;
+DEFINE FIELD memory_mb ON TABLE deleted_vm TYPE int;
+DEFINE FIELD dtrfs_sha ON TABLE deleted_vm TYPE string;
+DEFINE FIELD kernel_sha ON TABLE deleted_vm TYPE string;
+DEFINE FIELD created_at ON TABLE deleted_vm TYPE datetime;
+DEFINE FIELD deleted_at ON TABLE deleted_vm TYPE datetime;
+DEFINE FIELD price_per_unit ON TABLE deleted_vm TYPE int;
 
 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;
@@ -49,22 +90,36 @@ DEFINE FIELD max_ports_per_app ON TABLE app_node TYPE int;
 DEFINE FIELD price ON TABLE app_node TYPE int;
 DEFINE FIELD offline_minutes ON TABLE app_node TYPE int;
 
-DEFINE TABLE app_contract TYPE RELATION FROM account TO app_node SCHEMAFULL;
+DEFINE TABLE active_app TYPE RELATION FROM account TO app_node SCHEMAFULL;
-DEFINE FIELD state ON TABLE app_contract TYPE string;
+DEFINE FIELD app_name ON TABLE active_app TYPE string;
-DEFINE FIELD app_name ON TABLE app_contract TYPE string;
+DEFINE FIELD mapped_ports ON TABLE active_app TYPE array<[int, int]>;
-DEFINE FIELD mapped_ports ON TABLE app_contract TYPE array<[int, int]>;
+DEFINE FIELD host_ipv4 ON TABLE active_app TYPE string;
-DEFINE FIELD host_ipv4 ON TABLE app_contract TYPE string;
+DEFINE FIELD vcpus ON TABLE active_app TYPE int;
-DEFINE FIELD vcpus ON TABLE app_contract TYPE int;
+DEFINE FIELD memory_mb ON TABLE active_app TYPE int;
-DEFINE FIELD memory_mb ON TABLE app_contract TYPE int;
+DEFINE FIELD disk_size_gb ON TABLE active_app TYPE int;
-DEFINE FIELD disk_size_gb ON TABLE app_contract TYPE int;
+DEFINE FIELD created_at ON TABLE active_app TYPE datetime;
-DEFINE FIELD created_at ON TABLE app_contract TYPE datetime;
+DEFINE FIELD price_per_unit ON TABLE active_app TYPE int;
-DEFINE FIELD updated_at ON TABLE app_contract TYPE datetime;
+DEFINE FIELD locked_nano ON TABLE active_app TYPE int;
-DEFINE FIELD price_per_unit ON TABLE app_contract TYPE int;
+DEFINE FIELD collected_at ON TABLE active_app TYPE datetime;
-DEFINE FIELD locked_nano ON TABLE app_contract TYPE int;
+DEFINE FIELD mr_enclave ON TABLE active_app TYPE string;
-DEFINE FIELD collected_at ON TABLE app_contract TYPE datetime;
+DEFINE FIELD package_url ON TABLE active_app TYPE string;
-DEFINE FIELD mr_enclave ON TABLE app_contract TYPE string;
+DEFINE FIELD hratls_pubkey ON TABLE active_app TYPE string;
-DEFINE FIELD package_url ON TABLE app_contract TYPE string;
+
-DEFINE FIELD hratls_pubkey ON TABLE app_contract TYPE string;
+DEFINE TABLE deleted_app TYPE RELATION FROM account TO app_node SCHEMAFULL;
+DEFINE FIELD app_name ON TABLE deleted_app TYPE string;
+DEFINE FIELD mapped_ports ON TABLE deleted_app TYPE array<[int, int]>;
+DEFINE FIELD host_ipv4 ON TABLE deleted_app TYPE string;
+DEFINE FIELD vcpus ON TABLE deleted_app TYPE int;
+DEFINE FIELD memory_mb ON TABLE deleted_app TYPE int;
+DEFINE FIELD disk_size_gb ON TABLE deleted_app TYPE int;
+DEFINE FIELD created_at ON TABLE deleted_app TYPE datetime;
+DEFINE FIELD deleted_at ON TABLE deleted_app TYPE datetime;
+DEFINE FIELD price_per_unit ON TABLE deleted_app TYPE int;
+DEFINE FIELD locked_nano ON TABLE deleted_app TYPE int;
+DEFINE FIELD collected_at ON TABLE deleted_app TYPE datetime;
+DEFINE FIELD mr_enclave ON TABLE deleted_app TYPE string;
+DEFINE FIELD package_url ON TABLE deleted_app TYPE string;
+DEFINE FIELD hratls_pubkey ON TABLE deleted_app TYPE string;
 
 DEFINE TABLE ban TYPE RELATION FROM account TO account;
 DEFINE FIELD created_at ON TABLE ban TYPE datetime;
@@ -72,10 +127,8 @@ DEFINE FIELD created_at ON TABLE ban TYPE datetime;
 DEFINE TABLE kick TYPE RELATION FROM account TO account;
 DEFINE FIELD created_at ON TABLE kick TYPE datetime;
 DEFINE FIELD reason ON TABLE kick TYPE string;
-DEFINE FIELD contract ON TABLE kick TYPE record<vm_contract|app_contract>;
+DEFINE FIELD contract ON TABLE kick TYPE record<deleted_vm|deleted_app>;
 
 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;

saved_data.yaml

@@ -129,7 +129,7 @@ operators:
     app_nodes: []
   7V3rEuh6j8VuwMVB5PyGqWKLmjJ4fYSv6WtrTL51NZTB:
     escrow: 0
-    email: ""
+    email: ''
     banned_users: []
     vm_nodes: []
     app_nodes:
@@ -238,7 +238,7 @@ vm_contracts:
   node_pubkey: Du3UfPSUUZmA5thQmc9Vrxdy7UimpygcpDsQNnwRQPtu
   exposed_ports: []
   public_ipv4: 156.146.63.216
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 10
   vcpus: 2
   memory_mb: 3000
@@ -255,7 +255,7 @@ vm_contracts:
   node_pubkey: 7Xw3RxbP5pvfjZ8U6yA3HHVSS9YXjKH5Vkas3JRbQYd9
   exposed_ports: []
   public_ipv4: 173.234.136.154
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 10
   vcpus: 2
   memory_mb: 3000
@@ -272,8 +272,8 @@ vm_contracts:
   node_pubkey: 7Xw3RxbP5pvfjZ8U6yA3HHVSS9YXjKH5Vkas3JRbQYd9
   exposed_ports:
   - 38288
-    public_ipv4: ""
+  public_ipv4: ''
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 10
   vcpus: 1
   memory_mb: 1000
@@ -290,7 +290,7 @@ vm_contracts:
   node_pubkey: DgkbsrwttkZXvzxY5kDwQQoDd79GLmZ5tc7fYJUFkQQb
   exposed_ports: []
   public_ipv4: 149.22.95.2
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 10
   vcpus: 2
   memory_mb: 3000
@@ -307,7 +307,7 @@ vm_contracts:
   node_pubkey: Du3UfPSUUZmA5thQmc9Vrxdy7UimpygcpDsQNnwRQPtu
   exposed_ports: []
   public_ipv4: 156.146.63.217
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 10
   vcpus: 2
   memory_mb: 3000
@@ -324,7 +324,7 @@ vm_contracts:
   node_pubkey: DgkbsrwttkZXvzxY5kDwQQoDd79GLmZ5tc7fYJUFkQQb
   exposed_ports: []
   public_ipv4: 149.22.95.2
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 30
   vcpus: 1
   memory_mb: 1000
@@ -341,7 +341,7 @@ vm_contracts:
   node_pubkey: 3zRxiGRnf46vd3zAEmpaYBJocTV9oJB6yXf5GZFR1Sq4
   exposed_ports: []
   public_ipv4: 149.36.48.100
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 10
   vcpus: 4
   memory_mb: 4000
@@ -358,8 +358,8 @@ vm_contracts:
   node_pubkey: 3zRxiGRnf46vd3zAEmpaYBJocTV9oJB6yXf5GZFR1Sq4
   exposed_ports:
   - 46393
-    public_ipv4: ""
+  public_ipv4: ''
-    public_ipv6: ""
+  public_ipv6: ''
   disk_size_gb: 10
   vcpus: 1
   memory_mb: 1000
@@ -384,57 +384,4 @@ app_nodes:
   max_ports_per_app: 9
   price: 20000
   offline_minutes: 0
-app_contracts:
+app_contracts: []
-  - uuid: e3d01f25-2b2a-410b-80e3-12f44e474334
-    package_url: https://registry.detee.ltd/sgx/packages/base_package_2025-04-17_11-01-08.tar.gz
-    admin_pubkey: H21Shi4iE7vgfjWEQNvzmpmBMJSaiZ17PYUcdNoAoKNc
-    node_pubkey: BiqoPUEoAxYxMRXUmyofoS9H1TBQgQqvLJ6MbWh88AQg
-    mapped_ports:
-      - - 27158
-        - 34500
-      - - 28667
-        - 8080
-    host_ipv4: 212.95.45.139
-    disk_size_mb: 1000
-    vcpus: 1
-    memory_mb: 1000
-    created_at: 2025-04-21T11:27:28.833236909Z
-    updated_at: 2025-04-21T11:27:28.833237729Z
-    price_per_unit: 200000
-    locked_nano: 121200000
-    collected_at: 2025-04-21T11:28:24.905665571Z
-    hratls_pubkey: 7E0F887AA6BB9104EEC1066F454D4C2D9063D676715F55F919D3FBCEDC63240B
-    public_package_mr_enclave:
-      - 52
-      - 183
-      - 102
-      - 210
-      - 251
-      - 219
-      - 218
-      - 140
-      - 168
-      - 118
-      - 10
-      - 193
-      - 98
-      - 240
-      - 147
-      - 124
-      - 240
-      - 189
-      - 46
-      - 95
-      - 138
-      - 172
-      - 15
-      - 246
-      - 227
-      - 114
-      - 70
-      - 159
-      - 232
-      - 212
-      - 9
-      - 234
-    app_name: diligent-seahorse

scripts/ca_cert.pem

@@ -0,0 +1,35 @@
+-----BEGIN CERTIFICATE-----
+MIIF/TCCA+WgAwIBAgIULPbWfncT/qhqcWgA+ryYqubND78wDQYJKoZIhvcNAQEL
+BQAwgY0xCzAJBgNVBAYTAlZHMRAwDgYDVQQIDAdUb3J0b2xhMRIwEAYDVQQHDAlS
+b2FkIFRvd24xEjAQBgNVBAoMCURlVEVFIEx0ZDENMAsGA1UECwwEV2ViMzETMBEG
+A1UEAwwKZGV0ZWUtcm9vdDEgMB4GCSqGSIb3DQEJARYRc3VwcG9ydEBkZXRlZS5s
+dGQwHhcNMjUwMzI3MTQ1OTQxWhcNMzUwMzI1MTQ1OTQxWjCBjTELMAkGA1UEBhMC
+VkcxEDAOBgNVBAgMB1RvcnRvbGExEjAQBgNVBAcMCVJvYWQgVG93bjESMBAGA1UE
+CgwJRGVURUUgTHRkMQ0wCwYDVQQLDARXZWIzMRMwEQYDVQQDDApkZXRlZS1yb290
+MSAwHgYJKoZIhvcNAQkBFhFzdXBwb3J0QGRldGVlLmx0ZDCCAiIwDQYJKoZIhvcN
+AQEBBQADggIPADCCAgoCggIBAOtwb0JqT61l058FKkXWxYwxcn9mkIQ3JY5t67xL
+dM/eaSYcLCFvQQ8LZilhYUxEIkqF2+qloyhMgru5erHcn/xul7RnIPpj/ActPFEU
+5Snr4lHP6WJebDSFHmKkh4ogwFqMpq3SvAJ0/h1MxZu1hf369hCmyMvevjINX1kB
+VzZGMKUe3M1YOi62Vbhfd3JUkAMedrVmcZoeOE63Fz4NYs/UMbAQYBtEvPp3qYvM
+vLnDJlCrb9fAia4qFOnzqIa40LEcbDiG7Yxw9jvacb9+rKboaPkNWcZqyNl7CQYr
+yOlSPYa6ehoZ4WGrDzrZMOGp88i3Qkd55VxuivSouUS4bkmSS+QPkRHyOGovatfp
+7AmhgQmfozjovSR2Tk+kGD3VxsAPAQWYxJLYHUtjidBUFQnwjAWpU0gh/iydpb0Q
+q1yEUkijMhUP6uHCLrEb+GGrkGgKgFKfgKsbyKjhXe5ftFdBJnMv8jeTxvkca3Ff
+/Tu8DXq3GVj3UZzCqv9w1a1UJTLH5WkAKrGcFsJ1QwW7yLHXW47cIgKUEAkurpWA
+TXJv7faUGcHBhywSMVCXuBRRg5zk/bdK4KXKtPt9U0QHmnNRXfl2t+1jZKVDAfF2
+x3S0x87URL4IZhGgmfTPoIlpc8ktplPQoxKpdrbMj+6BvXTRmRZRH/LoroRWunba
+1g4BAgMBAAGjUzBRMB0GA1UdDgQWBBQRnZzDpOUYk0CeW0R2pALfN80JFzAfBgNV
+HSMEGDAWgBQRnZzDpOUYk0CeW0R2pALfN80JFzAPBgNVHRMBAf8EBTADAQH/MA0G
+CSqGSIb3DQEBCwUAA4ICAQBQBjchxrS2LDH2HNlSOsKwar8F30tkjYG/E00Y0qjn
+x3ciC0Zlo4M0odhAF8rGkLorEbm5JN/k8lI8oKTzne1YF7g05kt4JDlr6C8vmEL2
+KpLkWc+h42z8jjc1Othj6vhHdl+vOKP+W3f2idoImvAijd2JS0+E3XWI8cgMiwHQ
+lxdMqpwk/dwS1D0E4zvXH041VAXJmlE/ys4DTEq234IwEp41AW0z9Pd9EN7QEDaq
+qUaDVOsaYCAdFCWuEucB2v0NcBNDAJVlepH+uGaQ7UH0afADTscIdrSNcNtf87ad
+1U20wiO2ayBTL5s1dz/XyGc/f3QzCSniE2fILkNg31O0wijrfLUhGbxdx0fVfcXS
+jTabojeQkmRoMguW1H5LaKvPSK06gHxFpaPqhJ8XC9Z5xrtvtVI60kquNHX7Sjwd
+wU7s40J3z5+btYHH4mPdXGsSWXS4xqmKvktzLKBJKVSgjjvzLTKspAAAFsHpIjwN
+YxxQYQl+/hmppCsp/XHE5FbT0051nIxepdtJgWfT4Xo8SxtoQy9C8RzWjMiTiYxG
+IuYkATUex//jBRxABy99v6Kx1Wa2agx7aqnAuC1VinTXG+c1RasAoNWg0vgvnUXn
+4x9HmZYJ4J3PxZjWXdn7Bna7ZV6tmmbDMlp4zy2hNEGtOVlE/ffXRyz/vkLD88Bq
+QA==
+-----END CERTIFICATE-----

scripts/ca_cert.srl

@@ -0,0 +1 @@
+449CCB0DA49A05BA82A5F123866D4822A64AAAC5

scripts/create_certs.sh

@@ -0,0 +1,49 @@
+#!/bin/bash
+cd -- "$( dirname -- "${BASH_SOURCE[0]}" )"
+
+mkdir -p secrets
+mkdir -p tmp
+chmod 700 secrets
+
+[[ -f "secrets/ca_key.pem" ]] || {
+  openssl genrsa -out secrets/ca_key.pem 4096
+  chmod 400 secrets/ca_key.pem
+}
+
+[[ -f "ca_cert.pem" ]] || {
+  openssl req -x509 -new -nodes \
+    -key secrets/ca_key.pem -sha256 \
+    -days 3650 -out ca_cert.pem
+}
+
+[[ -f "secrets/staging_key.pem" ]] || {
+  openssl genrsa -out secrets/staging_key.pem 2048
+  chmod 400 secrets/staging_key.pem 
+}
+
+[[ -f "tmp/staging_csr.pem" ]] || {
+  openssl req -new -key secrets/staging_key.pem \
+    -out tmp/staging_csr.pem -config staging_brain.cnf
+}
+
+[[ -f "staging_cert.pem" ]] || {
+  openssl x509 -req -in tmp/staging_csr.pem -CA ca_cert.pem -CAkey secrets/ca_key.pem \
+    -CAcreateserial -out staging_cert.pem -days 825 -sha256 \
+    -extfile staging_brain.cnf -extensions req_ext
+}
+
+[[ -f "secrets/testnet_key.pem" ]] || {
+  openssl genrsa -out secrets/testnet_key.pem 4096
+  chmod 400 secrets/testnet_key.pem 
+}
+
+[[ -f "tmp/testnet_csr.pem" ]] || {
+  openssl req -new -key secrets/testnet_key.pem \
+    -out tmp/testnet_csr.pem -config testnet_brain.cnf
+}
+
+[[ -f "testnet_cert.pem" ]] || {
+  openssl x509 -req -in tmp/testnet_csr.pem -CA ca_cert.pem -CAkey secrets/ca_key.pem \
+    -CAcreateserial -out testnet_cert.pem -days 825 -sha256 \
+    -extfile testnet_brain.cnf -extensions req_ext
+}

scripts/deploy.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+cd -- "$( dirname -- "${BASH_SOURCE[0]}" )"
+cd ..
+
+server="$1"
+
+[[ -z "$server" ]] && {
+  echo "Please specify server ip."
+  exit 1
+}
+
+[[ "$server" == "testnet" ]] && server="164.92.249.180"
+[[ "$server" == "staging" ]] && server="registry.detee.ltd"
+
+cargo build --release
+ssh $server systemctl stop detee-brain-mock.service
+scp target/release/brain-mock $server:/usr/local/bin/brain-mock
+ssh $server mkdir -p /etc/detee/brain-mock/
+scp scripts/detee-brain-mock.service $server:/etc/systemd/system/detee-brain-mock.service
+ssh $server systemctl daemon-reload
+ssh $server systemctl start detee-brain-mock.service

scripts/detee-brain-mock.service

@@ -0,0 +1,11 @@
+[Unit]
+Description=DeTEE Brain Mock
+After=network.target
+
+[Service]
+Type=simple
+ExecStart=/usr/local/bin/brain-mock
+Restart=always
+
+[Install]
+WantedBy=multi-user.target

scripts/staging_brain.cnf

@@ -0,0 +1,20 @@
+[ req ]
+default_bits       = 2048
+prompt             = no
+default_md         = sha256
+distinguished_name = req_distinguished_name
+req_extensions     = req_ext
+
+[ req_distinguished_name ]
+C  = VG
+ST = Tortola
+L  = Road Town
+O  = DeTEE Ltd
+OU = Web3
+CN = staging-brain
+
+[ req_ext ]
+subjectAltName = @alt_names
+
+[ alt_names ]
+DNS.1 = staging-brain

scripts/staging_cert.pem

@@ -0,0 +1,29 @@
+-----BEGIN CERTIFICATE-----
+MIIE5jCCAs6gAwIBAgIURJzLDaSaBbqCpfEjhm1IIqZKqsMwDQYJKoZIhvcNAQEL
+BQAwgY0xCzAJBgNVBAYTAlZHMRAwDgYDVQQIDAdUb3J0b2xhMRIwEAYDVQQHDAlS
+b2FkIFRvd24xEjAQBgNVBAoMCURlVEVFIEx0ZDENMAsGA1UECwwEV2ViMzETMBEG
+A1UEAwwKZGV0ZWUtcm9vdDEgMB4GCSqGSIb3DQEJARYRc3VwcG9ydEBkZXRlZS5s
+dGQwHhcNMjUwMzI4MTQxMzIwWhcNMjcwNzAxMTQxMzIwWjBuMQswCQYDVQQGEwJW
+RzEQMA4GA1UECAwHVG9ydG9sYTESMBAGA1UEBwwJUm9hZCBUb3duMRIwEAYDVQQK
+DAlEZVRFRSBMdGQxDTALBgNVBAsMBFdlYjMxFjAUBgNVBAMMDXN0YWdpbmctYnJh
+aW4wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDcXmOZ1GYsOghZzzS1
+c4139hs1VwB5kK4z2JLXR15SHc1dyDfQO1FBMDMUD/jfROVgTFx3l7X/MGv0hoeA
+h4QsyiDaHcba/WqFJ59rWDNIz5GvI4bDw8OctNrNMrhUtNYtfC9gTkk3N4c06TDE
+8ga9cTuPDw+fCKghvK7TJVF7UDZYaqjf4Et4zo+ahefkeJF8NUD+HTUbZcg5Ebuz
+me4/8b/zORMtXXmRUzcOCZTY5TnQfdGKtO+aYcUEuJusWPvjq3+8duprIElKn3sH
+e8Ju7qrJpX+NurJHEonbtzWspIgJP8/4GO4oetHN/ppXrHtE5qqG6qvS+Fna71DQ
+HGORAgMBAAGjXDBaMBgGA1UdEQQRMA+CDXN0YWdpbmctYnJhaW4wHQYDVR0OBBYE
+FP7OXu7YjWhacQVz7Xi9HqixkcoeMB8GA1UdIwQYMBaAFBGdnMOk5RiTQJ5bRHak
+At83zQkXMA0GCSqGSIb3DQEBCwUAA4ICAQBIQ/EboY0ZVf1VTWtBZKXIWFANDlGc
+vFgejlxeruXGsiJpeQCsAXP6ZMSgVTapSBzTCURbV64vwhlSMJGFzV8m8XFYw6/o
+7mn0VCJjM2309A9uKs/Vk8dhG+BAMUT+bgQW+yyO/agpi5I1ChEVHHNyVI5JVxAR
+wAmKHVKccGnW5Ji9OVFCt14IXWqPo3cE/Y+IaFG9OJYENa3JNRLfXMDoxHpiQ6I3
+v2/YcN2E0m1WwrMgsUpRE8hroLQWCghgzMGjJn0YQ6yTGeh6ibRkIg9yaXLxHygq
+sauPn+JFhY7V/AP0V212ksEfEPHciZPaNriK3y2m2SDVYpXRVHHqWhQxb5yc+B6A
+QWdu45pP1gVM6SGnJDuIrtihg9hUXVB22Uoea6kOGhdlS5m9fv1KRH1ScF7Onbzd
+TjxPLoEzvj6/cNu7XEixjQOSmcs68PX8t+Jp8I2gMCQ++ZzQ7oyS5xzwKcDYcjPm
+2rud5px7H8zwNdP+cNFifSYNHs4ltgXmTDKOhvntGWXjNsq3Olw2tvbLIPQETRQc
+T5BTDMcNPNeXquzer/OJZOkJrZeG5RvbVeQ8AfdldMUNoX9fhSOtIY1L99wculHU
+XqC2NVpZxXDUwR8GKQuLGuOkMQmCdTLd1svJh5Deih4IddII1LP6qP2Izo3CUgDV
+LuxVyvp8squzVg==
+-----END CERTIFICATE-----

scripts/testnet_brain.cnf

@@ -0,0 +1,20 @@
+[ req ]
+default_bits       = 4096
+prompt             = no
+default_md         = sha256
+distinguished_name = req_distinguished_name
+req_extensions     = req_ext
+
+[ req_distinguished_name ]
+C  = VG
+ST = Tortola
+L  = Road Town
+O  = DeTEE Ltd
+OU = Web3
+CN = testnet-brain
+
+[ req_ext ]
+subjectAltName = @alt_names
+
+[ alt_names ]
+DNS.1 = testnet-brain

scripts/testnet_cert.pem

@@ -0,0 +1,34 @@
+-----BEGIN CERTIFICATE-----
+MIIF5jCCA86gAwIBAgIURJzLDaSaBbqCpfEjhm1IIqZKqsUwDQYJKoZIhvcNAQEL
+BQAwgY0xCzAJBgNVBAYTAlZHMRAwDgYDVQQIDAdUb3J0b2xhMRIwEAYDVQQHDAlS
+b2FkIFRvd24xEjAQBgNVBAoMCURlVEVFIEx0ZDENMAsGA1UECwwEV2ViMzETMBEG
+A1UEAwwKZGV0ZWUtcm9vdDEgMB4GCSqGSIb3DQEJARYRc3VwcG9ydEBkZXRlZS5s
+dGQwHhcNMjUwMzI4MTQxNDE1WhcNMjcwNzAxMTQxNDE1WjBuMQswCQYDVQQGEwJW
+RzEQMA4GA1UECAwHVG9ydG9sYTESMBAGA1UEBwwJUm9hZCBUb3duMRIwEAYDVQQK
+DAlEZVRFRSBMdGQxDTALBgNVBAsMBFdlYjMxFjAUBgNVBAMMDXRlc3RuZXQtYnJh
+aW4wggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQCwbfN9iq/Zl97etbqW
+q1DR8WOupTK94r1pZ2cGAyozED2JuQVdslaLU8Jt9QZUlhLYYVf/vXanPmgxh+NA
+VxvLPUjI/RZ84qGi58Uii3YCEQm6AwW0M21HPRQyctBqFc2KvxBBsCNg/G0wLpEI
+qdEP9mHP1k/hlW6nxKsM/jAgeIuEGWH83sqkZnzRb6jVjQw9yPYvY/4UzSX13fdD
+J+ML9H+0qfYgYkpv0Y9LjEwJEOM7tZY5y9LcOBb/CAKgBjb1MWqJuKhTen9ZryGi
+snmqXRMgOygTMiUUKV/cy4SUnlKymUDUxt5sSbV+2f/lzamYzjXreycR+6m4ol4n
+Xz8IfLROEDp0lk07r33fj1z9Z4huH7J9L1eOQpViDNI762hrZzz8my1VyOfyyhOr
+wbg2GtIaM5pnakeWFXRw/+NGQEy2quSDBrHsWliEN2F/pDiDByZhaotNxNTN56Jq
+mmOyOv4HjPgmK5iSmUp+Dpf8CWT+PPec9UqRQfV/6gUeRD68VvRsPD5UmQ+wl/3Y
+DLq0y2GkGmzGOFk2LN916Tl4gynlj1EDtsiXFyHGvJk05ZjjNy/QDJG4BlQIR1yV
+a7uHVCi4GOhE5CUS2XVLZ+kQC/IAiXM2Cw8z9W0JPv4p/CnH8riWacCY22kIz9oH
+Rn7x31YKRsrULjgRMA57up/ycwIDAQABo1wwWjAYBgNVHREEETAPgg10ZXN0bmV0
+LWJyYWluMB0GA1UdDgQWBBSsKdb7/zgpiNza73tRZc+Iw83f3zAfBgNVHSMEGDAW
+gBQRnZzDpOUYk0CeW0R2pALfN80JFzANBgkqhkiG9w0BAQsFAAOCAgEAeDG2X2Lk
+wgbwSrx3fzRVP2KIho7C3rBVX/6p4eisl4s50pHXF9UAHwc2BXY4r+gl1TisF24y
+hWTD9OfYW+q4d7+gcF5smQVeSmwIPSZgIRRaz4YI7p5grICw9+7Qh6IgLw+WsEUw
+URCll5a81CdpITmrKxy4O+MScBY4+M4PZziqaZw60cdjC6hFikrndox91hEYvNdc
+EQXoivYjfB9TO55gwzKHdmBHGzI1hPlTJMdBn4l0QixkJeIk2TBCWWhp15tgrNTC
+HdawZ0cTwVH1CkeXr4jdi1afvX7cGbHPufjKW2KeyasLNaUagVH13NdYTe9et4Nf
+rY3byqXICj9UMZuuMc7GJv07hRJ4DNyZMWtRr0duqAo3frGzJk4C4v25nU9msfCY
+YjqM0KWOlrVPpnH7e8eMLFKZgrD6rV1a+cqvtjGSwNhbOZJ3xCPe/m+zeIOPkgDH
+hDKoOagHVyBS+9ryIeEYmipxg7yjpbFUmI9Z8FE+teZdA0iBRjyikqzgtten7ZP8
+uJiSAEbqn0l1O/qAyI6SlD/nsCX513KRk6kvFEWSud2vePsSQ9gtwjKCw3E4/OdL
+AWUEOWQlCHVQioyrVc2WwRtO6o+prb+Nk/TTp9Gyp1fQjqMquESIsNoUvhrMOwXf
+nIUh6pszMpdBlOnyry4RUK3I0sgM5TACZ1Y=
+-----END CERTIFICATE-----

src/bin/brain.rs

@@ -1,7 +1,7 @@
 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::grpc::BrainGeneralCliForReal;
-use surreal_brain::grpc::BrainVmCliMock;
+use surreal_brain::grpc::BrainVmCliForReal;
 use surreal_brain::db;
 use tonic::transport::{Identity, Server, ServerTlsConfig};
 
@@ -11,8 +11,8 @@ async fn main() {
     db::init().await.unwrap();
     let addr = "0.0.0.0:31337".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 key = std::fs::read_to_string("./tmp/brain-key.pem").unwrap();

src/db.rs

@@ -5,19 +5,32 @@ use surrealdb::{
     engine::remote::ws::{Client, Ws},
     opt::auth::Root,
     sql::Datetime,
-    RecordId, Surreal,
+    Notification, RecordId, Surreal,
 };
+use tokio::sync::mpsc::Sender;
+use tokio_stream::StreamExt as _;
 
 static DB: LazyLock<Surreal<Client>> = LazyLock::new(Surreal::init);
-const ACCOUNT: &str = "account";
+pub const ACCOUNT: &str = "account";
-const OPERATOR: &str = "operator";
+pub const VM_NODE: &str = "vm_node";
-const VM_CONTRACT: &str = "vm_contract";
+pub const ACTIVE_VM: &str = "active_vm";
-const VM_NODE: &str = "vm_node";
+pub const NEW_VM_REQ: &str = "new_vm_req";
+pub const UPDATE_VM_REQ: &str = "update_vm_req";
+pub const DELETED_VM: &str = "deleted_vm";
+
+pub const ID_ALPHABET: [char; 62] = [
+    '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i',
+    'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z', 'A', 'B',
+    'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U',
+    'V', 'W', 'X', 'Y', 'Z',
+];
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
-    #[error(transparent)]
+    #[error("Internal DB error: {0}")]
     DataBase(#[from] surrealdb::Error),
+    #[error("Daemon channel got closed: {0}")]
+    DaemonConnection(#[from] tokio::sync::mpsc::error::SendError<DaemonNotification>),
 }
 
 pub async fn init() -> surrealdb::Result<()> {
@@ -28,13 +41,22 @@ pub async fn init() -> surrealdb::Result<()> {
     Ok(())
 }
 
+pub async fn upsert_record<SomeRecord: Serialize + 'static>(
+    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(old_data: &old_brain::BrainData) -> surrealdb::Result<()> {
     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<VmContract> = old_data.into();
+    let vm_contracts: Vec<ActiveVm> = old_data.into();
-    let operators: Vec<OperatorRelation> = old_data.into();
-    let app_contracts: Vec<AppContract> = old_data.into();
 
     init().await?;
 
@@ -45,11 +67,7 @@ pub async fn migration0(old_data: &old_brain::BrainData) -> surrealdb::Result<()
     println!("Inserting app nodes...");
     let _: Vec<AppNode> = DB.insert(()).content(app_nodes).await?;
     println!("Inserting vm contracts...");
-    let _: Vec<VmContract> = DB.insert("vm_contract").relation(vm_contracts).await?;
+    let _: Vec<ActiveVm> = DB.insert("vm_contract").relation(vm_contracts).await?;
-    println!("Inserting app contracts...");
-    let _: Vec<AppContract> = DB.insert("app_contract").relation(app_contracts).await?;
-    println!("Inserting operators...");
-    let _: Vec<OperatorRelation> = DB.insert(OPERATOR).relation(operators).await?;
 
     Ok(())
 }
@@ -85,9 +103,25 @@ impl Account {
     }
 }
 
+impl Account {
+    pub async fn is_banned_by_node(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,
+    pub operator: RecordId,
     pub country: String,
     pub region: String,
     pub city: String,
@@ -103,14 +137,224 @@ pub struct VmNode {
     pub offline_minutes: u64,
 }
 
+#[derive(Serialize)]
+pub struct VmNodeResources {
+    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,
+}
+
+impl VmNodeResources {
+    pub async fn merge(self, node_id: &str) -> Result<(), Error> {
+        let _: Option<VmNode> = DB.update((VM_NODE, node_id)).merge(self).await?;
+        Ok(())
+    }
+}
+
+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 VmContract {
+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>,
+}
+
+impl VmNodeWithReports {
+    // TODO: find a more elegant way to do this than importing gRPC in the DB module
+    // https://en.wikipedia.org/wiki/Dependency_inversion_principle
+    pub async fn find_by_filters(
+        filters: detee_shared::snp::pb::vm_proto::VmNodeFilters,
+    ) -> Result<Vec<Self>, Error> {
+        let mut query = format!(
+            "select *, <-report.* as reports from {VM_NODE} where
+            avail_ports >= {} &&
+            max_ports_per_vm >= {} &&
+            avail_ipv4 >= {} &&
+            avail_ipv6 >= {} &&
+            avail_vcpus >= {} &&
+            avail_mem_mb >= {} &&
+            avail_storage_gbs >= {}\n",
+            filters.free_ports,
+            filters.free_ports,
+            filters.offers_ipv4 as u32,
+            filters.offers_ipv6 as u32,
+            filters.vcpus,
+            filters.memory_mb,
+            filters.storage_gb
+        );
+        if !filters.city.is_empty() {
+            query += &format!(r#"&& city = "{}"\n"#, filters.city);
+        }
+        if !filters.region.is_empty() {
+            query += &format!(r#"&& region = "{}"\n"#, filters.region);
+        }
+        if !filters.country.is_empty() {
+            query += &format!(r#"&& country = "{}"\n"#, filters.country);
+        }
+        if !filters.ip.is_empty() {
+            query += &format!(r#"&& ip = "{}"\n"#, filters.ip);
+        }
+        query += ";";
+        let mut result =
+            DB.query(query).await?;
+        let vm_nodes: Vec<Self> = result.take(0)?;
+        Ok(vm_nodes)
+    }
+}
+
+pub enum DaemonNotification {
+    Create(NewVmReq),
+    Update(UpdateVmReq),
+    Delete(DeletedVm),
+}
+
+impl From<NewVmReq> for DaemonNotification {
+    fn from(value: NewVmReq) -> Self {
+        Self::Create(value)
+    }
+}
+
+impl From<UpdateVmReq> for DaemonNotification {
+    fn from(value: UpdateVmReq) -> Self {
+        Self::Update(value)
+    }
+}
+
+impl From<DeletedVm> for DaemonNotification {
+    fn from(value: DeletedVm) -> Self {
+        Self::Delete(value)
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct NewVmReq {
+    pub id: RecordId,
+    #[serde(rename = "in")]
+    pub admin: RecordId,
+    #[serde(rename = "out")]
+    pub vm_node: RecordId,
+    pub hostname: String,
+    pub extra_ports: Vec<u32>,
+    pub public_ipv4: bool,
+    pub public_ipv6: bool,
+    pub disk_size_gb: u32,
+    pub vcpus: u32,
+    pub memory_mb: u32,
+    pub dtrfs_url: String,
+    pub dtrfs_sha: String,
+    pub kernel_sha: String,
+    pub kernel_url: String,
+    pub created_at: Datetime,
+    pub price_per_unit: u64,
+    pub locked_nano: u64,
+    pub error: String,
+}
+
+impl NewVmReq {
+    pub async fn submit_error(id: &str, error: String) -> Result<(), Error> {
+        #[derive(Serialize)]
+        struct NewVmError {
+            error: String,
+        }
+        let _: Option<VmNode> = DB.update((NEW_VM_REQ, id)).merge(NewVmError { error }).await?;
+        Ok(())
+    }
+
+    pub async fn submit(self) -> Result<(), Error> {
+        let _: Option<Self> = DB.create(self.id.clone()).content(self).await?;
+        Ok(())
+    }
+}
+
+/// first string is the vm_id
+pub enum NewVmResp {
+    // TODO: find a more elegant way to do this than importing gRPC in the DB module
+    // https://en.wikipedia.org/wiki/Dependency_inversion_principle
+    Args(String, detee_shared::snp::pb::vm_proto::MeasurementArgs),
+    Error(String, String),
+}
+
+impl NewVmResp {
+    pub async fn listen(vm_id: &str) -> Result<NewVmResp, Error> {
+        let mut resp = DB
+            .query(format!("live select * from {NEW_VM_REQ} where id = {NEW_VM_REQ}:{vm_id};"))
+            .query(format!(
+                "live select * from measurement_args where id = measurement_args:{vm_id};"
+            ))
+            .await?;
+        let mut live_stream1 = resp.stream::<Notification<NewVmReq>>(0)?;
+        let mut live_stream2 =
+            resp.stream::<Notification<detee_shared::snp::pb::vm_proto::MeasurementArgs>>(1)?;
+
+        loop {
+            tokio::select! {
+                new_vm_req_notif = live_stream1.next() => {
+                    if let Some(new_vm_req_notif) = new_vm_req_notif {
+                        match new_vm_req_notif {
+                            Ok(new_vm_req_notif) => {
+                                match new_vm_req_notif.action {
+                                    surrealdb::Action::Update => {
+                                        if !new_vm_req_notif.data.error.is_empty() {
+                                            return Ok(Self::Error(vm_id.to_string(), new_vm_req_notif.data.error));
+                                        }
+                                    },
+                                    _ => {}
+                                };
+                            },
+                            Err(e) => return Err(e.into()),
+                        }
+                    }
+                }
+                args_notif = live_stream2.next() => {
+                    if let Some(args_notif) = args_notif {
+                        match args_notif {
+                            Ok(args_notif) => {
+                                match args_notif.action {
+                                    surrealdb::Action::Create => {
+                                        return Ok(Self::Args(vm_id.to_string(), args_notif.data));
+                                    },
+                                    _ => {}
+                                };
+                            },
+                            Err(e) => return Err(e.into()),
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ActiveVm {
     pub id: RecordId,
     #[serde(rename = "in")]
     pub admin: RecordId,
     #[serde(rename = "out")]
     pub vm_node: RecordId,
-    pub state: String,
     pub hostname: String,
     pub mapped_ports: Vec<(u32, u32)>,
     pub public_ipv4: String,
@@ -121,13 +365,144 @@ pub struct VmContract {
     pub dtrfs_sha: String,
     pub kernel_sha: String,
     pub created_at: Datetime,
-    pub updated_at: Datetime,
     pub price_per_unit: u64,
     pub locked_nano: u64,
     pub collected_at: Datetime,
 }
 
-impl VmContract {
+#[derive(Debug, Serialize, Deserialize)]
+pub struct UpdateVmReq {
+    pub id: RecordId,
+    #[serde(rename = "in")]
+    pub admin: RecordId,
+    #[serde(rename = "out")]
+    pub vm_node: RecordId,
+    pub disk_size_gb: u32,
+    pub vcpus: u32,
+    pub memory_mb: u32,
+    pub dtrfs_url: String,
+    pub dtrfs_sha: String,
+    pub kernel_sha: String,
+    pub kernel_url: String,
+    pub created_at: Datetime,
+    pub price_per_unit: u64,
+    pub locked_nano: u64,
+}
+
+pub async fn listen_for_node<
+    T: Into<DaemonNotification> + std::marker::Unpin + for<'de> Deserialize<'de>,
+>(
+    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) => match 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,
+    #[serde(rename = "in")]
+    pub admin: RecordId,
+    #[serde(rename = "out")]
+    pub vm_node: RecordId,
+    pub hostname: String,
+    pub mapped_ports: Vec<(u32, u32)>,
+    pub public_ipv4: String,
+    pub public_ipv6: String,
+    pub disk_size_gb: u32,
+    pub vcpus: u32,
+    pub memory_mb: u32,
+    pub dtrfs_sha: String,
+    pub kernel_sha: String,
+    pub created_at: Datetime,
+    pub deleted_at: Datetime,
+    pub price_per_unit: u64,
+}
+
+impl DeletedVm {
+    pub async fn get_by_uuid(uuid: &str) -> Result<Option<Self>, Error> {
+        let contract: Option<Self> =
+            DB.query(format!("select * from {DELETED_VM}:{uuid};")).await?.take(0)?;
+        Ok(contract)
+    }
+
+    pub async fn list_by_admin(admin: &str) -> Result<Vec<Self>, Error> {
+        let mut result =
+            DB.query(format!("select * from {DELETED_VM} where in = {ACCOUNT}:{admin};")).await?;
+        let contracts: Vec<Self> = result.take(0)?;
+        Ok(contracts)
+    }
+
+    pub async fn list_by_node(admin: &str) -> Result<Vec<Self>, Error> {
+        let mut result =
+            DB.query(format!("select * from {DELETED_VM} where out = {VM_NODE}:{admin};")).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!(
+                "select
+                (select * from ->operator->vm_node<-{DELETED_VM}) as contracts
+            from {ACCOUNT}:{operator};"
+            ))
+            .await?;
+
+        #[derive(Deserialize)]
+        struct Wrapper {
+            contracts: Vec<DeletedVm>,
+        }
+
+        let c: Option<Wrapper> = result.take(0)?;
+        match c {
+            Some(c) => Ok(c.contracts),
+            None => Ok(Vec::new()),
+        }
+    }
+
+    /// total hardware units of this VM
+    fn total_units(&self) -> u64 {
+        // TODO: Optimize this based on price of hardware.
+        // I tried, but this can be done better.
+        // Storage cost should also be based on tier
+        (self.vcpus as u64 * 10)
+            + ((self.memory_mb + 256) as u64 / 200)
+            + (self.disk_size_gb as u64 / 10)
+            + (!self.public_ipv4.is_empty() as u64 * 10)
+    }
+
+    /// Returns price per minute in nanoLP
+    pub fn price_per_minute(&self) -> u64 {
+        self.total_units() * self.price_per_unit
+    }
+}
+
+impl ActiveVm {
     /// total hardware units of this VM
     fn total_units(&self) -> u64 {
         // TODO: Optimize this based on price of hardware.
@@ -146,7 +521,7 @@ impl VmContract {
 }
 
 #[derive(Debug, Serialize, Deserialize)]
-pub struct VmContractWithNode {
+pub struct ActiveVmWithNode {
     pub id: RecordId,
     #[serde(rename = "in")]
     pub admin: RecordId,
@@ -169,16 +544,24 @@ pub struct VmContractWithNode {
     pub collected_at: Datetime,
 }
 
-impl VmContractWithNode {
+impl ActiveVmWithNode {
     pub async fn get_by_uuid(uuid: &str) -> Result<Option<Self>, Error> {
         let contract: Option<Self> =
-            DB.query(format!("select * from {VM_CONTRACT}:{uuid} fetch out;")).await?.take(0)?;
+            DB.query(format!("select * from {ACTIVE_VM}:{uuid} fetch out;")).await?.take(0)?;
         Ok(contract)
     }
 
     pub async fn list_by_admin(admin: &str) -> Result<Vec<Self>, Error> {
         let mut result = DB
-            .query(format!("select * from {VM_CONTRACT} where in = {ACCOUNT}:{admin} fetch out;"))
+            .query(format!("select * from {ACTIVE_VM} where in = {ACCOUNT}:{admin} fetch out;"))
+            .await?;
+        let contracts: Vec<Self> = result.take(0)?;
+        Ok(contracts)
+    }
+
+    pub async fn list_by_node(admin: &str) -> Result<Vec<Self>, Error> {
+        let mut result = DB
+            .query(format!("select * from {ACTIVE_VM} where out = {VM_NODE}:{admin} fetch out;"))
             .await?;
         let contracts: Vec<Self> = result.take(0)?;
         Ok(contracts)
@@ -188,14 +571,14 @@ impl VmContractWithNode {
         let mut result = DB
             .query(format!(
                 "select
-                (select * from ->operator->vm_node<-vm_contract fetch out) as contracts
+                (select * from ->operator->vm_node<-{ACTIVE_VM} fetch out) as contracts
             from {ACCOUNT}:{operator};"
             ))
             .await?;
 
         #[derive(Deserialize)]
         struct Wrapper {
-            contracts: Vec<VmContractWithNode>,
+            contracts: Vec<ActiveVmWithNode>,
         }
 
         let c: Option<Wrapper> = result.take(0)?;
@@ -224,18 +607,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)]
@@ -247,11 +649,11 @@ pub struct AppContract {
     app_node: RecordId,
     state: String,
     app_name: String,
-    mapped_ports: Vec<(u32, u32)>,
+    mapped_ports: Vec<(u64, u64)>,
     host_ipv4: String,
-    vcpus: u32,
+    vcpus: u64,
-    memory_mb: u32,
+    memory_mb: u64,
-    disk_size_gb: u32,
+    disk_size_gb: u64,
     created_at: Datetime,
     updated_at: Datetime,
     price_per_unit: u64,
@@ -291,7 +693,7 @@ pub struct Report {
     #[serde(rename = "out")]
     to_node: RecordId,
     created_at: Datetime,
-    reason: String,
+    pub reason: String,
 }
 
 impl Report {
@@ -309,23 +711,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)]
@@ -342,21 +727,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
@@ -366,7 +792,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(),
@@ -386,7 +813,7 @@ impl From<&old_brain::BrainData> for Vec<VmNode> {
     }
 }
 
-impl From<&old_brain::BrainData> for Vec<VmContract> {
+impl From<&old_brain::BrainData> for Vec<ActiveVm> {
     fn from(old_data: &old_brain::BrainData) -> Self {
         let mut contracts = Vec::new();
         for old_c in old_data.vm_contracts.iter() {
@@ -394,11 +821,10 @@ impl From<&old_brain::BrainData> for Vec<VmContract> {
             for port in old_c.exposed_ports.iter() {
                 mapped_ports.push((*port, 8080 as u32));
             }
-            contracts.push(VmContract {
+            contracts.push(ActiveVm {
-                id: RecordId::from((VM_CONTRACT, old_c.uuid.replace("-", ""))),
+                id: RecordId::from((ACTIVE_VM, old_c.uuid.replace("-", ""))),
                 admin: RecordId::from((ACCOUNT, old_c.admin_pubkey.clone())),
                 vm_node: RecordId::from((VM_NODE, old_c.node_pubkey.clone())),
-                state: "active".to_string(),
                 hostname: old_c.hostname.clone(),
                 mapped_ports,
                 public_ipv4: old_c.public_ipv4.clone(),
@@ -411,7 +837,6 @@ impl From<&old_brain::BrainData> for Vec<VmContract> {
                 price_per_unit: old_c.price_per_unit,
                 locked_nano: old_c.locked_nano,
                 created_at: old_c.created_at.into(),
-                updated_at: old_c.updated_at.into(),
                 collected_at: old_c.collected_at.into(),
             });
         }
@@ -425,6 +850,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(),
@@ -442,48 +868,6 @@ impl From<&old_brain::BrainData> for Vec<AppNode> {
     }
 }
 
-impl From<&old_brain::BrainData> for Vec<AppContract> {
-    fn from(old_data: &old_brain::BrainData) -> Self {
-        let mut contracts = Vec::new();
-        for old_c in old_data.app_contracts.iter() {
-            let mut mapped_ports = Vec::new();
-            for port in old_c.mapped_ports.clone().into_iter().map(|(b, c)| (b as u32, c as u32)) {
-                mapped_ports.push(port);
-            }
-
-            let mr_enclave_hex = old_c
-                .public_package_mr_enclave
-                .clone()
-                .unwrap_or_default()
-                .iter()
-                .map(|byte| format!("{:02X}", byte))
-                .collect();
-
-            contracts.push(AppContract {
-                id: RecordId::from(("app_contract", old_c.uuid.replace("-", ""))),
-                admin: RecordId::from(("account", old_c.admin_pubkey.clone())),
-                app_node: RecordId::from(("app_node", old_c.node_pubkey.clone())),
-                state: "active".to_string(),
-                mapped_ports,
-                host_ipv4: old_c.host_ipv4.clone(),
-                disk_size_gb: old_c.disk_size_mb * 1024,
-                vcpus: old_c.vcpus,
-                memory_mb: old_c.memory_mb,
-                price_per_unit: old_c.price_per_unit,
-                locked_nano: old_c.locked_nano,
-                created_at: old_c.created_at.into(),
-                updated_at: old_c.updated_at.into(),
-                collected_at: old_c.collected_at.into(),
-                app_name: old_c.app_name.clone(),
-                mr_enclave: mr_enclave_hex,
-                package_url: old_c.package_url.clone(),
-                hratls_pubkey: old_c.hratls_pubkey.clone(),
-            });
-        }
-        contracts
-    }
-}
-
 impl From<&old_brain::BrainData> for Vec<Account> {
     fn from(old_data: &old_brain::BrainData) -> Self {
         let mut accounts = Vec::new();
@@ -504,18 +888,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,6 @@
 #![allow(dead_code)]
 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,19 +8,22 @@ 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 nanoid::nanoid;
 
 use log::info;
 use std::pin::Pin;
+use surrealdb::RecordId;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
-// use tokio::sync::mpsc;
+use tokio_stream::{Stream, StreamExt};
-// use tokio_stream::{wrappers::ReceiverStream, Stream};
+use tonic::{Request, Response, Status, Streaming};
-use tokio_stream::Stream;
-use tonic::{Request, Response, Status};
 
-pub struct BrainGeneralCliMock {}
+pub struct BrainGeneralCliForReal {}
 
 impl From<db::Account> for AccountBalance {
     fn from(account: db::Account) -> Self {
@@ -28,8 +31,112 @@ impl From<db::Account> for AccountBalance {
     }
 }
 
-impl From<db::VmContractWithNode> for VmContract {
+impl From<NewVmReq> for db::NewVmReq {
-    fn from(db_c: db::VmContractWithNode) -> Self {
+    fn from(new_vm_req: NewVmReq) -> Self {
+        Self {
+            id: RecordId::from((db::NEW_VM_REQ, nanoid!(40, &db::ID_ALPHABET))),
+            hostname: new_vm_req.hostname,
+            admin: RecordId::from((db::ACCOUNT, new_vm_req.admin_pubkey)),
+            vm_node: RecordId::from((db::VM_NODE, new_vm_req.node_pubkey)),
+            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);
@@ -84,8 +191,200 @@ 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,
+        }
+    }
+}
+
+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,
+        }
+    }
+}
+
+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((db::VM_NODE, req.node_pubkey.clone())),
+            operator: surrealdb::RecordId::from((db::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::ActiveVmWithNode::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> {
+        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 pubkey = pubkey.clone();
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                match db::listen_for_node::<db::DeletedVm>(&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 pubkey = pubkey.clone();
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                let _ = db::listen_for_node::<db::NewVmReq>(&pubkey, tx.clone()).await;
+            });
+        }
+        {
+            let pubkey = pubkey.clone();
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                let _ = db::listen_for_node::<db::UpdateVmReq>(&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)) => {
+                        if !new_vm_resp.error.is_empty() {
+                        } else {
+                            db::upsert_record(
+                                "measurement_args",
+                                &new_vm_resp.uuid,
+                                new_vm_resp.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(&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>>;
@@ -100,7 +399,7 @@ impl BrainGeneralCli for BrainGeneralCliMock {
 
     async fn report_node(&self, req: Request<ReportNodeReq>) -> Result<Response<Empty>, Status> {
         let req = check_sig_from_req(req)?;
-        let (account, node) = match db::VmContractWithNode::get_by_uuid(&req.contract).await? {
+        let (account, node) = match db::ActiveVmWithNode::get_by_uuid(&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 =>
@@ -134,13 +433,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,40 +546,37 @@ 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>>;
 
     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:?}");
-        todo!();
+        if db::Account::is_banned_by_node(&req.admin_pubkey, &req.node_pubkey).await? {
-        // if self
+            return Err(Status::permission_denied("This operator banned you. What did you do?"));
-        //     .data
+        }
-        //     .is_user_banned_by_node(&req.admin_pubkey, &req.node_pubkey)
+
-        // {
+        let new_vm_req: db::NewVmReq = req.into();
-        //     return Err(Status::permission_denied(
+        let id = new_vm_req.id.key().to_string();
-        //         "This operator banned you. What did you do?",
+        let (oneshot_tx, oneshot_rx) = tokio::sync::oneshot::channel();
-        //     ));
+        tokio::spawn(async move {
-        // }
+            let _ = oneshot_tx.send(db::NewVmResp::listen(&id).await);
-        // let admin_pubkey = req.admin_pubkey.clone();
+        });
-        // let (oneshot_tx, oneshot_rx) = tokio::sync::oneshot::channel();
+        new_vm_req.submit().await?;
-        // self.data.submit_newvm_req(req, oneshot_tx).await;
+
-        // match oneshot_rx.await {
+        match oneshot_rx.await {
-        //     Ok(response) => {
+            Ok(new_vm_resp) => Ok(Response::new(new_vm_resp?.into())),
-        //         info!("Sending VM confirmation to {admin_pubkey}: {response:?}");
+            Err(e) => {
-        //         Ok(Response::new(response))
+                log::error!("Something weird happened. Reached error {e:?}");
-        //     }
+                Err(Status::unknown(
-        //     Err(e) => {
+                    "Request failed due to unknown error. Please try again or contact the DeTEE devs team.",
-        //         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> {
@@ -329,7 +628,7 @@ impl BrainVmCli for BrainVmCliMock {
         );
         let mut contracts = Vec::new();
         if !req.uuid.is_empty() {
-            if let Some(specific_contract) = db::VmContractWithNode::get_by_uuid(&req.uuid).await? {
+            if let Some(specific_contract) = db::ActiveVmWithNode::get_by_uuid(&req.uuid).await? {
                 if specific_contract.admin.key().to_string() == req.wallet {
                     contracts.push(specific_contract.into());
                 }
@@ -337,12 +636,11 @@ impl BrainVmCli for BrainVmCliMock {
             }
         } else {
             if req.as_operator {
-                contracts.append(
+                contracts
-                    &mut db::VmContractWithNode::list_by_operator(&req.wallet).await?.into(),
+                    .append(&mut db::ActiveVmWithNode::list_by_operator(&req.wallet).await?.into());
-                );
             } else {
                 contracts
-                    .append(&mut db::VmContractWithNode::list_by_admin(&req.wallet).await?.into());
+                    .append(&mut db::ActiveVmWithNode::list_by_admin(&req.wallet).await?.into());
             }
         }
         let (tx, rx) = mpsc::channel(6);
@@ -361,18 +659,15 @@ impl BrainVmCli for BrainVmCliMock {
     ) -> Result<Response<Self::ListVmNodesStream>, tonic::Status> {
         let req = check_sig_from_req(req)?;
         info!("CLI requested ListVmNodesStream: {req:?}");
-        todo!();
+        let nodes = db::VmNodeWithReports::find_by_filters(req).await?;
-        // let nodes = self.data.find_vm_nodes_by_filters(&req);
+        let (tx, rx) = mpsc::channel(6);
-        // let (tx, rx) = mpsc::channel(6);
+        tokio::spawn(async move {
-        // tokio::spawn(async move {
+            for node in nodes {
-        //     for node in nodes {
+                let _ = tx.send(Ok(node.into())).await;
-        //         let _ = tx.send(Ok(node.into())).await;
+            }
-        //     }
+        });
-        // });
+        let output_stream = ReceiverStream::new(rx);
-        // let output_stream = ReceiverStream::new(rx);
+        Ok(Response::new(Box::pin(output_stream) as Self::ListVmNodesStream))
-        // Ok(Response::new(
-        //     Box::pin(output_stream) as Self::ListVmNodesStream
-        // ))
     }
 
     async fn get_one_vm_node(
@@ -499,6 +794,46 @@ fn check_sig_from_req<T: std::fmt::Debug + PubkeyGetter>(req: Request<T>) -> Res
     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 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(())
+}
+
 const ADMIN_ACCOUNTS: &[&str] = &[
     "x52w7jARC5erhWWK65VZmjdGXzBK6ZDgfv1A283d8XK",
     "FHuecMbeC1PfjkW2JKyoicJAuiU7khgQT16QUB3Q1XdL",