darkfi/bin/dnetview/src/parser.rs

/* This file is part of DarkFi (https://dark.fi)
 *
 * Copyright (C) 2020-2023 Dyne.org foundation
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

use std::collections::hash_map::Entry;

use async_std::sync::Arc;
use log::{debug, error, info};
use serde_json::Value;
use smol::Executor;
use url::Url;

use darkfi::util::{async_util, time::NanoTimestamp};

use crate::{
    config::{DnvConfig, Node, NodeType},
    error::{DnetViewError, DnetViewResult},
    model::{
        LilithInfo, Model, NetworkInfo, NodeInfo, SelectableObject, Session, SessionInfo, SlotInfo,
    },
    rpc::RpcConnect,
    util::{make_empty_id, make_info_id, make_node_id, make_session_id},
};

pub struct DataParser {
    model: Arc<Model>,
    config: DnvConfig,
}

impl DataParser {
    pub fn new(model: Arc<Model>, config: DnvConfig) -> Arc<Self> {
        Arc::new(Self { model, config })
    }

    pub async fn start_connect_slots(self: Arc<Self>, ex: Arc<Executor<'_>>) -> DnetViewResult<()> {
        debug!(target: "dnetview", "start_connect_slots() START");
        for node in &self.config.nodes {
            debug!(target: "dnetview", "attempting to spawn...");
            ex.clone().spawn(self.clone().try_connect(node.clone())).detach();
        }
        Ok(())
    }

    async fn try_connect(self: Arc<Self>, node: Node) -> DnetViewResult<()> {
        debug!(target: "dnetview", "try_connect() START");
        loop {
            info!("Attempting to poll {}, RPC URL: {}", node.name, node.rpc_url);
            // Parse node config and execute poll.
            // On any failure, sleep and retry.
            match RpcConnect::new(Url::parse(&node.rpc_url)?, node.name.clone()).await {
                Ok(client) => {
                    if let Err(e) = self.poll(&node, client).await {
                        error!("Poll execution error: {:?}", e);
                    }
                }
                Err(e) => {
                    error!("RPC client creation error: {:?}", e);
                }
            }
            self.parse_offline(node.name.clone()).await?;
            async_util::sleep(2000).await;
        }
    }

    async fn poll(&self, node: &Node, client: RpcConnect) -> DnetViewResult<()> {
        loop {
            // Ping the node to verify if its online.
            if let Err(e) = client.ping().await {
                return Err(DnetViewError::Darkfi(e))
            }

            // Retrieve node info, based on its type
            let response = match &node.node_type {
                NodeType::LILITH => client.lilith_spawns().await,
                NodeType::NORMAL => client.dnet_info().await,
                NodeType::CONSENSUS => client.get_consensus_info().await,
            };

            // Parse response
            match response {
                Ok(reply) => {
                    debug!("dnetview:: poll() reply {:?}", reply);
                    if reply.as_object().is_none() || reply.as_object().unwrap().is_empty() {
                        return Err(DnetViewError::EmptyRpcReply)
                    }

                    match &node.node_type {
                        NodeType::LILITH => {
                            self.parse_lilith_data(
                                reply.as_object().unwrap().clone(),
                                node.name.clone(),
                            )
                            .await?
                        }
                        _ => self.parse_data(reply.as_object().unwrap(), node.name.clone()).await?,
                    };
                }
                Err(e) => return Err(e),
            }

            // Sleep until next poll
            async_util::sleep(2000).await;
        }
    }

    // If poll times out, inititalize data structures with empty values.
    async fn parse_offline(&self, node_name: String) -> DnetViewResult<()> {
        debug!(target: "dnetview", "parse_offline() START");
        let name = "Offline".to_string();
        let session_type = Session::Offline;

        let mut sessions: Vec<SessionInfo> = Vec::new();
        let hosts = Vec::new();

        let node_id = make_node_id(&node_name)?;
        let dnet_id = make_empty_id(&node_id, &session_type, 0)?;
        let addr = "Null".to_string();
        let state = None;
        let random_id = 0;
        let remote_id = "Null".to_string();
        let log = Vec::new();
        let is_empty = true;

        let slot = SlotInfo::new(
            dnet_id.clone(),
            node_id.clone(),
            addr.clone(),
            random_id,
            remote_id,
            log,
            is_empty,
        );

        let session_info = SessionInfo::new(
            dnet_id,
            node_id.clone(),
            //name.clone(),
            addr.clone(),
            state,
            slot,
            is_empty,
        );
        sessions.push(session_info);

        // TODO: clean this up
        let node = NodeInfo::new(
            node_id.clone(),
            name.clone(),
            hosts,
            sessions.clone(),
            sessions.clone(),
            is_empty,
        );

        self.update_selectables(node).await?;
        Ok(())
    }

    async fn parse_data(
        &self,
        reply: &serde_json::Map<String, Value>,
        name: String,
    ) -> DnetViewResult<()> {
        let hosts = &reply["hosts"];
        let inbound = &reply["inbound"];
        let outbound = &reply["outbound"];

        let node_id = make_node_id(&name)?;

        let hosts = self.parse_hosts(hosts).await?;
        let inbound = self.parse_session(inbound, &node_id, Session::Inbound).await?;
        let outbound = self.parse_session(outbound, &node_id, Session::Outbound).await?;

        let node = NodeInfo::new(node_id, name, hosts, inbound.clone(), outbound.clone(), false);

        self.update_selectables(node).await?;
        self.update_msgs(inbound.clone(), outbound.clone()).await?;

        Ok(())
    }

    async fn parse_lilith_data(
        &self,
        reply: serde_json::Map<String, Value>,
        name: String,
    ) -> DnetViewResult<()> {
        let spawns: Vec<serde_json::Map<String, Value>> =
            serde_json::from_value(reply.get("spawns").unwrap().clone()).unwrap();

        let mut networks = vec![];
        for spawn in spawns {
            let name = spawn.get("name").unwrap().as_str().unwrap().to_string();
            let id = make_node_id(&name)?;
            let urls: Vec<String> =
                serde_json::from_value(spawn.get("urls").unwrap().clone()).unwrap();
            let nodes: Vec<String> =
                serde_json::from_value(spawn.get("hosts").unwrap().clone()).unwrap();
            let network = NetworkInfo::new(id, name, urls, nodes);
            networks.push(network);
        }
        let id = make_node_id(&name)?;
        let lilith = LilithInfo::new(id.clone(), name, networks);
        let lilith_obj = SelectableObject::Lilith(lilith.clone());

        self.model.selectables.lock().await.insert(id, lilith_obj);
        for network in lilith.networks {
            let network_obj = SelectableObject::Network(network.clone());
            self.model.selectables.lock().await.insert(network.id, network_obj);
        }

        Ok(())
    }

    async fn update_msgs(
        &self,
        inbounds: Vec<SessionInfo>,
        outbounds: Vec<SessionInfo>,
    ) -> DnetViewResult<()> {
        for inbound in inbounds {
            if !self.model.msg_map.lock().await.contains_key(&inbound.info.dnet_id) {
                // we don't have this ID: it is a new node
                self.model
                    .msg_map
                    .lock()
                    .await
                    .insert(inbound.info.dnet_id, inbound.info.log.clone());
            } else {
                // we have this id: append the msg values
                match self.model.msg_map.lock().await.entry(inbound.info.dnet_id) {
                    Entry::Vacant(e) => {
                        e.insert(inbound.info.log);
                    }
                    Entry::Occupied(mut e) => {
                        for msg in inbound.info.log {
                            e.get_mut().push(msg);
                        }
                    }
                }
            }
        }
        for outbound in outbounds {
            if !self.model.msg_map.lock().await.contains_key(&outbound.info.dnet_id) {
                // we don't have this ID: it is a new node
                self.model
                    .msg_map
                    .lock()
                    .await
                    .insert(outbound.info.dnet_id, outbound.info.log.clone());
            } else {
                // we have this id: append the msg values
                match self.model.msg_map.lock().await.entry(outbound.info.dnet_id) {
                    Entry::Vacant(e) => {
                        e.insert(outbound.info.log);
                    }
                    Entry::Occupied(mut e) => {
                        for msg in outbound.info.log {
                            e.get_mut().push(msg);
                        }
                    }
                }
            }
        }

        Ok(())
    }

    async fn update_selectables(&self, node: NodeInfo) -> DnetViewResult<()> {
        if node.is_offline {
            let node_obj = SelectableObject::Node(node.clone());
            self.model.selectables.lock().await.insert(node.dnet_id.clone(), node_obj.clone());
        } else {
            let node_obj = SelectableObject::Node(node.clone());
            self.model.selectables.lock().await.insert(node.dnet_id.clone(), node_obj.clone());
            for inbound in node.inbound {
                if !inbound.is_empty {
                    let inbound_obj = SelectableObject::Session(inbound.clone());
                    self.model
                        .selectables
                        .lock()
                        .await
                        .insert(inbound.clone().dnet_id, inbound_obj.clone());
                    let info_obj = SelectableObject::Connect(inbound.info.clone());
                    self.model
                        .selectables
                        .lock()
                        .await
                        .insert(inbound.info.clone().dnet_id, info_obj.clone());
                }
            }
            for outbound in node.outbound {
                if !outbound.is_empty {
                    let outbound_obj = SelectableObject::Session(outbound.clone());
                    self.model
                        .selectables
                        .lock()
                        .await
                        .insert(outbound.clone().dnet_id, outbound_obj.clone());
                    let info_obj = SelectableObject::Connect(outbound.info.clone());
                    self.model
                        .selectables
                        .lock()
                        .await
                        .insert(outbound.info.clone().dnet_id, info_obj.clone());
                }
            }
        }
        Ok(())
    }

    async fn parse_session(
        &self,
        reply: &Value,
        node_id: &String,
        prefix: Session,
    ) -> DnetViewResult<Vec<SessionInfo>> {
        let session_id = make_session_id(&node_id, &prefix)?;
        let mut session_info: Vec<SessionInfo> = Vec::new();

        // TODO: improve this ugly hack.
        let mut slot_count = 0;

        // Dnetview is not enabled.
        if reply.is_null() {
            debug!(target: "dnetview", "parse_outbound() reply.is_null() == True");
            slot_count += 1;
            let info_id = make_empty_id(&node_id, &prefix, slot_count)?;
            let node_id = node_id.to_string();
            let addr = "Null".to_string();
            let random_id = 0;
            let remote_id = "Null".to_string();
            let log = Vec::new();
            let is_empty = false;

            let slot = SlotInfo::new(
                info_id.clone(),
                node_id.clone(),
                addr,
                random_id,
                remote_id,
                log,
                is_empty,
            );
            let is_empty = true;

            let addr = "Null".to_string();
            let state = None;
            let session =
                SessionInfo::new(session_id.clone(), node_id.clone(), addr, state, slot, is_empty);
            session_info.push(session);

            return Ok(session_info)
        }

        let sessions = reply.as_array().unwrap();
        debug!(target: "dnetview", "parse_outbound() len session{:?}", sessions.len());

        for session in sessions {
            match session.as_object() {
                Some(obj) => {
                    debug!(target: "dnetview", "parse_outbound() obj {:?}", session);
                    let addr = obj.get("addr").unwrap().as_str().unwrap().to_string();

                    let state: Option<String> = match obj.get("state") {
                        Some(state) => Some(state.as_str().unwrap().to_string()),
                        None => None,
                    };

                    let info: serde_json::Map<String, Value> =
                        serde_json::from_value(obj.get("info").unwrap().clone()).unwrap();

                    let slot_addr = info.get("addr").unwrap().as_str().unwrap().to_string();
                    let random_id = info.get("random_id").unwrap().as_u64().unwrap();
                    let remote_id = info.get("remote_id").unwrap().as_str().unwrap().to_string();
                    let info_id = make_info_id(&random_id)?;

                    let log: Vec<(NanoTimestamp, String, String)> =
                        serde_json::from_value(info.get("log").unwrap().clone()).unwrap();

                    // ...
                    let node_id = node_id.to_string();
                    let is_empty = false;

                    let slot = SlotInfo::new(
                        info_id.clone(),
                        node_id.clone(),
                        slot_addr,
                        random_id,
                        remote_id,
                        log,
                        is_empty,
                    );

                    let session = SessionInfo::new(
                        session_id.clone(),
                        node_id.clone(),
                        addr.clone(),
                        state,
                        slot,
                        is_empty,
                    );
                    session_info.push(session);
                }
                None => {
                    // TODO: clean up empty info boilerplate.
                    slot_count += 1;
                    let info_id = make_empty_id(node_id, &prefix, slot_count)?;
                    let node_id = node_id.to_string();
                    let addr = "Null".to_string();
                    let random_id = 0;
                    let remote_id = "Null".to_string();
                    let log = Vec::new();
                    let is_empty = true;

                    let slot = SlotInfo::new(
                        info_id.clone(),
                        node_id.clone(),
                        addr.clone(),
                        random_id,
                        remote_id,
                        log,
                        is_empty,
                    );
                    let is_empty = true;

                    let state = None;
                    let session = SessionInfo::new(
                        session_id.clone(),
                        node_id.clone(),
                        addr.clone(),
                        state,
                        slot,
                        is_empty,
                    );
                    session_info.push(session);
                }
            }
        }

        Ok(session_info)
    }

    async fn parse_hosts(&self, hosts: &Value) -> DnetViewResult<Vec<String>> {
        match hosts.as_array() {
            Some(h) => match h.is_empty() {
                true => Ok(Vec::new()),
                false => {
                    let hosts: Vec<String> =
                        h.iter().map(|addr| addr.as_str().unwrap().to_string()).collect();
                    Ok(hosts)
                }
            },

            None => {
                if hosts.is_null() {
                    // TODO: this should probs just say null
                    let h = Vec::new();
                    return Ok(h)
                }
                debug!("dnetview::parse_hosts() hosts returns None and !is_null() {}", hosts);
                Err(DnetViewError::ValueIsNotObject)
            }
        }
    }
}