Add validator rewards chart service. Dependency updates.

Author: kukabi

Cargo.lock

@@ -17,7 +17,7 @@ jsonrpsee = { version = "0.16", features = ["full"] }
 lazy_static = { workspace = true }
 log = { workspace = true }
 once_cell = "1"
-redis = { version = "0.22.1", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 subvt-config = { path = "../subvt-config" }

subvt-network-status-server/Cargo.toml

@@ -11,7 +11,7 @@ chrono = "0.4"
 lazy_static = { workspace = true }
 log = { workspace = true }
 once_cell = "1"
-redis = { version = "0.22.1", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 subvt-config = { path = "../subvt-config" }

subvt-network-status-updater/Cargo.toml

@@ -14,7 +14,7 @@ futures-util = "0.3"
 lazy_static = { workspace = true }
 log = { workspace = true }
 once_cell = "1"
-redis = "0.22"
+redis = "0.23"
 rustc-hash = "1.1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

subvt-notification-generator/Cargo.toml

@@ -21,7 +21,7 @@ lettre = { version = "0.10", default-features = true, features = ["tokio1-native
 lettre_email = "0.9.4"
 log = { workspace = true }
 once_cell = "1"
-redis = { version = "0.22", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 rustc-hash = "1.1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

subvt-notification-processor/Cargo.toml

@@ -11,7 +11,7 @@ hex = "0.4"
 lazy_static = { workspace = true }
 log = { workspace = true }
 parity-scale-codec = "3.4"
-redis = { version = "0.22", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 rustc-hash = "1.1.0"
 serde = { version = "1.0" }
 serde_json = "1.0"

subvt-persistence/Cargo.toml

@@ -1,5 +1,7 @@
 use crate::postgres::network::PostgreSQLNetworkStorage;
+use std::str::FromStr;
 use subvt_types::crypto::AccountId;
+use subvt_types::report::ValidatorTotalReward;
 use subvt_types::substrate::{Balance, Era};
 
 impl PostgreSQLNetworkStorage {
@@ -39,4 +41,39 @@ impl PostgreSQLNetworkStorage {
         }
         Ok(result)
     }
+
+    pub async fn get_validator_total_rewards(
+        &self,
+        start_timestamp: u64,
+        end_timestamp: u64,
+    ) -> anyhow::Result<Vec<ValidatorTotalReward>> {
+        let validator_total_rewards: Vec<(String, i64)> = sqlx::query_as(
+            r#"
+            SELECT ER.rewardee_account_id as validator_account_id, SUM(ER.amount::bigint) AS total_reward
+            FROM sub_event_rewarded ER
+            INNER JOIN sub_block B
+                ON ER.block_hash = B.hash
+                AND B.timestamp > $1
+                AND B.timestamp < $2
+            WHERE EXISTS (
+                SELECT * FROM sub_era_validator EV
+                WHERE EV.validator_account_id = ER.rewardee_account_id
+            )
+            GROUP BY ER.rewardee_account_id
+            ORDER BY total_reward DESC;
+            "#,
+        )
+        .bind(start_timestamp as i64)
+        .bind(end_timestamp as i64)
+        .fetch_all(&self.connection_pool)
+        .await?;
+        let mut result = vec![];
+        for validator_total_reward in validator_total_rewards {
+            result.push(ValidatorTotalReward {
+                validator_account_id: AccountId::from_str(&validator_total_reward.0)?,
+                total_reward: validator_total_reward.1 as Balance,
+            })
+        }
+        Ok(result)
+    }
 }

subvt-persistence/src/postgres/network/report/rewards.rs

@@ -12,13 +12,14 @@ futures-util = "0.3"
 lazy_static = { workspace = true }
 log = { workspace = true }
 once_cell = "1"
-redis = { version = "0.22", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 subvt-config = { path = "../subvt-config" }
 subvt-metrics = { path = "../subvt-metrics" }
 subvt-persistence = { path = "../subvt-persistence" }
 subvt-service-common = { path = "../subvt-service-common" }
+subvt-substrate-client = { path = "../subvt-substrate-client" }
 subvt-types = { path = "../subvt-types" }
 subvt-logging = { path = "../subvt-logging" }
 tokio = { version = "1.26", features = ["full"] }

subvt-report-service/Cargo.toml

@@ -12,6 +12,7 @@ use subvt_config::Config;
 use subvt_persistence::postgres::network::PostgreSQLNetworkStorage;
 use subvt_persistence::redis::Redis;
 use subvt_service_common::{err::InternalServerError, Service};
+use subvt_substrate_client::SubstrateClient;
 use subvt_types::report::BlockSummary;
 use subvt_types::subvt::ValidatorSummary;
 
@@ -31,6 +32,7 @@ pub(crate) type ResultResponse = Result<HttpResponse, InternalServerError>;
 pub(crate) struct ServiceState {
     postgres: Arc<PostgreSQLNetworkStorage>,
     redis: Arc<Redis>,
+    substrate_client: Arc<SubstrateClient>,
     finalized_block_summary: Arc<RwLock<BlockSummary>>,
     active_validator_list: Arc<RwLock<Vec<ValidatorSummary>>>,
     inactive_validator_list: Arc<RwLock<Vec<ValidatorSummary>>>,
@@ -65,6 +67,7 @@ impl Service for ReportService {
             "Cannot connect to Redis at URL {}.",
             CONFIG.redis.url
         ))?;
+        let substrate_client = Arc::new(SubstrateClient::new(&CONFIG).await?);
         let mut pubsub_connection = redis_client.get_async_connection().await?.into_pubsub();
         pubsub_connection
             .subscribe(format!(
@@ -168,6 +171,7 @@ impl Service for ReportService {
                 .app_data(web::Data::new(ServiceState {
                     postgres: postgres.clone(),
                     redis: redis.clone(),
+                    substrate_client: substrate_client.clone(),
                     finalized_block_summary: finalized_block_summary.clone(),
                     active_validator_list: active_validator_list.clone(),
                     inactive_validator_list: inactive_validator_list.clone(),
@@ -214,6 +218,7 @@ impl Service for ReportService {
                 .service(validator::validator_search_service)
                 .service(validator::validator_era_rewards_service)
                 .service(validator::validator_era_payouts_service)
+                .service(validator::validator_reward_chart_service)
         })
         .workers(10)
         .disable_signals()

subvt-report-service/src/lib.rs

@@ -6,7 +6,7 @@ use subvt_types::crypto::AccountId;
 use subvt_types::err::ServiceError;
 use subvt_types::report::{
     BlockSummary, EraValidatorPayoutReport, EraValidatorRewardReport, ValidatorDetailsReport,
-    ValidatorListReport, ValidatorSummaryReport,
+    ValidatorListReport, ValidatorSummaryReport, ValidatorTotalRewardChartData,
 };
 use subvt_types::subvt::{ValidatorSearchSummary, ValidatorSummary};
 
@@ -241,3 +241,35 @@ pub(crate) async fn validator_era_payouts_service(
         .collect();
     Ok(HttpResponse::Ok().json(era_payouts))
 }
+
+#[derive(Deserialize)]
+pub(crate) struct ValidatorRewardChartQueryParameters {
+    start_timestamp: u64,
+    end_timestamp: u64,
+}
+
+#[get("/validator/reward/chart")]
+pub(crate) async fn validator_reward_chart_service(
+    query: web::Query<ValidatorRewardChartQueryParameters>,
+    data: web::Data<ServiceState>,
+) -> ResultResponse {
+    let rewards = data
+        .postgres
+        .get_validator_total_rewards(query.start_timestamp, query.end_timestamp)
+        .await?;
+    let block_hash = data.substrate_client.get_current_block_hash().await?;
+    let account_ids: Vec<AccountId> = rewards
+        .iter()
+        .map(|reward| reward.validator_account_id)
+        .collect();
+    let accounts = data
+        .substrate_client
+        .get_accounts(&account_ids, &block_hash)
+        .await?;
+    Ok(HttpResponse::Ok().json(ValidatorTotalRewardChartData {
+        accounts,
+        rewards,
+        start_timestamp: query.start_timestamp,
+        end_timestamp: query.end_timestamp,
+    }))
+}

subvt-report-service/src/validator/mod.rs

@@ -1,7 +1,7 @@
 //! Report presentation types. Utilized by the `subvt-report-service` crate to server era and
 //! validator reports.
 use crate::crypto::AccountId;
-use crate::substrate::{Balance, Epoch, Era};
+use crate::substrate::{Account, Balance, Epoch, Era};
 use crate::subvt::{ValidatorDetails, ValidatorSummary};
 use serde::{Deserialize, Serialize};
 
@@ -193,3 +193,17 @@ impl From<&(Era, Balance)> for EraValidatorPayoutReport {
         }
     }
 }
+
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+pub struct ValidatorTotalReward {
+    pub validator_account_id: AccountId,
+    pub total_reward: Balance,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+pub struct ValidatorTotalRewardChartData {
+    pub accounts: Vec<Account>,
+    pub rewards: Vec<ValidatorTotalReward>,
+    pub start_timestamp: u64,
+    pub end_timestamp: u64,
+}

subvt-types/src/report.rs

@@ -16,7 +16,7 @@ jsonrpsee-core = "0.16"
 lazy_static = { workspace = true }
 log = { workspace = true }
 once_cell = "1"
-redis = { version = "0.22", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 subvt-config = { path = "../subvt-config" }

subvt-validator-details-server/Cargo.toml

@@ -18,7 +18,7 @@ jsonrpsee = { version = "0.16", features = ["full"] }
 lazy_static = { workspace = true }
 log = { workspace = true }
 once_cell = "1"
-redis = { version = "0.22", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 rustc-hash = "1.1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

subvt-validator-list-server/Cargo.toml

@@ -15,7 +15,7 @@ hex = "0.4"
 lazy_static = { workspace = true }
 log = { workspace = true }
 once_cell = "1"
-redis = { version = "0.22", features = ["tokio-comp"] }
+redis = { version = "0.23", features = ["tokio-comp"] }
 rustc-hash = "1.1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"