From c81454e00b08553b678fd04cff1aafc12cdbfe4e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 22 Jul 2021 15:15:08 +0100 Subject: [PATCH] Validator API storage fixes and adjustments --- ... 20210722120000_create_initial_tables.sql} | 12 +- validator-api/sqlx-data.json | 264 ++--- validator-api/src/main.rs | 3 +- validator-api/src/network_monitor/mod.rs | 2 +- .../src/network_monitor/monitor/mod.rs | 2 +- .../monitor/summary_producer.rs | 7 +- validator-api/src/node_status_api/mod.rs | 2 +- validator-api/src/node_status_api/models.rs | 3 +- validator-api/src/node_status_api/routes.rs | 2 +- validator-api/src/node_status_api/storage.rs | 1045 ----------------- validator-api/src/node_status_api/utils.rs | 16 +- validator-api/src/storage/manager.rs | 645 ++++++++++ validator-api/src/storage/mod.rs | 461 ++++++++ validator-api/src/storage/models.rs | 15 + 14 files changed, 1273 insertions(+), 1206 deletions(-) rename validator-api/migrations/{20210714120000_create_initial_tables.sql => 20210722120000_create_initial_tables.sql} (93%) delete mode 100644 validator-api/src/node_status_api/storage.rs create mode 100644 validator-api/src/storage/manager.rs create mode 100644 validator-api/src/storage/mod.rs create mode 100644 validator-api/src/storage/models.rs diff --git a/validator-api/migrations/20210714120000_create_initial_tables.sql b/validator-api/migrations/20210722120000_create_initial_tables.sql similarity index 93% rename from validator-api/migrations/20210714120000_create_initial_tables.sql rename to validator-api/migrations/20210722120000_create_initial_tables.sql index 26e88d07e5..eeafc220db 100644 --- a/validator-api/migrations/20210714120000_create_initial_tables.sql +++ b/validator-api/migrations/20210722120000_create_initial_tables.sql @@ -1,15 +1,15 @@ CREATE TABLE mixnode_details ( id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - owner VARCHAR NOT NULL UNIQUE, - pub_key VARCHAR NOT NULL UNIQUE + owner VARCHAR NOT NULL, + identity VARCHAR NOT NULL UNIQUE ); CREATE TABLE gateway_details ( id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - owner VARCHAR NOT NULL UNIQUE, - pub_key VARCHAR NOT NULL UNIQUE + owner VARCHAR NOT NULL, + identity VARCHAR NOT NULL UNIQUE ); create TABLE mixnode_historical_uptime @@ -87,5 +87,5 @@ INDEX `gateway_ipv4_status_index` ON `gateway_ipv4_status` (`gateway_details_id` CREATE INDEX `gateway_ipv6_status_index` ON `gateway_ipv6_status` (`gateway_details_id`, `timestamp` desc); -CREATE INDEX `mixnode_identity_index` ON `mixnode_details` (`id`, `pub_key`); -CREATE INDEX `gateway_identity_index` ON `gateway_details` (`id`, `pub_key`); \ No newline at end of file +CREATE INDEX `mixnode_identity_index` ON `mixnode_details` (`id`, `identity`); +CREATE INDEX `gateway_identity_index` ON `gateway_details` (`id`, `identity`); \ No newline at end of file diff --git a/validator-api/sqlx-data.json b/validator-api/sqlx-data.json index 124fef609f..3dd9a07a32 100644 --- a/validator-api/sqlx-data.json +++ b/validator-api/sqlx-data.json @@ -1,17 +1,7 @@ { "db": "SQLite", - "01de0a1f0b7c432dfdcad2cc2dedd2c1fa2c9618a0166207a1920e3b25d2b7e7": { - "query": "DELETE FROM mixnode_ipv4_status WHERE timestamp < ?", - "describe": { - "columns": [], - "parameters": { - "Right": 1 - }, - "nullable": [] - } - }, - "1915fb818696abb354f1bc7915399be69fb7b97204f7fb81baba4f4be1777fce": { - "query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM gateway_historical_uptime\n JOIN gateway_details\n ON gateway_historical_uptime.gateway_details_id = gateway_details.id\n WHERE gateway_details.pub_key = ?\n ORDER BY date ASC\n ", + "000bae8722e573327cbd660fc3df5c96d0b6d0f56a872e486557bbbe7e423638": { + "query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM gateway_historical_uptime\n JOIN gateway_details\n ON gateway_historical_uptime.gateway_details_id = gateway_details.id\n WHERE gateway_details.identity = ?\n ORDER BY date ASC\n ", "describe": { "columns": [ { @@ -40,6 +30,16 @@ ] } }, + "01de0a1f0b7c432dfdcad2cc2dedd2c1fa2c9618a0166207a1920e3b25d2b7e7": { + "query": "DELETE FROM mixnode_ipv4_status WHERE timestamp < ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + } + }, "20b7eee88b4bfb196d0dbbbef19d789d9ffd0790349b7ae6bea6b536123be3b3": { "query": "INSERT INTO mixnode_historical_uptime(mixnode_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)", "describe": { @@ -50,6 +50,24 @@ "nullable": [] } }, + "21671ecfb386373f57a260385b6ca5ae0eead07b371805a5b1d37ae0092625f9": { + "query": "SELECT owner FROM mixnode_details WHERE identity = ?", + "describe": { + "columns": [ + { + "name": "owner", + "ordinal": 0, + "type_info": "Text" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + } + }, "26a6cec333e3187d1bee9a583529b4126cb12d711771c5276e9598ccaa0a144d": { "query": "\n INSERT INTO mixnode_ipv4_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?);\n ", "describe": { @@ -70,18 +88,18 @@ "nullable": [] } }, - "2d87cc44c28330492c7cf90589e5d8113ccff07d6f67c485226bbe59246db522": { - "query": "\n INSERT OR IGNORE INTO gateway_details(pub_key, owner) VALUES (?, ?);\n SELECT id FROM gateway_details WHERE pub_key = ?;\n ", + "32bb50f08facfb033a0eeb61b74ce546bd36263f364fe0dfebcd0c9ac8818af6": { + "query": "SELECT owner FROM gateway_details WHERE identity = ?", "describe": { "columns": [ { - "name": "id", + "name": "owner", "ordinal": 0, - "type_info": "Int64" + "type_info": "Text" } ], "parameters": { - "Right": 3 + "Right": 1 }, "nullable": [ false @@ -136,6 +154,36 @@ ] } }, + "49b59f0fcede59a095a47be3acb9035f441d6d533443ae73f85c553b346be356": { + "query": "\n SELECT DISTINCT identity, owner, id\n FROM mixnode_details\n JOIN mixnode_ipv4_status\n ON mixnode_details.id = mixnode_ipv4_status.mixnode_details_id\n WHERE EXISTS (\n SELECT 1 FROM mixnode_ipv4_status WHERE timestamp > ?\n )\n ", + "describe": { + "columns": [ + { + "name": "identity", + "ordinal": 0, + "type_info": "Text" + }, + { + "name": "owner", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "id", + "ordinal": 2, + "type_info": "Int64" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false, + false, + false + ] + } + }, "4c32c31247672549621c5b817625228f1a777919ae711d0d29a192494183e9d3": { "query": "DELETE FROM gateway_ipv6_status WHERE timestamp < ?", "describe": { @@ -156,30 +204,6 @@ "nullable": [] } }, - "6b421289099f59d2f61287e87152621039d47b311dd2474887dcc91c0ca7db7e": { - "query": "\n SELECT timestamp, up\n FROM mixnode_ipv4_status\n JOIN mixnode_details\n ON mixnode_ipv4_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.pub_key=? AND mixnode_ipv4_status.timestamp > ?;\n ", - "describe": { - "columns": [ - { - "name": "timestamp", - "ordinal": 0, - "type_info": "Int64" - }, - { - "name": "up", - "ordinal": 1, - "type_info": "Bool" - } - ], - "parameters": { - "Right": 2 - }, - "nullable": [ - false, - false - ] - } - }, "76c9645a1d79ce4294bff73e0fb60d211490bd810687f8e7d9963d05e7970cee": { "query": "DELETE FROM gateway_ipv4_status WHERE timestamp < ?", "describe": { @@ -190,8 +214,56 @@ "nullable": [] } }, - "9212a4484ce59c30e9ef7d9e6796eb837e00e618db6f21b711cbf17661011054": { - "query": "\n SELECT timestamp, up\n FROM mixnode_ipv6_status\n JOIN mixnode_details\n ON mixnode_ipv6_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.pub_key=? AND mixnode_ipv6_status.timestamp > ?;\n ", + "786cf93daf23b03a5e6ffcf15f03013375be1f94a43aa8b614c9240d7dee6da4": { + "query": "\n INSERT OR IGNORE INTO gateway_details(identity, owner) VALUES (?, ?);\n SELECT id FROM gateway_details WHERE identity = ?;\n ", + "describe": { + "columns": [ + { + "name": "id", + "ordinal": 0, + "type_info": "Int64" + } + ], + "parameters": { + "Right": 3 + }, + "nullable": [ + false + ] + } + }, + "7b37a1a850def3f82ab64c4825af256ef79a6e8a7edad122d4770be8bc84e32a": { + "query": "\n SELECT DISTINCT identity, owner, id\n FROM gateway_details\n JOIN gateway_ipv4_status\n ON gateway_details.id = gateway_ipv4_status.gateway_details_id\n WHERE EXISTS (\n SELECT 1 FROM gateway_ipv4_status WHERE timestamp > ?\n )\n ", + "describe": { + "columns": [ + { + "name": "identity", + "ordinal": 0, + "type_info": "Text" + }, + { + "name": "owner", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "id", + "ordinal": 2, + "type_info": "Int64" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false, + false, + false + ] + } + }, + "8395b08fa17d904217b19c5d03ab4d66dd0a0f648487d81ae09b1f926111d614": { + "query": "\n SELECT timestamp, up\n FROM mixnode_ipv4_status\n JOIN mixnode_details\n ON mixnode_ipv4_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.identity=? AND mixnode_ipv4_status.timestamp > ?;\n ", "describe": { "columns": [ { @@ -248,44 +320,8 @@ "nullable": [] } }, - "97e108e08c4a8aa2c0e2a32f93ae7c65193a26be2d06f8f97ab8fdc0f7004fa4": { - "query": "SELECT owner FROM gateway_details WHERE pub_key = ?", - "describe": { - "columns": [ - { - "name": "owner", - "ordinal": 0, - "type_info": "Text" - } - ], - "parameters": { - "Right": 1 - }, - "nullable": [ - false - ] - } - }, - "9cd914e2f1e8058c6935a50f2a02c0e04c4ee5a1c4a400f08d9d0eb47f5665d8": { - "query": "SELECT owner FROM mixnode_details WHERE pub_key = ?", - "describe": { - "columns": [ - { - "name": "owner", - "ordinal": 0, - "type_info": "Text" - } - ], - "parameters": { - "Right": 1 - }, - "nullable": [ - false - ] - } - }, - "aa9708202a664aa8b85e8881deb9f0137efd91f88eba18a7fa05ee1ad7e8e621": { - "query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM mixnode_historical_uptime\n JOIN mixnode_details\n ON mixnode_historical_uptime.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.pub_key = ?\n ORDER BY date ASC\n ", + "96bf94cf7b81a527718b74bd456587f00f12827c72792eff9346e0b1683feede": { + "query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM mixnode_historical_uptime\n JOIN mixnode_details\n ON mixnode_historical_uptime.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.identity = ?\n ORDER BY date ASC\n ", "describe": { "columns": [ { @@ -314,38 +350,26 @@ ] } }, - "b34e1f8595764d2706e80b317c7aed9ee9a11b23028aa44ff56ce840349e0ff9": { - "query": "\n SELECT DISTINCT pub_key, owner, id\n FROM gateway_details\n JOIN gateway_ipv4_status\n ON gateway_details.id = gateway_ipv4_status.gateway_details_id\n WHERE EXISTS (\n SELECT 1 FROM gateway_ipv4_status WHERE timestamp > ?\n )\n ", + "ba7d4f4134902f6ccec1973c9acfc5052539d35911ba238607ce8ca6111f219d": { + "query": "\n INSERT OR IGNORE INTO mixnode_details(identity, owner) VALUES (?, ?);\n SELECT id FROM mixnode_details WHERE identity = ?;\n ", "describe": { "columns": [ - { - "name": "pub_key", - "ordinal": 0, - "type_info": "Text" - }, - { - "name": "owner", - "ordinal": 1, - "type_info": "Text" - }, { "name": "id", - "ordinal": 2, + "ordinal": 0, "type_info": "Int64" } ], "parameters": { - "Right": 1 + "Right": 3 }, "nullable": [ - false, - false, false ] } }, - "b8e05a65bf3a6f0d4173e0c27cb5491e4f5d1615407c9b46651fcc2dd5970feb": { - "query": "\n SELECT timestamp, up\n FROM gateway_ipv4_status\n JOIN gateway_details\n ON gateway_ipv4_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.pub_key=? AND gateway_ipv4_status.timestamp > ?;\n ", + "bcd2069178c3f006a640983a1bc8d8d0293d163044f30b859b8878a96787333b": { + "query": "\n SELECT timestamp, up\n FROM gateway_ipv4_status\n JOIN gateway_details\n ON gateway_ipv4_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.identity=? AND gateway_ipv4_status.timestamp > ?;\n ", "describe": { "columns": [ { @@ -378,56 +402,32 @@ "nullable": [] } }, - "d378bade0e786e3efdda4a02a5f403da9a23c20f4eecb86fceb18930cf2cceec": { - "query": "\n SELECT DISTINCT pub_key, owner, id\n FROM mixnode_details\n JOIN mixnode_ipv4_status\n ON mixnode_details.id = mixnode_ipv4_status.mixnode_details_id\n WHERE EXISTS (\n SELECT 1 FROM mixnode_ipv4_status WHERE timestamp > ?\n )\n ", + "bfc4a0aec14ac76da012f011afb8ff637666b67d8585f2cbb681520a93f3b472": { + "query": "\n SELECT timestamp, up\n FROM mixnode_ipv6_status\n JOIN mixnode_details\n ON mixnode_ipv6_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.identity=? AND mixnode_ipv6_status.timestamp > ?;\n ", "describe": { "columns": [ { - "name": "pub_key", + "name": "timestamp", "ordinal": 0, - "type_info": "Text" + "type_info": "Int64" }, { - "name": "owner", + "name": "up", "ordinal": 1, - "type_info": "Text" - }, - { - "name": "id", - "ordinal": 2, - "type_info": "Int64" + "type_info": "Bool" } ], "parameters": { - "Right": 1 + "Right": 2 }, "nullable": [ - false, false, false ] } }, - "d4b9a3e22a1230635523aec12a0b5a980e99949232369ebd748c70fed3f4532e": { - "query": "\n INSERT OR IGNORE INTO mixnode_details(pub_key, owner) VALUES (?, ?);\n SELECT id FROM mixnode_details WHERE pub_key = ?;\n ", - "describe": { - "columns": [ - { - "name": "id", - "ordinal": 0, - "type_info": "Int64" - } - ], - "parameters": { - "Right": 3 - }, - "nullable": [ - false - ] - } - }, - "d85a0985f6e0d818217f6733a1df0c9b83821ecf0e824b0a606c3723ad87362a": { - "query": "\n SELECT timestamp, up\n FROM gateway_ipv6_status\n JOIN gateway_details\n ON gateway_ipv6_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.pub_key=? AND gateway_ipv6_status.timestamp > ?;\n ", + "d10fd4560017ab9229567ab3419e4fe6912f61aa4c17ffd87e3f724ec8518023": { + "query": "\n SELECT timestamp, up\n FROM gateway_ipv6_status\n JOIN gateway_details\n ON gateway_ipv6_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.identity=? AND gateway_ipv6_status.timestamp > ?;\n ", "describe": { "columns": [ { diff --git a/validator-api/src/main.rs b/validator-api/src/main.rs index bda832366b..2a1d0b9435 100644 --- a/validator-api/src/main.rs +++ b/validator-api/src/main.rs @@ -8,7 +8,7 @@ use crate::cache::ValidatorCacheRefresher; use crate::config::Config; use crate::network_monitor::new_monitor_runnables; use crate::network_monitor::tested_network::good_topology::parse_topology_file; -use crate::node_status_api::storage::NodeStatusStorage; +use crate::storage::NodeStatusStorage; use ::config::NymConfig; use anyhow::Result; use cache::ValidatorCache; @@ -22,6 +22,7 @@ pub(crate) mod cache; pub(crate) mod config; mod network_monitor; mod node_status_api; +pub(crate) mod storage; const MONITORING_ENABLED: &str = "enable-monitor"; const V4_TOPOLOGY_ARG: &str = "v4-topology-filepath"; diff --git a/validator-api/src/network_monitor/mod.rs b/validator-api/src/network_monitor/mod.rs index 62ff5f43bf..3a289ced15 100644 --- a/validator-api/src/network_monitor/mod.rs +++ b/validator-api/src/network_monitor/mod.rs @@ -14,7 +14,7 @@ use crate::network_monitor::monitor::sender::PacketSender; use crate::network_monitor::monitor::summary_producer::SummaryProducer; use crate::network_monitor::monitor::Monitor; use crate::network_monitor::tested_network::TestedNetwork; -use crate::node_status_api::storage::NodeStatusStorage; +use crate::storage::NodeStatusStorage; use crypto::asymmetric::{encryption, identity}; use futures::channel::mpsc; use nymsphinx::addressing::clients::Recipient; diff --git a/validator-api/src/network_monitor/monitor/mod.rs b/validator-api/src/network_monitor/monitor/mod.rs index 6cdff95610..093ef6fd3a 100644 --- a/validator-api/src/network_monitor/monitor/mod.rs +++ b/validator-api/src/network_monitor/monitor/mod.rs @@ -8,7 +8,7 @@ use crate::network_monitor::monitor::sender::PacketSender; use crate::network_monitor::monitor::summary_producer::{NodeResult, SummaryProducer, TestReport}; use crate::network_monitor::test_packet::NodeType; use crate::network_monitor::tested_network::TestedNetwork; -use crate::node_status_api::storage::NodeStatusStorage; +use crate::storage::NodeStatusStorage; use log::{debug, info}; use std::process; use tokio::time::{sleep, Duration, Instant}; diff --git a/validator-api/src/network_monitor/monitor/summary_producer.rs b/validator-api/src/network_monitor/monitor/summary_producer.rs index 1d72510161..c8041c86fc 100644 --- a/validator-api/src/network_monitor/monitor/summary_producer.rs +++ b/validator-api/src/network_monitor/monitor/summary_producer.rs @@ -6,8 +6,9 @@ use crate::network_monitor::test_packet::{NodeType, TestPacket}; use crate::PENALISE_OUTDATED; use std::collections::HashMap; +#[derive(Debug)] pub(crate) struct NodeResult { - pub(crate) pub_key: String, + pub(crate) identity: String, pub(crate) owner: String, pub(crate) working_ipv4: bool, pub(crate) working_ipv6: bool, @@ -20,9 +21,9 @@ struct NodeStatus { } impl NodeStatus { - fn into_node_status(self, pub_key: String, owner: String) -> NodeResult { + fn into_node_status(self, identity: String, owner: String) -> NodeResult { NodeResult { - pub_key, + identity, owner, working_ipv4: self.ip_v4_compatible, working_ipv6: self.ip_v6_compatible, diff --git a/validator-api/src/node_status_api/mod.rs b/validator-api/src/node_status_api/mod.rs index aae4e45f6f..91fe9bbb83 100644 --- a/validator-api/src/node_status_api/mod.rs +++ b/validator-api/src/node_status_api/mod.rs @@ -1,6 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::storage; use rocket::fairing::AdHoc; use std::path::PathBuf; use std::time::Duration; @@ -8,7 +9,6 @@ use std::time::Duration; pub(crate) mod local_guard; pub(crate) mod models; pub(crate) mod routes; -pub(crate) mod storage; pub(crate) mod utils; pub(crate) const FIFTEEN_MINUTES: Duration = Duration::from_secs(900); diff --git a/validator-api/src/node_status_api/models.rs b/validator-api/src/node_status_api/models.rs index 32bc58cb77..742b3c0a78 100644 --- a/validator-api/src/node_status_api/models.rs +++ b/validator-api/src/node_status_api/models.rs @@ -1,7 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::node_status_api::utils::{NodeStatus, NodeUptimes}; +use crate::node_status_api::utils::NodeUptimes; use rocket::http::{ContentType, Status}; use rocket::response::{self, Responder, Response}; use rocket::Request; @@ -9,6 +9,7 @@ use serde::{Deserialize, Serialize}; use std::convert::TryFrom; use std::fmt::{self, Display, Formatter}; use std::io::Cursor; +use crate::storage::models::NodeStatus; // todo: put into some error enum #[derive(Debug)] diff --git a/validator-api/src/node_status_api/routes.rs b/validator-api/src/node_status_api/routes.rs index dedfac1d6c..63ca4d572e 100644 --- a/validator-api/src/node_status_api/routes.rs +++ b/validator-api/src/node_status_api/routes.rs @@ -6,7 +6,7 @@ use crate::node_status_api::models::{ ErrorResponse, GatewayStatusReport, GatewayUptimeHistory, MixnodeStatusReport, MixnodeUptimeHistory, }; -use crate::node_status_api::storage::NodeStatusStorage; +use crate::storage::NodeStatusStorage; use rocket::http::Status; use rocket::serde::json::Json; use rocket::State; diff --git a/validator-api/src/node_status_api/storage.rs b/validator-api/src/node_status_api/storage.rs deleted file mode 100644 index 7e9761c133..0000000000 --- a/validator-api/src/node_status_api/storage.rs +++ /dev/null @@ -1,1045 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::network_monitor::monitor::summary_producer::NodeResult; -use crate::node_status_api::models::{ - GatewayStatusReport, GatewayUptimeHistory, HistoricalUptime, MixnodeStatusReport, - MixnodeUptimeHistory, NodeStatusApiError, Uptime, -}; -use crate::node_status_api::utils::{ActiveNode, ActiveNodeDayStatuses, NodeStatus}; -use crate::node_status_api::ONE_DAY; -use rocket::fairing::{self, AdHoc}; -use rocket::{Build, Rocket}; -use sqlx::types::time::OffsetDateTime; -use sqlx::ConnectOptions; -use std::convert::TryFrom; -use std::path::PathBuf; - -// A type alias to be more explicit about type of timestamp used. -type UnixTimestamp = i64; - -// note that clone here is fine as upon cloning the same underlying pool will be used -// -// note2: the reason 'inner' was introduced was so that there would be an explicit split to -// where pure SQL is used (i.e. `Inner` should be the only place containing any sort -// of SQL while `NodeStatusStorage` should provide a slightly higher level API) -#[derive(Clone)] -pub(crate) struct NodeStatusStorage { - inner: NodeStatusStorageInner, -} - -#[derive(Clone)] -struct NodeStatusStorageInner { - connection_pool: sqlx::SqlitePool, -} - -impl NodeStatusStorage { - async fn init(rocket: Rocket, database_path: PathBuf) -> fairing::Result { - // TODO: we can inject here more stuff based on our validator-api global config - // struct. Maybe different pool size or timeout intervals? - let mut opts = sqlx::sqlite::SqliteConnectOptions::new() - .filename(&database_path) - .create_if_missing(true); - - // TODO: do we want auto_vacuum ? - - opts.disable_statement_logging(); - - let connection_pool = match sqlx::SqlitePool::connect_with(opts).await { - Ok(db) => db, - Err(e) => { - error!("Failed to connect to SQLx database: {}", e); - return Err(rocket); - } - }; - - if let Err(e) = sqlx::migrate!("./migrations").run(&connection_pool).await { - error!("Failed to initialize SQLx database: {}", e); - return Err(rocket); - } - - info!("Database migration finished!"); - - let storage = NodeStatusStorage { - inner: NodeStatusStorageInner { connection_pool }, - }; - - Ok(rocket.manage(storage)) - } - - pub(crate) fn stage(database_path: PathBuf) -> AdHoc { - AdHoc::try_on_ignite("SQLx Database", |rocket| { - NodeStatusStorage::init(rocket, database_path) - }) - } - - /// Gets all statuses for particular mixnode (ipv4 and ipv6) that were inserted in last 24h. - async fn get_mixnode_daily_statuses( - &self, - identity: &str, - ) -> Result<(Vec, Vec), NodeStatusApiError> { - let now = OffsetDateTime::now_utc(); - let day_ago = now - ONE_DAY; - - let ipv4_statuses = self - .inner - .get_mixnode_ipv4_statuses_since(identity, day_ago.unix_timestamp()) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - let ipv6_statuses = self - .inner - .get_mixnode_ipv6_statuses_since(identity, day_ago.unix_timestamp()) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - Ok((ipv4_statuses, ipv6_statuses)) - } - - /// Gets all statuses for particular gateway (ipv4 and ipv6) that were inserted in last 24h. - async fn get_gateway_daily_statuses( - &self, - identity: &str, - ) -> Result<(Vec, Vec), NodeStatusApiError> { - let now = OffsetDateTime::now_utc(); - let day_ago = now - ONE_DAY; - - let ipv4_statuses = self - .inner - .get_gateway_ipv4_statuses_since(identity, day_ago.unix_timestamp()) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - let ipv6_statuses = self - .inner - .get_gateway_ipv6_statuses_since(identity, day_ago.unix_timestamp()) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - Ok((ipv4_statuses, ipv6_statuses)) - } - - /// Tries to construct a status report for mixnode with the specified identity. - pub(crate) async fn construct_mixnode_report( - &self, - identity: &str, - ) -> Result { - let (ipv4_statuses, ipv6_statuses) = self.get_mixnode_daily_statuses(identity).await?; - - // if we have no statuses, the node doesn't exist (or monitor is down), but either way, we can't make a report - if ipv4_statuses.is_empty() { - return Err(NodeStatusApiError::MixnodeReportNotFound( - identity.to_owned(), - )); - } - - // now, technically this is not a critical error, but this should have NEVER happened in the first place - // so something super weird is going on - if ipv4_statuses.len() != ipv6_statuses.len() { - error!("Somehow we have different number of ipv4 and ipv6 statuses for mixnode {}! (ipv4: {}, ipv6: {})", - identity, - ipv4_statuses.len(), - ipv6_statuses.len(), - ) - } - - let mixnode_owner = self - .inner - .get_mixnode_owner(identity) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)? - .expect("The node doesn't have an owner even though we have status information on it!"); - - Ok(MixnodeStatusReport::construct_from_last_day_reports( - identity.to_owned(), - mixnode_owner, - ipv4_statuses, - ipv6_statuses, - )) - } - - pub(crate) async fn construct_gateway_report( - &self, - identity: &str, - ) -> Result { - let (ipv4_statuses, ipv6_statuses) = self.get_gateway_daily_statuses(identity).await?; - - // if we have no statuses, the node doesn't exist (or monitor is down), but either way, we can't make a report - if ipv4_statuses.is_empty() { - return Err(NodeStatusApiError::GatewayReportNotFound( - identity.to_owned(), - )); - } - - // now, technically this is not a critical error, but this should have NEVER happened in the first place - // so something super weird is going on - if ipv4_statuses.len() != ipv6_statuses.len() { - error!("Somehow we have different number of ipv4 and ipv6 statuses for gateway {}! (ipv4: {}, ipv6: {})", - identity, - ipv4_statuses.len(), - ipv6_statuses.len(), - ) - } - - let gateway_owner = self - .inner - .get_gateway_owner(identity) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)? - .expect( - "The gateway doesn't have an owner even though we have status information on it!", - ); - - Ok(GatewayStatusReport::construct_from_last_day_reports( - identity.to_owned(), - gateway_owner, - ipv4_statuses, - ipv6_statuses, - )) - } - - pub(crate) async fn get_mixnode_uptime_history( - &self, - identity: &str, - ) -> Result { - let history = self - .inner - .get_mixnode_historical_uptimes(identity) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - if history.is_empty() { - return Err(NodeStatusApiError::MixnodeUptimeHistoryNotFound( - identity.to_owned(), - )); - } - - let mixnode_owner = self - .inner - .get_mixnode_owner(identity) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)? - .expect("The node doesn't have an owner even though we have uptime history for it!"); - - Ok(MixnodeUptimeHistory::new( - identity.to_owned(), - mixnode_owner, - history, - )) - } - - pub(crate) async fn get_gateway_uptime_history( - &self, - identity: &str, - ) -> Result { - let history = self - .inner - .get_gateway_historical_uptimes(identity) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - if history.is_empty() { - return Err(NodeStatusApiError::GatewayUptimeHistoryNotFound( - identity.to_owned(), - )); - } - - let gateway_owner = self - .inner - .get_gateway_owner(identity) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)? - .expect("The gateway doesn't have an owner even though we have uptime history for it!"); - - Ok(GatewayUptimeHistory::new( - identity.to_owned(), - gateway_owner, - history, - )) - } - - // NOTE: this method will go away once we move payments into the validator-api - // it just helps us to get rid of having to query for reports of each node individually - pub(crate) async fn get_all_mixnode_reports( - &self, - ) -> Result, NodeStatusApiError> { - let reports = self - .inner - .get_all_active_mixnodes_statuses() - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)? - .into_iter() - .map(|statuses| { - MixnodeStatusReport::construct_from_last_day_reports( - statuses.pub_key, - statuses.owner, - statuses.ipv4_statuses, - statuses.ipv6_statuses, - ) - }) - .collect(); - - Ok(reports) - } - - // NOTE: this method will go away once we move payments into the validator-api - // it just helps us to get rid of having to query for reports of each node individually - pub(crate) async fn get_all_gateway_reports( - &self, - ) -> Result, NodeStatusApiError> { - let reports = self - .inner - .get_all_active_gateways_statuses() - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)? - .into_iter() - .map(|statuses| { - GatewayStatusReport::construct_from_last_day_reports( - statuses.pub_key, - statuses.owner, - statuses.ipv4_statuses, - statuses.ipv6_statuses, - ) - }) - .collect(); - - Ok(reports) - } - - // Used by network monitor - pub(crate) async fn submit_new_statuses( - &self, - mixnode_results: Vec, - gateway_results: Vec, - ) -> Result<(), NodeStatusApiError> { - self.inner - .submit_new_statuses(mixnode_results, gateway_results) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError) - } - - // Called on timer/reward script - async fn update_historical_uptimes( - &self, - today_iso_8601: &str, - ) -> Result<(), NodeStatusApiError> { - // get statuses for all active mixnodes... - let active_mixnodes_statuses = self - .inner - .get_all_active_mixnodes_statuses() - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - for statuses in active_mixnodes_statuses.into_iter() { - let ipv4_day_up = statuses - .ipv4_statuses - .iter() - .filter(|status| status.up) - .count(); - let ipv6_day_up = statuses - .ipv6_statuses - .iter() - .filter(|status| status.up) - .count(); - - // calculate their uptimes for the last 24h - let ipv4_uptime = Uptime::from_ratio(ipv4_day_up, statuses.ipv4_statuses.len()) - .unwrap() - .u8(); - let ipv6_uptime = Uptime::from_ratio(ipv6_day_up, statuses.ipv6_statuses.len()) - .unwrap() - .u8(); - - // and insert into the database - self.inner - .insert_mixnode_historical_uptime( - statuses.node_id, - today_iso_8601, - ipv4_uptime, - ipv6_uptime, - ) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - } - - // get statuses for all active gateways... - let active_gateways_statuses = self - .inner - .get_all_active_gateways_statuses() - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - - for statuses in active_gateways_statuses.into_iter() { - let ipv4_day_up = statuses - .ipv4_statuses - .iter() - .filter(|status| status.up) - .count(); - let ipv6_day_up = statuses - .ipv6_statuses - .iter() - .filter(|status| status.up) - .count(); - - // calculate their uptimes for the last 24h - let ipv4_uptime = Uptime::from_ratio(ipv4_day_up, statuses.ipv4_statuses.len()) - .unwrap() - .u8(); - let ipv6_uptime = Uptime::from_ratio(ipv6_day_up, statuses.ipv6_statuses.len()) - .unwrap() - .u8(); - - // and insert into the database - self.inner - .insert_gateway_historical_uptime( - statuses.node_id, - today_iso_8601, - ipv4_uptime, - ipv6_uptime, - ) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; - } - - Ok(()) - } - - async fn check_if_historical_uptimes_exist_for_date( - &self, - date_iso_8601: &str, - ) -> Result { - self.inner - .check_for_historical_uptime_existence(date_iso_8601) - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError) - } - - // Called on timer/reward script - async fn purge_old_statuses(&self) -> Result<(), NodeStatusApiError> { - self.inner - .purge_old_statuses() - .await - .map_err(|_| NodeStatusApiError::InternalDatabaseError) - } - - pub(crate) async fn daily_chores(&self) -> Result { - let today_iso_8601 = OffsetDateTime::now_utc().date().to_string(); - - // if we have already performed the update for today's date, don't do anything - if self - .check_if_historical_uptimes_exist_for_date(&today_iso_8601) - .await? - { - Ok(false) - } else { - info!( - "Updating historical daily uptimes of all nodes and purging old status reports..." - ); - self.update_historical_uptimes(&today_iso_8601).await?; - self.purge_old_statuses().await?; - Ok(true) - } - } -} - -// all SQL goes here -impl NodeStatusStorageInner { - /// Tries to obtain owner value of given mixnode given its identity - async fn get_mixnode_owner(&self, identity: &str) -> Result, sqlx::Error> { - let owner = sqlx::query!( - "SELECT owner FROM mixnode_details WHERE pub_key = ?", - identity - ) - .fetch_optional(&self.connection_pool) - .await? - .map(|row| row.owner); - - Ok(owner) - } - - /// Tries to obtain owner value of given gateway given its identity - async fn get_gateway_owner(&self, identity: &str) -> Result, sqlx::Error> { - let owner = sqlx::query!( - "SELECT owner FROM gateway_details WHERE pub_key = ?", - identity - ) - .fetch_optional(&self.connection_pool) - .await? - .map(|row| row.owner); - - Ok(owner) - } - - /// Gets all ipv4 statuses for mixnode with particular identity that were inserted - /// into the database after the specified unix timestamp. - async fn get_mixnode_ipv4_statuses_since( - &self, - identity: &str, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM mixnode_ipv4_status - JOIN mixnode_details - ON mixnode_ipv4_status.mixnode_details_id = mixnode_details.id - WHERE mixnode_details.pub_key=? AND mixnode_ipv4_status.timestamp > ?; - "#, - identity, - timestamp, - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets all ipv6 statuses for mixnode with particular identity that were inserted - /// into the database after the specified unix timestamp. - async fn get_mixnode_ipv6_statuses_since( - &self, - identity: &str, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM mixnode_ipv6_status - JOIN mixnode_details - ON mixnode_ipv6_status.mixnode_details_id = mixnode_details.id - WHERE mixnode_details.pub_key=? AND mixnode_ipv6_status.timestamp > ?; - "#, - identity, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets all ipv4 statuses for gateway with particular identity that were inserted - /// into the database after the specified unix timestamp. - async fn get_gateway_ipv4_statuses_since( - &self, - identity: &str, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM gateway_ipv4_status - JOIN gateway_details - ON gateway_ipv4_status.gateway_details_id = gateway_details.id - WHERE gateway_details.pub_key=? AND gateway_ipv4_status.timestamp > ?; - "#, - identity, - timestamp, - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets all ipv6 statuses for gateway with particular identity that were inserted - /// into the database after the specified unix timestamp. - async fn get_gateway_ipv6_statuses_since( - &self, - identity: &str, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM gateway_ipv6_status - JOIN gateway_details - ON gateway_ipv6_status.gateway_details_id = gateway_details.id - WHERE gateway_details.pub_key=? AND gateway_ipv6_status.timestamp > ?; - "#, - identity, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets the historical daily uptime associated with the particular mixnode - async fn get_mixnode_historical_uptimes( - &self, - identity: &str, - ) -> Result, sqlx::Error> { - let uptimes = sqlx::query!( - r#" - SELECT date, ipv4_uptime, ipv6_uptime - FROM mixnode_historical_uptime - JOIN mixnode_details - ON mixnode_historical_uptime.mixnode_details_id = mixnode_details.id - WHERE mixnode_details.pub_key = ? - ORDER BY date ASC - "#, - identity - ) - .fetch_all(&self.connection_pool) - .await? - .into_iter() - // filter out nodes with valid uptime (in theory all should be 100% valid since we insert them ourselves, but - // better safe than sorry and not use an unwrap) - .filter_map(|row| { - Uptime::try_from(row.ipv4_uptime) - .ok() - .map(|ipv4_uptime| { - Uptime::try_from(row.ipv6_uptime) - .ok() - .map(|ipv6_uptime| HistoricalUptime { - date: row.date, - ipv4_uptime, - ipv6_uptime, - }) - }) - .flatten() - }) - .collect(); - - Ok(uptimes) - } - - /// Gets the historical daily uptime associated with the particular gateway - async fn get_gateway_historical_uptimes( - &self, - identity: &str, - ) -> Result, sqlx::Error> { - let uptimes = sqlx::query!( - r#" - SELECT date, ipv4_uptime, ipv6_uptime - FROM gateway_historical_uptime - JOIN gateway_details - ON gateway_historical_uptime.gateway_details_id = gateway_details.id - WHERE gateway_details.pub_key = ? - ORDER BY date ASC - "#, - identity - ) - .fetch_all(&self.connection_pool) - .await? - .into_iter() - // filter out nodes with valid uptime (in theory all should be 100% valid since we insert them ourselves, but - // better safe than sorry and not use an unwrap) - .filter_map(|row| { - Uptime::try_from(row.ipv4_uptime) - .ok() - .map(|ipv4_uptime| { - Uptime::try_from(row.ipv6_uptime) - .ok() - .map(|ipv6_uptime| HistoricalUptime { - date: row.date, - ipv4_uptime, - ipv6_uptime, - }) - }) - .flatten() - }) - .collect(); - - Ok(uptimes) - } - - // NOTE: this method will go away once we move payments into the validator-api - // it just helps us to get rid of having to query for reports of each node individually - /// Returns public key, owner and id of all mixnodes that have had any ipv4 statuses submitted - /// since provided timestamp. - async fn get_all_active_mixnodes( - &self, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - // find mixnode details of all nodes that have had at least 1 ipv4 status since the provided - // timestamp - // TODO: I dont know if theres a potential issue of if we have a lot of inactive nodes that - // haven't mixed in ages, they might increase the query times? - sqlx::query_as!( - ActiveNode, - r#" - SELECT DISTINCT pub_key, owner, id - FROM mixnode_details - JOIN mixnode_ipv4_status - ON mixnode_details.id = mixnode_ipv4_status.mixnode_details_id - WHERE EXISTS ( - SELECT 1 FROM mixnode_ipv4_status WHERE timestamp > ? - ) - "#, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - // NOTE: this method will go away once we move payments into the validator-api - // it just helps us to get rid of having to query for reports of each node individually - /// Returns public key, owner and id of all gateways that have had any ipv4 statuses submitted - /// since provided timestamp. - async fn get_all_active_gateways( - &self, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - ActiveNode, - r#" - SELECT DISTINCT pub_key, owner, id - FROM gateway_details - JOIN gateway_ipv4_status - ON gateway_details.id = gateway_ipv4_status.gateway_details_id - WHERE EXISTS ( - SELECT 1 FROM gateway_ipv4_status WHERE timestamp > ? - ) - "#, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets all ipv4 statuses for mixnode with particular id that were inserted - /// into the database after the specified unix timestamp. - async fn get_mixnode_ipv4_statuses_since_by_id( - &self, - id: i64, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM mixnode_ipv4_status - WHERE mixnode_details_id=? AND timestamp > ?; - "#, - id, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets all ipv6 statuses for mixnode with particular id that were inserted - /// into the database after the specified unix timestamp. - async fn get_mixnode_ipv6_statuses_since_by_id( - &self, - id: i64, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM mixnode_ipv6_status - WHERE mixnode_details_id=? AND timestamp > ?; - "#, - id, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets all ipv4 statuses for gateway with particular id that were inserted - /// into the database after the specified unix timestamp. - async fn get_gateway_ipv4_statuses_since_by_id( - &self, - id: i64, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM gateway_ipv4_status - WHERE gateway_details_id=? AND timestamp > ?; - "#, - id, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - /// Gets all ipv6 statuses for gateway with particular id that were inserted - /// into the database after the specified unix timestamp. - async fn get_gateway_ipv6_statuses_since_by_id( - &self, - id: i64, - timestamp: UnixTimestamp, - ) -> Result, sqlx::Error> { - sqlx::query_as!( - NodeStatus, - r#" - SELECT timestamp, up - FROM gateway_ipv6_status - WHERE gateway_details_id=? AND timestamp > ?; - "#, - id, - timestamp - ) - .fetch_all(&self.connection_pool) - .await - } - - // NOTE: this method will go away once we move payments into the validator-api - // it just helps us to get rid of having to query for reports of each node individually - // TODO: should that live on the 'Inner' struct or should it rather exist on the actual storage struct - // since technically it doesn't touch any SQL directly - async fn get_all_active_mixnodes_statuses( - &self, - ) -> Result, sqlx::Error> { - let now = OffsetDateTime::now_utc(); - let day_ago = (now - ONE_DAY).unix_timestamp(); - - let active_nodes = self.get_all_active_mixnodes(day_ago).await?; - - let mut active_day_statuses = Vec::with_capacity(active_nodes.len()); - for active_node in active_nodes.into_iter() { - let ipv4_statuses = self - .get_mixnode_ipv4_statuses_since_by_id(active_node.id, day_ago) - .await?; - let ipv6_statuses = self - .get_mixnode_ipv6_statuses_since_by_id(active_node.id, day_ago) - .await?; - - let statuses = ActiveNodeDayStatuses { - pub_key: active_node.pub_key, - owner: active_node.owner, - node_id: active_node.id, - ipv4_statuses, - ipv6_statuses, - }; - - active_day_statuses.push(statuses); - } - - Ok(active_day_statuses) - } - - // NOTE: this method will go away once we move payments into the validator-api - // it just helps us to get rid of having to query for reports of each node individually - // TODO: should that live on the 'Inner' struct or should it rather exist on the actual storage struct - // since technically it doesn't touch any SQL directly - async fn get_all_active_gateways_statuses( - &self, - ) -> Result, sqlx::Error> { - let now = OffsetDateTime::now_utc(); - let day_ago = (now - ONE_DAY).unix_timestamp(); - - let active_nodes = self.get_all_active_gateways(day_ago).await?; - - let mut active_day_statuses = Vec::with_capacity(active_nodes.len()); - for active_node in active_nodes.into_iter() { - let ipv4_statuses = self - .get_gateway_ipv4_statuses_since_by_id(active_node.id, day_ago) - .await?; - let ipv6_statuses = self - .get_gateway_ipv6_statuses_since_by_id(active_node.id, day_ago) - .await?; - - let statuses = ActiveNodeDayStatuses { - pub_key: active_node.pub_key, - owner: active_node.owner, - node_id: active_node.id, - ipv4_statuses, - ipv6_statuses, - }; - - active_day_statuses.push(statuses); - } - - Ok(active_day_statuses) - } - - /// Tries to submit [`NodeResult`] from the network monitor to the database. - async fn submit_new_statuses( - &self, - mixnode_results: Vec, - gateway_results: Vec, - ) -> Result<(), sqlx::Error> { - // TODO: lower that to debug before creating PR - info!("Submitting new node results to the database. There are {} mixnode results and {} gateway results", mixnode_results.len(), gateway_results.len()); - - let now = OffsetDateTime::now_utc().unix_timestamp(); - - // insert it all in a transaction to make sure all nodes are updated at the same time - // (plus it's a nice guard against new nodes) - let mut tx = self.connection_pool.begin().await?; - for mixnode_result in mixnode_results { - // if mixnode info doesn't exist, insert it and get its id - - // TODO: the potential "problem" (if you can call it that way) is that if entry DID exist - // then the id field will be incremented for the next node we create thus we will - // have gaps in our ids. ask @DH if that's fine (I don't see why not because nodes - // are still correctly ordered and you can get their total number with a simple query - // and we'd have to run the system until the heat death of the universe to run out of id numbers) - let mixnode_id = sqlx::query!( - r#" - INSERT OR IGNORE INTO mixnode_details(pub_key, owner) VALUES (?, ?); - SELECT id FROM mixnode_details WHERE pub_key = ?; - "#, - mixnode_result.pub_key, - mixnode_result.owner, - mixnode_result.pub_key, - ) - .fetch_one(&mut tx) - .await? - .id; - - // insert ipv4 status - sqlx::query!( - r#" - INSERT INTO mixnode_ipv4_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?); - "#, - mixnode_id, - mixnode_result.working_ipv4, - now - ) - .execute(&mut tx) - .await?; - - // insert ipv6 status - sqlx::query!( - r#" - INSERT INTO mixnode_ipv6_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?); - "#, - mixnode_id, - mixnode_result.working_ipv6, - now - ) - .execute(&mut tx) - .await?; - } - - // repeat the procedure for gateways - for gateway_result in gateway_results { - // if gateway info doesn't exist, insert it and get its id - - // same ID "problem" as described for mixnode insertion - let gateway_id = sqlx::query!( - r#" - INSERT OR IGNORE INTO gateway_details(pub_key, owner) VALUES (?, ?); - SELECT id FROM gateway_details WHERE pub_key = ?; - "#, - gateway_result.pub_key, - gateway_result.owner, - gateway_result.pub_key, - ) - .fetch_one(&mut tx) - .await? - .id; - - // insert ipv4 status - sqlx::query!( - r#" - INSERT INTO gateway_ipv4_status (gateway_details_id, up, timestamp) VALUES (?, ?, ?); - "#, - gateway_id, - gateway_result.working_ipv4, - now - ) - .execute(&mut tx) - .await?; - - // insert ipv6 status - sqlx::query!( - r#" - INSERT INTO gateway_ipv6_status (gateway_details_id, up, timestamp) VALUES (?, ?, ?); - "#, - gateway_id, - gateway_result.working_ipv6, - now - ) - .execute(&mut tx) - .await?; - } - - // finally commit the transaction - tx.commit().await - } - - /// Checks whether there are already any historical uptimes with this particular date. - async fn check_for_historical_uptime_existence( - &self, - today_iso_8601: &str, - ) -> Result { - sqlx::query!( - "SELECT EXISTS (SELECT 1 FROM mixnode_historical_uptime WHERE date = ?) AS 'exists'", - today_iso_8601 - ) - .fetch_one(&self.connection_pool) - .await - .map(|result| result.exists == 1) - } - - /// Creates new entry for mixnode historical uptime - async fn insert_mixnode_historical_uptime( - &self, - node_id: i64, - date: &str, - ipv4_uptime: u8, - ipv6_uptime: u8, - ) -> Result<(), sqlx::Error> { - sqlx::query!( - "INSERT INTO mixnode_historical_uptime(mixnode_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)", - node_id, - date, - ipv4_uptime, - ipv6_uptime, - ).execute(&self.connection_pool).await?; - Ok(()) - } - - /// Creates new entry for gatewy historical uptime - async fn insert_gateway_historical_uptime( - &self, - node_id: i64, - date: &str, - ipv4_uptime: u8, - ipv6_uptime: u8, - ) -> Result<(), sqlx::Error> { - sqlx::query!( - "INSERT INTO gateway_historical_uptime(gateway_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)", - node_id, - date, - ipv4_uptime, - ipv6_uptime, - ).execute(&self.connection_pool).await?; - Ok(()) - } - - /// Removes all statuses from the database that are older than 48h. - async fn purge_old_statuses(&self) -> Result<(), sqlx::Error> { - let now = OffsetDateTime::now_utc(); - let two_days_ago = (now - 2 * ONE_DAY).unix_timestamp(); - - sqlx::query!( - "DELETE FROM mixnode_ipv4_status WHERE timestamp < ?", - two_days_ago - ) - .execute(&self.connection_pool) - .await?; - - sqlx::query!( - "DELETE FROM mixnode_ipv6_status WHERE timestamp < ?", - two_days_ago - ) - .execute(&self.connection_pool) - .await?; - - sqlx::query!( - "DELETE FROM gateway_ipv4_status WHERE timestamp < ?", - two_days_ago - ) - .execute(&self.connection_pool) - .await?; - - sqlx::query!( - "DELETE FROM gateway_ipv6_status WHERE timestamp < ?", - two_days_ago - ) - .execute(&self.connection_pool) - .await?; - - Ok(()) - } -} diff --git a/validator-api/src/node_status_api/utils.rs b/validator-api/src/node_status_api/utils.rs index a173089451..a7d4f2d52d 100644 --- a/validator-api/src/node_status_api/utils.rs +++ b/validator-api/src/node_status_api/utils.rs @@ -3,24 +3,12 @@ use crate::node_status_api::models::Uptime; use crate::node_status_api::{FIFTEEN_MINUTES, ONE_HOUR}; +use crate::storage::models::NodeStatus; use sqlx::types::time::OffsetDateTime; -// Internally used struct to catch results from the database to calculate uptimes for given mixnode/gateway -pub(crate) struct NodeStatus { - pub(crate) timestamp: i64, - pub(crate) up: bool, -} - -// Internally used struct to catch results from the database to find active mixnodes/gateways -pub(crate) struct ActiveNode { - pub(crate) id: i64, - pub(crate) pub_key: String, - pub(crate) owner: String, -} - // A temporary helper struct used to produce reports for active nodes. pub(crate) struct ActiveNodeDayStatuses { - pub(crate) pub_key: String, + pub(crate) identity: String, pub(crate) owner: String, pub(crate) node_id: i64, diff --git a/validator-api/src/storage/manager.rs b/validator-api/src/storage/manager.rs new file mode 100644 index 0000000000..c95dbfbb0c --- /dev/null +++ b/validator-api/src/storage/manager.rs @@ -0,0 +1,645 @@ +// Copyright 2021 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::network_monitor::monitor::summary_producer::NodeResult; +use crate::node_status_api::models::{HistoricalUptime, Uptime}; +use crate::node_status_api::utils::ActiveNodeDayStatuses; +use crate::node_status_api::ONE_DAY; +use crate::storage::models::{ActiveNode, NodeStatus}; +use crate::storage::UnixTimestamp; +use sqlx::types::time::OffsetDateTime; +use std::convert::TryFrom; + +#[derive(Clone)] +pub(crate) struct StorageManager { + pub(super) connection_pool: sqlx::SqlitePool, +} + +// all SQL goes here +impl StorageManager { + /// Tries to obtain owner value of given mixnode given its identity + pub(crate) async fn get_mixnode_owner( + &self, + identity: &str, + ) -> Result, sqlx::Error> { + let owner = sqlx::query!( + "SELECT owner FROM mixnode_details WHERE identity = ?", + identity + ) + .fetch_optional(&self.connection_pool) + .await? + .map(|row| row.owner); + + Ok(owner) + } + + /// Tries to obtain owner value of given gateway given its identity + pub(crate) async fn get_gateway_owner( + &self, + identity: &str, + ) -> Result, sqlx::Error> { + let owner = sqlx::query!( + "SELECT owner FROM gateway_details WHERE identity = ?", + identity + ) + .fetch_optional(&self.connection_pool) + .await? + .map(|row| row.owner); + + Ok(owner) + } + + /// Gets all ipv4 statuses for mixnode with particular identity that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_mixnode_ipv4_statuses_since( + &self, + identity: &str, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM mixnode_ipv4_status + JOIN mixnode_details + ON mixnode_ipv4_status.mixnode_details_id = mixnode_details.id + WHERE mixnode_details.identity=? AND mixnode_ipv4_status.timestamp > ?; + "#, + identity, + timestamp, + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Gets all ipv6 statuses for mixnode with particular identity that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_mixnode_ipv6_statuses_since( + &self, + identity: &str, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM mixnode_ipv6_status + JOIN mixnode_details + ON mixnode_ipv6_status.mixnode_details_id = mixnode_details.id + WHERE mixnode_details.identity=? AND mixnode_ipv6_status.timestamp > ?; + "#, + identity, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Gets all ipv4 statuses for gateway with particular identity that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_gateway_ipv4_statuses_since( + &self, + identity: &str, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM gateway_ipv4_status + JOIN gateway_details + ON gateway_ipv4_status.gateway_details_id = gateway_details.id + WHERE gateway_details.identity=? AND gateway_ipv4_status.timestamp > ?; + "#, + identity, + timestamp, + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Gets all ipv6 statuses for gateway with particular identity that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_gateway_ipv6_statuses_since( + &self, + identity: &str, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM gateway_ipv6_status + JOIN gateway_details + ON gateway_ipv6_status.gateway_details_id = gateway_details.id + WHERE gateway_details.identity=? AND gateway_ipv6_status.timestamp > ?; + "#, + identity, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Gets the historical daily uptime associated with the particular mixnode + pub(crate) async fn get_mixnode_historical_uptimes( + &self, + identity: &str, + ) -> Result, sqlx::Error> { + let uptimes = sqlx::query!( + r#" + SELECT date, ipv4_uptime, ipv6_uptime + FROM mixnode_historical_uptime + JOIN mixnode_details + ON mixnode_historical_uptime.mixnode_details_id = mixnode_details.id + WHERE mixnode_details.identity = ? + ORDER BY date ASC + "#, + identity + ) + .fetch_all(&self.connection_pool) + .await? + .into_iter() + // filter out nodes with valid uptime (in theory all should be 100% valid since we insert them ourselves, but + // better safe than sorry and not use an unwrap) + .filter_map(|row| { + Uptime::try_from(row.ipv4_uptime) + .ok() + .map(|ipv4_uptime| { + Uptime::try_from(row.ipv6_uptime) + .ok() + .map(|ipv6_uptime| HistoricalUptime { + date: row.date, + ipv4_uptime, + ipv6_uptime, + }) + }) + .flatten() + }) + .collect(); + + Ok(uptimes) + } + + /// Gets the historical daily uptime associated with the particular gateway + pub(crate) async fn get_gateway_historical_uptimes( + &self, + identity: &str, + ) -> Result, sqlx::Error> { + let uptimes = sqlx::query!( + r#" + SELECT date, ipv4_uptime, ipv6_uptime + FROM gateway_historical_uptime + JOIN gateway_details + ON gateway_historical_uptime.gateway_details_id = gateway_details.id + WHERE gateway_details.identity = ? + ORDER BY date ASC + "#, + identity + ) + .fetch_all(&self.connection_pool) + .await? + .into_iter() + // filter out nodes with valid uptime (in theory all should be 100% valid since we insert them ourselves, but + // better safe than sorry and not use an unwrap) + .filter_map(|row| { + Uptime::try_from(row.ipv4_uptime) + .ok() + .map(|ipv4_uptime| { + Uptime::try_from(row.ipv6_uptime) + .ok() + .map(|ipv6_uptime| HistoricalUptime { + date: row.date, + ipv4_uptime, + ipv6_uptime, + }) + }) + .flatten() + }) + .collect(); + + Ok(uptimes) + } + + /// Gets all ipv4 statuses for mixnode with particular id that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_mixnode_ipv4_statuses_since_by_id( + &self, + id: i64, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM mixnode_ipv4_status + WHERE mixnode_details_id=? AND timestamp > ?; + "#, + id, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Gets all ipv6 statuses for mixnode with particular id that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_mixnode_ipv6_statuses_since_by_id( + &self, + id: i64, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM mixnode_ipv6_status + WHERE mixnode_details_id=? AND timestamp > ?; + "#, + id, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Gets all ipv4 statuses for gateway with particular id that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_gateway_ipv4_statuses_since_by_id( + &self, + id: i64, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM gateway_ipv4_status + WHERE gateway_details_id=? AND timestamp > ?; + "#, + id, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Gets all ipv6 statuses for gateway with particular id that were inserted + /// into the database after the specified unix timestamp. + pub(crate) async fn get_gateway_ipv6_statuses_since_by_id( + &self, + id: i64, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + NodeStatus, + r#" + SELECT timestamp, up + FROM gateway_ipv6_status + WHERE gateway_details_id=? AND timestamp > ?; + "#, + id, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + /// Tries to submit mixnode [`NodeResult`] from the network monitor to the database. + pub(crate) async fn submit_mixnode_statuses( + &self, + timestamp: UnixTimestamp, + mixnode_results: Vec, + ) -> Result<(), sqlx::Error> { + // insert it all in a transaction to make sure all nodes are updated at the same time + // (plus it's a nice guard against new nodes) + let mut tx = self.connection_pool.begin().await?; + for mixnode_result in mixnode_results { + let mixnode_id = sqlx::query!( + r#" + INSERT OR IGNORE INTO mixnode_details(identity, owner) VALUES (?, ?); + SELECT id FROM mixnode_details WHERE identity = ?; + "#, + mixnode_result.identity, + mixnode_result.owner, + mixnode_result.identity, + ) + .fetch_one(&mut tx) + .await? + .id; + + // insert ipv4 status + sqlx::query!( + r#" + INSERT INTO mixnode_ipv4_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?); + "#, + mixnode_id, + mixnode_result.working_ipv4, + timestamp + ) + .execute(&mut tx) + .await?; + + // insert ipv6 status + sqlx::query!( + r#" + INSERT INTO mixnode_ipv6_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?); + "#, + mixnode_id, + mixnode_result.working_ipv6, + timestamp + ) + .execute(&mut tx) + .await?; + } + + // finally commit the transaction + tx.commit().await + } + + /// Tries to submit gateway [`NodeResult`] from the network monitor to the database. + pub(crate) async fn submit_gateway_statuses( + &self, + timestamp: UnixTimestamp, + gateway_results: Vec, + ) -> Result<(), sqlx::Error> { + // insert it all in a transaction to make sure all nodes are updated at the same time + // (plus it's a nice guard against new nodes) + let mut tx = self.connection_pool.begin().await?; + + for gateway_result in gateway_results { + // if gateway info doesn't exist, insert it and get its id + + // same ID "problem" as described for mixnode insertion + let gateway_id = sqlx::query!( + r#" + INSERT OR IGNORE INTO gateway_details(identity, owner) VALUES (?, ?); + SELECT id FROM gateway_details WHERE identity = ?; + "#, + gateway_result.identity, + gateway_result.owner, + gateway_result.identity, + ) + .fetch_one(&mut tx) + .await? + .id; + + // insert ipv4 status + sqlx::query!( + r#" + INSERT INTO gateway_ipv4_status (gateway_details_id, up, timestamp) VALUES (?, ?, ?); + "#, + gateway_id, + gateway_result.working_ipv4, + timestamp + ) + .execute(&mut tx) + .await?; + + // insert ipv6 status + sqlx::query!( + r#" + INSERT INTO gateway_ipv6_status (gateway_details_id, up, timestamp) VALUES (?, ?, ?); + "#, + gateway_id, + gateway_result.working_ipv6, + timestamp + ) + .execute(&mut tx) + .await?; + } + + // finally commit the transaction + tx.commit().await + } + + /// Checks whether there are already any historical uptimes with this particular date. + pub(crate) async fn check_for_historical_uptime_existence( + &self, + today_iso_8601: &str, + ) -> Result { + sqlx::query!( + "SELECT EXISTS (SELECT 1 FROM mixnode_historical_uptime WHERE date = ?) AS 'exists'", + today_iso_8601 + ) + .fetch_one(&self.connection_pool) + .await + .map(|result| result.exists == 1) + } + + /// Creates new entry for mixnode historical uptime + pub(crate) async fn insert_mixnode_historical_uptime( + &self, + node_id: i64, + date: &str, + ipv4_uptime: u8, + ipv6_uptime: u8, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "INSERT INTO mixnode_historical_uptime(mixnode_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)", + node_id, + date, + ipv4_uptime, + ipv6_uptime, + ).execute(&self.connection_pool).await?; + Ok(()) + } + + /// Creates new entry for gatewy historical uptime + pub(crate) async fn insert_gateway_historical_uptime( + &self, + node_id: i64, + date: &str, + ipv4_uptime: u8, + ipv6_uptime: u8, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "INSERT INTO gateway_historical_uptime(gateway_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)", + node_id, + date, + ipv4_uptime, + ipv6_uptime, + ).execute(&self.connection_pool).await?; + Ok(()) + } + + pub(crate) async fn purge_old_mixnode_ipv4_statuses( + &self, + timestamp: UnixTimestamp, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "DELETE FROM mixnode_ipv4_status WHERE timestamp < ?", + timestamp + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } + + pub(crate) async fn purge_old_mixnode_ipv6_statuses( + &self, + timestamp: UnixTimestamp, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "DELETE FROM mixnode_ipv6_status WHERE timestamp < ?", + timestamp + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } + + pub(crate) async fn purge_old_gateway_ipv4_statuses( + &self, + timestamp: UnixTimestamp, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "DELETE FROM gateway_ipv4_status WHERE timestamp < ?", + timestamp + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } + + pub(crate) async fn purge_old_gateway_ipv6_statuses( + &self, + timestamp: UnixTimestamp, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "DELETE FROM gateway_ipv6_status WHERE timestamp < ?", + timestamp + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } + + // #################################################################################################### + // ALL THE METHODS BELOW ARE TEMPORARY AND WILL BE REMOVED ONCE PAYMENTS ARE DONE INSIDE VALIDATOR API + // #################################################################################################### + + // NOTE: this method will go away once we move payments into the validator-api + // it just helps us to get rid of having to query for reports of each node individually + /// Returns public key, owner and id of all mixnodes that have had any ipv4 statuses submitted + /// since provided timestamp. + pub(crate) async fn get_all_active_mixnodes( + &self, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + // find mixnode details of all nodes that have had at least 1 ipv4 status since the provided + // timestamp + // TODO: I dont know if theres a potential issue of if we have a lot of inactive nodes that + // haven't mixed in ages, they might increase the query times? + sqlx::query_as!( + ActiveNode, + r#" + SELECT DISTINCT identity, owner, id + FROM mixnode_details + JOIN mixnode_ipv4_status + ON mixnode_details.id = mixnode_ipv4_status.mixnode_details_id + WHERE EXISTS ( + SELECT 1 FROM mixnode_ipv4_status WHERE timestamp > ? + ) + "#, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + // NOTE: this method will go away once we move payments into the validator-api + // it just helps us to get rid of having to query for reports of each node individually + /// Returns public key, owner and id of all gateways that have had any ipv4 statuses submitted + /// since provided timestamp. + pub(crate) async fn get_all_active_gateways( + &self, + timestamp: UnixTimestamp, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + ActiveNode, + r#" + SELECT DISTINCT identity, owner, id + FROM gateway_details + JOIN gateway_ipv4_status + ON gateway_details.id = gateway_ipv4_status.gateway_details_id + WHERE EXISTS ( + SELECT 1 FROM gateway_ipv4_status WHERE timestamp > ? + ) + "#, + timestamp + ) + .fetch_all(&self.connection_pool) + .await + } + + // NOTE: this method will go away once we move payments into the validator-api + // it just helps us to get rid of having to query for reports of each node individually + // TODO: should that live on the 'Inner' struct or should it rather exist on the actual storage struct + // since technically it doesn't touch any SQL directly + pub(crate) async fn get_all_active_mixnodes_statuses( + &self, + ) -> Result, sqlx::Error> { + let now = OffsetDateTime::now_utc(); + let day_ago = (now - ONE_DAY).unix_timestamp(); + + let active_nodes = self.get_all_active_mixnodes(day_ago).await?; + + let mut active_day_statuses = Vec::with_capacity(active_nodes.len()); + for active_node in active_nodes.into_iter() { + let ipv4_statuses = self + .get_mixnode_ipv4_statuses_since_by_id(active_node.id, day_ago) + .await?; + let ipv6_statuses = self + .get_mixnode_ipv6_statuses_since_by_id(active_node.id, day_ago) + .await?; + + let statuses = ActiveNodeDayStatuses { + identity: active_node.identity, + owner: active_node.owner, + node_id: active_node.id, + ipv4_statuses, + ipv6_statuses, + }; + + active_day_statuses.push(statuses); + } + + Ok(active_day_statuses) + } + + // NOTE: this method will go away once we move payments into the validator-api + // it just helps us to get rid of having to query for reports of each node individually + // TODO: should that live on the 'Inner' struct or should it rather exist on the actual storage struct + // since technically it doesn't touch any SQL directly + pub(crate) async fn get_all_active_gateways_statuses( + &self, + ) -> Result, sqlx::Error> { + let now = OffsetDateTime::now_utc(); + let day_ago = (now - ONE_DAY).unix_timestamp(); + + let active_nodes = self.get_all_active_gateways(day_ago).await?; + + let mut active_day_statuses = Vec::with_capacity(active_nodes.len()); + for active_node in active_nodes.into_iter() { + let ipv4_statuses = self + .get_gateway_ipv4_statuses_since_by_id(active_node.id, day_ago) + .await?; + let ipv6_statuses = self + .get_gateway_ipv6_statuses_since_by_id(active_node.id, day_ago) + .await?; + + let statuses = ActiveNodeDayStatuses { + identity: active_node.identity, + owner: active_node.owner, + node_id: active_node.id, + ipv4_statuses, + ipv6_statuses, + }; + + active_day_statuses.push(statuses); + } + + Ok(active_day_statuses) + } +} diff --git a/validator-api/src/storage/mod.rs b/validator-api/src/storage/mod.rs new file mode 100644 index 0000000000..253566ac8a --- /dev/null +++ b/validator-api/src/storage/mod.rs @@ -0,0 +1,461 @@ +// Copyright 2021 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::network_monitor::monitor::summary_producer::NodeResult; +use crate::node_status_api::models::{ + GatewayStatusReport, GatewayUptimeHistory, MixnodeStatusReport, MixnodeUptimeHistory, + NodeStatusApiError, Uptime, +}; +use crate::node_status_api::ONE_DAY; +use crate::storage::manager::StorageManager; +use crate::storage::models::NodeStatus; +use rocket::fairing::{self, AdHoc}; +use rocket::{Build, Rocket}; +use sqlx::types::time::OffsetDateTime; +use sqlx::ConnectOptions; +use std::path::PathBuf; + +pub(crate) mod manager; +pub(crate) mod models; + +// A type alias to be more explicit about type of timestamp used. +type UnixTimestamp = i64; + +// note that clone here is fine as upon cloning the same underlying pool will be used +#[derive(Clone)] +pub(crate) struct NodeStatusStorage { + manager: StorageManager, +} + +impl NodeStatusStorage { + async fn init(rocket: Rocket, database_path: PathBuf) -> fairing::Result { + // TODO: we can inject here more stuff based on our validator-api global config + // struct. Maybe different pool size or timeout intervals? + let mut opts = sqlx::sqlite::SqliteConnectOptions::new() + .filename(&database_path) + .create_if_missing(true); + + // TODO: do we want auto_vacuum ? + + opts.disable_statement_logging(); + + let connection_pool = match sqlx::SqlitePool::connect_with(opts).await { + Ok(db) => db, + Err(e) => { + error!("Failed to connect to SQLx database: {}", e); + return Err(rocket); + } + }; + + if let Err(e) = sqlx::migrate!("./migrations").run(&connection_pool).await { + error!("Failed to initialize SQLx database: {}", e); + return Err(rocket); + } + + info!("Database migration finished!"); + + let storage = NodeStatusStorage { + manager: StorageManager { connection_pool }, + }; + + Ok(rocket.manage(storage)) + } + + pub(crate) fn stage(database_path: PathBuf) -> AdHoc { + AdHoc::try_on_ignite("SQLx Database", |rocket| { + NodeStatusStorage::init(rocket, database_path) + }) + } + + /// Gets all statuses for particular mixnode (ipv4 and ipv6) that were inserted in last 24h. + async fn get_mixnode_daily_statuses( + &self, + identity: &str, + ) -> Result<(Vec, Vec), NodeStatusApiError> { + let now = OffsetDateTime::now_utc(); + let day_ago = now - ONE_DAY; + + let ipv4_statuses = self + .manager + .get_mixnode_ipv4_statuses_since(identity, day_ago.unix_timestamp()) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + let ipv6_statuses = self + .manager + .get_mixnode_ipv6_statuses_since(identity, day_ago.unix_timestamp()) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + Ok((ipv4_statuses, ipv6_statuses)) + } + + /// Gets all statuses for particular gateway (ipv4 and ipv6) that were inserted in last 24h. + async fn get_gateway_daily_statuses( + &self, + identity: &str, + ) -> Result<(Vec, Vec), NodeStatusApiError> { + let now = OffsetDateTime::now_utc(); + let day_ago = now - ONE_DAY; + + let ipv4_statuses = self + .manager + .get_gateway_ipv4_statuses_since(identity, day_ago.unix_timestamp()) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + let ipv6_statuses = self + .manager + .get_gateway_ipv6_statuses_since(identity, day_ago.unix_timestamp()) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + Ok((ipv4_statuses, ipv6_statuses)) + } + + /// Tries to construct a status report for mixnode with the specified identity. + pub(crate) async fn construct_mixnode_report( + &self, + identity: &str, + ) -> Result { + let (ipv4_statuses, ipv6_statuses) = self.get_mixnode_daily_statuses(identity).await?; + + // if we have no statuses, the node doesn't exist (or monitor is down), but either way, we can't make a report + if ipv4_statuses.is_empty() { + return Err(NodeStatusApiError::MixnodeReportNotFound( + identity.to_owned(), + )); + } + + // now, technically this is not a critical error, but this should have NEVER happened in the first place + // so something super weird is going on + if ipv4_statuses.len() != ipv6_statuses.len() { + error!("Somehow we have different number of ipv4 and ipv6 statuses for mixnode {}! (ipv4: {}, ipv6: {})", + identity, + ipv4_statuses.len(), + ipv6_statuses.len(), + ) + } + + let mixnode_owner = self + .manager + .get_mixnode_owner(identity) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)? + .expect("The node doesn't have an owner even though we have status information on it!"); + + Ok(MixnodeStatusReport::construct_from_last_day_reports( + identity.to_owned(), + mixnode_owner, + ipv4_statuses, + ipv6_statuses, + )) + } + + pub(crate) async fn construct_gateway_report( + &self, + identity: &str, + ) -> Result { + let (ipv4_statuses, ipv6_statuses) = self.get_gateway_daily_statuses(identity).await?; + + // if we have no statuses, the node doesn't exist (or monitor is down), but either way, we can't make a report + if ipv4_statuses.is_empty() { + return Err(NodeStatusApiError::GatewayReportNotFound( + identity.to_owned(), + )); + } + + // now, technically this is not a critical error, but this should have NEVER happened in the first place + // so something super weird is going on + if ipv4_statuses.len() != ipv6_statuses.len() { + error!("Somehow we have different number of ipv4 and ipv6 statuses for gateway {}! (ipv4: {}, ipv6: {})", + identity, + ipv4_statuses.len(), + ipv6_statuses.len(), + ) + } + + let gateway_owner = self + .manager + .get_gateway_owner(identity) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)? + .expect( + "The gateway doesn't have an owner even though we have status information on it!", + ); + + Ok(GatewayStatusReport::construct_from_last_day_reports( + identity.to_owned(), + gateway_owner, + ipv4_statuses, + ipv6_statuses, + )) + } + + pub(crate) async fn get_mixnode_uptime_history( + &self, + identity: &str, + ) -> Result { + let history = self + .manager + .get_mixnode_historical_uptimes(identity) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + if history.is_empty() { + return Err(NodeStatusApiError::MixnodeUptimeHistoryNotFound( + identity.to_owned(), + )); + } + + let mixnode_owner = self + .manager + .get_mixnode_owner(identity) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)? + .expect("The node doesn't have an owner even though we have uptime history for it!"); + + Ok(MixnodeUptimeHistory::new( + identity.to_owned(), + mixnode_owner, + history, + )) + } + + pub(crate) async fn get_gateway_uptime_history( + &self, + identity: &str, + ) -> Result { + let history = self + .manager + .get_gateway_historical_uptimes(identity) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + if history.is_empty() { + return Err(NodeStatusApiError::GatewayUptimeHistoryNotFound( + identity.to_owned(), + )); + } + + let gateway_owner = self + .manager + .get_gateway_owner(identity) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)? + .expect("The gateway doesn't have an owner even though we have uptime history for it!"); + + Ok(GatewayUptimeHistory::new( + identity.to_owned(), + gateway_owner, + history, + )) + } + + // NOTE: this method will go away once we move payments into the validator-api + // it just helps us to get rid of having to query for reports of each node individually + pub(crate) async fn get_all_mixnode_reports( + &self, + ) -> Result, NodeStatusApiError> { + let reports = self + .manager + .get_all_active_mixnodes_statuses() + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)? + .into_iter() + .map(|statuses| { + MixnodeStatusReport::construct_from_last_day_reports( + statuses.identity, + statuses.owner, + statuses.ipv4_statuses, + statuses.ipv6_statuses, + ) + }) + .collect(); + + Ok(reports) + } + + // NOTE: this method will go away once we move payments into the validator-api + // it just helps us to get rid of having to query for reports of each node individually + pub(crate) async fn get_all_gateway_reports( + &self, + ) -> Result, NodeStatusApiError> { + let reports = self + .manager + .get_all_active_gateways_statuses() + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)? + .into_iter() + .map(|statuses| { + GatewayStatusReport::construct_from_last_day_reports( + statuses.identity, + statuses.owner, + statuses.ipv4_statuses, + statuses.ipv6_statuses, + ) + }) + .collect(); + + Ok(reports) + } + + // Used by network monitor + pub(crate) async fn submit_new_statuses( + &self, + mixnode_results: Vec, + gateway_results: Vec, + ) -> Result<(), NodeStatusApiError> { + info!("Submitting new node results to the database. There are {} mixnode results and {} gateway results", mixnode_results.len(), gateway_results.len()); + + let now = OffsetDateTime::now_utc().unix_timestamp(); + + self.manager + .submit_mixnode_statuses(now, mixnode_results) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + self.manager + .submit_gateway_statuses(now, gateway_results) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError) + } + + // Called on timer/reward script + async fn update_historical_uptimes( + &self, + today_iso_8601: &str, + ) -> Result<(), NodeStatusApiError> { + // get statuses for all active mixnodes... + let active_mixnodes_statuses = self + .manager + .get_all_active_mixnodes_statuses() + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + for statuses in active_mixnodes_statuses.into_iter() { + let ipv4_day_up = statuses + .ipv4_statuses + .iter() + .filter(|status| status.up) + .count(); + let ipv6_day_up = statuses + .ipv6_statuses + .iter() + .filter(|status| status.up) + .count(); + + // calculate their uptimes for the last 24h + let ipv4_uptime = Uptime::from_ratio(ipv4_day_up, statuses.ipv4_statuses.len()) + .unwrap() + .u8(); + let ipv6_uptime = Uptime::from_ratio(ipv6_day_up, statuses.ipv6_statuses.len()) + .unwrap() + .u8(); + + // and insert into the database + self.manager + .insert_mixnode_historical_uptime( + statuses.node_id, + today_iso_8601, + ipv4_uptime, + ipv6_uptime, + ) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + } + + // get statuses for all active gateways... + let active_gateways_statuses = self + .manager + .get_all_active_gateways_statuses() + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + + for statuses in active_gateways_statuses.into_iter() { + let ipv4_day_up = statuses + .ipv4_statuses + .iter() + .filter(|status| status.up) + .count(); + let ipv6_day_up = statuses + .ipv6_statuses + .iter() + .filter(|status| status.up) + .count(); + + // calculate their uptimes for the last 24h + let ipv4_uptime = Uptime::from_ratio(ipv4_day_up, statuses.ipv4_statuses.len()) + .unwrap() + .u8(); + let ipv6_uptime = Uptime::from_ratio(ipv6_day_up, statuses.ipv6_statuses.len()) + .unwrap() + .u8(); + + // and insert into the database + self.manager + .insert_gateway_historical_uptime( + statuses.node_id, + today_iso_8601, + ipv4_uptime, + ipv6_uptime, + ) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + } + + Ok(()) + } + + async fn check_if_historical_uptimes_exist_for_date( + &self, + date_iso_8601: &str, + ) -> Result { + self.manager + .check_for_historical_uptime_existence(date_iso_8601) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError) + } + + // Called on timer/reward script + async fn purge_old_statuses(&self) -> Result<(), NodeStatusApiError> { + let now = OffsetDateTime::now_utc(); + let two_days_ago = (now - 2 * ONE_DAY).unix_timestamp(); + + self.manager + .purge_old_mixnode_ipv4_statuses(two_days_ago) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + self.manager + .purge_old_mixnode_ipv6_statuses(two_days_ago) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + self.manager + .purge_old_gateway_ipv4_statuses(two_days_ago) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError)?; + self.manager + .purge_old_gateway_ipv6_statuses(two_days_ago) + .await + .map_err(|_| NodeStatusApiError::InternalDatabaseError) + } + + pub(crate) async fn daily_chores(&self) -> Result { + let today_iso_8601 = OffsetDateTime::now_utc().date().to_string(); + + // if we have already performed the update for today's date, don't do anything + if self + .check_if_historical_uptimes_exist_for_date(&today_iso_8601) + .await? + { + Ok(false) + } else { + info!( + "Updating historical daily uptimes of all nodes and purging old status reports..." + ); + self.update_historical_uptimes(&today_iso_8601).await?; + self.purge_old_statuses().await?; + Ok(true) + } + } +} diff --git a/validator-api/src/storage/models.rs b/validator-api/src/storage/models.rs new file mode 100644 index 0000000000..a83a3ad66f --- /dev/null +++ b/validator-api/src/storage/models.rs @@ -0,0 +1,15 @@ +// Copyright 2021 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +// Internally used struct to catch results from the database to calculate uptimes for given mixnode/gateway +pub(crate) struct NodeStatus { + pub(crate) timestamp: i64, + pub(crate) up: bool, +} + +// Internally used struct to catch results from the database to find active mixnodes/gateways +pub(crate) struct ActiveNode { + pub(crate) id: i64, + pub(crate) identity: String, + pub(crate) owner: String, +}