ghe0/brain-to-surreal
1
0
Fork 1
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add operator list

Browse Source
This commit is contained in:
ghe0 2025-04-23 14:14:00 +03:00
parent f51e29fc4b
commit a5622e59fc
Signed by: ghe0
GPG Key ID: 451028EE56A0FBB4
2 changed files with 76 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
src/db.rs
View File

@ -33,7 +33,7 @@ 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<Operator> = old_data.into();
let operators: Vec<OperatorRelation> = old_data.into();
init().await?;
@ -46,7 +46,7 @@ pub async fn migration0(old_data: &old_brain::BrainData) -> surrealdb::Result<()
println!("Inserting vm contracts...");
let _: Vec<VmContract> = DB.insert("vm_contract").relation(vm_contracts).await?;
println!("Inserting operators...");
let _: Vec<Operator> = DB.insert("operator").relation(operators).await?;
let _: Vec<OperatorRelation> = DB.insert(OPERATOR).relation(operators).await?;
Ok(())
}
@ -175,13 +175,13 @@ pub struct VmContractWithNode {
}
impl VmContractWithNode {
pub async fn by_uuid(uuid: &str) -> Result<Option<Self>, Error> {
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)?;
Ok(contract)
}
pub async fn by_admin(admin: &str) -> Result<Vec<Self>, Error> {
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;"))
.await?;
@ -189,7 +189,7 @@ impl VmContractWithNode {
Ok(contracts)
}
pub async fn by_operator(operator: &str) -> Result<Vec<Self>, Error> {
pub async fn list_by_operator(operator: &str) -> Result<Vec<Self>, Error> {
let mut result = DB
.query(format!(
"select
@ -301,14 +301,14 @@ pub struct Report {
}
#[derive(Debug, Serialize, Deserialize)]
pub struct Operator {
pub struct OperatorRelation {
#[serde(rename = "in")]
account: RecordId,
pub account: RecordId,
#[serde(rename = "out")]
node: RecordId,
pub node: RecordId,
}
impl Operator {
impl OperatorRelation {
fn new(account: &str, vm_node: &str) -> Self {
Self {
account: RecordId::from(("account", account.to_string())),
@ -317,6 +317,37 @@ impl Operator {
}
}
#[derive(Debug, Serialize, Deserialize)]
pub struct Operator {
pub account: RecordId,
pub app_nodes: u64,
pub vm_nodes: u64,
pub email: String,
pub escrow: u64,
pub reports: u64,
}
impl Operator {
pub async fn list() -> Result<Vec<Self>, Error> {
let mut result = DB
.query(format!(
"select *,
in as account,
<-account.email[0] as email,
<-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)?;
Ok(operators)
}
}
// TODO: delete all of these From implementation after migration 0 gets executed
impl From<&old_brain::BrainData> for Vec<VmNode> {
@ -353,9 +384,9 @@ 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("-", ""))),
admin: RecordId::from(("account", old_c.admin_pubkey.clone())),
vm_node: RecordId::from(("vm_node", old_c.node_pubkey.clone())),
id: RecordId::from((VM_CONTRACT, 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,
@ -421,15 +452,15 @@ impl From<&old_brain::BrainData> for Vec<Account> {
}
}
impl From<&old_brain::BrainData> for Vec<Operator> {
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(Operator::new(&operator.0, vm_node));
operator_entries.push(OperatorRelation::new(&operator.0, vm_node));
}
for app_node in operator.1.app_nodes.iter() {
operator_entries.push(Operator::new(&operator.0, app_node));
operator_entries.push(OperatorRelation::new(&operator.0, app_node));
}
}
operator_entries

46
src/grpc.rs
View File

@ -71,6 +71,19 @@ impl From<db::Error> for tonic::Status {
}
}
impl From<db::Operator> for ListOperatorsResp {
fn from(db_o: db::Operator) -> Self {
ListOperatorsResp {
pubkey: db_o.account.key().to_string(),
escrow: db_o.escrow,
email: db_o.email,
app_nodes: db_o.app_nodes,
vm_nodes: db_o.vm_nodes,
reports: db_o.reports,
}
}
}
#[tonic::async_trait]
impl BrainGeneralCli for BrainGeneralCliMock {
type ListAccountsStream = Pin<Box<dyn Stream<Item = Result<Account, Status>> + Send>>;
@ -109,19 +122,18 @@ impl BrainGeneralCli for BrainGeneralCliMock {
async fn list_operators(
&self,
_req: Request<Empty>,
req: Request<Empty>,
) -> Result<Response<Self::ListOperatorsStream>, Status> {
todo!();
// let _ = check_sig_from_req(req)?;
// let operators = self.data.list_operators();
// let (tx, rx) = mpsc::channel(6);
// tokio::spawn(async move {
// for op in operators {
// let _ = tx.send(Ok(op.into())).await;
// }
// });
// let output_stream = ReceiverStream::new(rx);
// Ok(Response::new(Box::pin(output_stream) as Self::ListOperatorsStream))
let _ = check_sig_from_req(req)?;
let operators = db::Operator::list().await?;
let (tx, rx) = mpsc::channel(6);
tokio::spawn(async move {
for op in operators {
let _ = tx.send(Ok(op.into())).await;
}
});
let output_stream = ReceiverStream::new(rx);
Ok(Response::new(Box::pin(output_stream) as Self::ListOperatorsStream))
}
async fn inspect_operator(
@ -322,7 +334,7 @@ impl BrainVmCli for BrainVmCliMock {
);
let mut contracts = Vec::new();
if !req.uuid.is_empty() {
if let Some(specific_contract) = db::VmContractWithNode::by_uuid(&req.uuid).await? {
if let Some(specific_contract) = db::VmContractWithNode::get_by_uuid(&req.uuid).await? {
if specific_contract.admin.key().to_string() == req.wallet {
contracts.push(specific_contract.into());
}
@ -330,10 +342,12 @@ impl BrainVmCli for BrainVmCliMock {
}
} else {
if req.as_operator {
contracts
.append(&mut db::VmContractWithNode::by_operator(&req.wallet).await?.into());
contracts.append(
&mut db::VmContractWithNode::list_by_operator(&req.wallet).await?.into(),
);
} else {
contracts.append(&mut db::VmContractWithNode::by_admin(&req.wallet).await?.into());
contracts
.append(&mut db::VmContractWithNode::list_by_admin(&req.wallet).await?.into());
}
}
let (tx, rx) = mpsc::channel(6);