Monthly income report service completed. Clippy fixes.

Cargo.toml

@@ -34,4 +34,5 @@ anyhow = "1.0"
 itertools = "0.12.0"
 lazy_static = "1.4"
 log = "0.4"
+num-traits = "0.2"
 thiserror = "1.0"

subvt-network-status-updater/src/lib.rs

@@ -38,7 +38,7 @@ impl NetworkStatusUpdater {
             ))?;
         let status_json_string = serde_json::to_string(status)?;
         let mut redis_cmd_pipeline = Pipeline::new();
-        redis_cmd_pipeline
+        () = redis_cmd_pipeline
             .cmd("SET")
             .arg(format!("subvt:{}:network_status", CONFIG.substrate.chain))
             .arg(status_json_string)

subvt-notification-generator/src/lib.rs

@@ -1,9 +1,9 @@
 //! Generates notifications according to the notification rules depending on three sources of data:
 //! 1. Validator list updates from Redis, updated by `subvt-validator-list-updater`, and published
-//! using the Redis notification (PUBLISH) support.
+//!    using the Redis notification (PUBLISH) support.
 //! 2. Events and extrinsics in new blocks. Block are processed by `subvt-block-processor`, and the
-//! finishing of the processing of a block is signalled by the processor by means of PostgreSQL
-//! notifications.
+//!    finishing of the processing of a block is signalled by the processor by means of PostgreSQL
+//!    notifications.
 //! 3. Regular Telemetry checks (this is work in progress still).
 #![warn(clippy::disallowed_types)]
 use async_trait::async_trait;

subvt-persistence/src/postgres/app/notification.rs

@@ -148,7 +148,7 @@ impl PostgreSQLAppStorage {
             .bind(&notification.validator_account_json)
             .bind(&notification.notification_type_code)
             .bind(notification.user_notification_channel_id as i32)
-            .bind(&notification.notification_channel.to_string())
+            .bind(notification.notification_channel.to_string())
             .bind(&notification.notification_target)
             .bind(&notification.data_json)
             .bind(&notification.error_log)

subvt-persistence/src/postgres/app/user.rs

@@ -218,7 +218,7 @@ impl PostgreSQLAppStorage {
             "#,
         )
             .bind(user_notification_channel.user_id as i32)
-            .bind(&user_notification_channel.channel.to_string())
+            .bind(user_notification_channel.channel.to_string())
             .bind(&user_notification_channel.target)
             .fetch_one(&self.connection_pool)
             .await?;
@@ -237,7 +237,7 @@ impl PostgreSQLAppStorage {
             "#,
         )
         .bind(user_notification_channel.user_id as i32)
-        .bind(&user_notification_channel.channel.to_string())
+        .bind(user_notification_channel.channel.to_string())
         .bind(&user_notification_channel.target)
         .fetch_one(&self.connection_pool)
         .await?;

subvt-persistence/src/postgres/network/kline/mod.rs

@@ -0,0 +1,106 @@
+use crate::postgres::network::PostgreSQLNetworkStorage;
+use sqlx::types::BigDecimal;
+use subvt_types::kline::KLine;
+
+type DBKline = (
+    i32,
+    i64,
+    String,
+    String,
+    BigDecimal,
+    BigDecimal,
+    BigDecimal,
+    BigDecimal,
+    BigDecimal,
+    i64,
+    BigDecimal,
+    i32,
+    BigDecimal,
+    BigDecimal,
+);
+
+impl PostgreSQLNetworkStorage {
+    pub async fn kline_exists(
+        &self,
+        timestamp: i64,
+        source_ticker: &str,
+        target_ticker: &str,
+    ) -> anyhow::Result<bool> {
+        let record_count: (i64,) = sqlx::query_as(
+            r#"
+            SELECT COUNT(id) FROM sub_kline_historical
+            WHERE open_time = $1 AND source_ticker = $2 AND target_ticker = $3
+            "#,
+        )
+        .bind(timestamp)
+        .bind(source_ticker)
+        .bind(target_ticker)
+        .fetch_one(&self.connection_pool)
+        .await?;
+        Ok(record_count.0 > 0)
+    }
+
+    pub async fn save_kline(&self, kline: &KLine) -> anyhow::Result<i32> {
+        let save_result: (i32,) = sqlx::query_as(
+            r#"
+            INSERT INTO sub_kline_historical (open_time, source_ticker, target_ticker, "open", high, low, "close", volume, close_time, quote_volume, "count", taker_buy_volume, taker_buy_quote_volume)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
+            ON CONFLICT (open_time, source_ticker, target_ticker)
+            DO UPDATE SET taker_buy_quote_volume = EXCLUDED.taker_buy_quote_volume
+            RETURNING id
+            "#,
+        )
+            .bind(kline.open_time as i64)
+            .bind(kline.source_ticker.as_str())
+            .bind(kline.target_ticker.as_str())
+            .bind(&kline.open)
+            .bind(&kline.high)
+            .bind(&kline.low)
+            .bind(&kline.close)
+            .bind(&kline.volume)
+            .bind(kline.close_time as i64)
+            .bind(&kline.quote_volume)
+            .bind(kline.count as i32)
+            .bind(&kline.taker_buy_volume)
+            .bind(&kline.taker_buy_quote_volume)
+            .fetch_one(&self.connection_pool)
+            .await?;
+        Ok(save_result.0)
+    }
+
+    pub async fn get_kline_count(&self) -> anyhow::Result<u64> {
+        let record_count: (i64,) = sqlx::query_as("SELECT COUNT(id) FROM sub_kline_historical")
+            .fetch_one(&self.connection_pool)
+            .await?;
+        Ok(record_count.0 as u64)
+    }
+
+    pub async fn get_kline(&self, timestamp: u64) -> anyhow::Result<KLine> {
+        let db_kline: DBKline = sqlx::query_as(
+            r#"
+            SELECT id, open_time, source_ticker, target_ticker, "open", high, low, "close", volume, close_time, quote_volume, "count", taker_buy_volume, taker_buy_quote_volume
+            FROM sub_kline_historical
+            WHERE open_time = $1
+            "#,
+        )
+            .bind(timestamp as i64)
+            .fetch_one(&self.connection_pool)
+            .await?;
+        Ok(KLine {
+            id: db_kline.0 as u32,
+            open_time: db_kline.1 as u64,
+            source_ticker: db_kline.2,
+            target_ticker: db_kline.3,
+            open: db_kline.4,
+            high: db_kline.5,
+            low: db_kline.6,
+            close: db_kline.7,
+            volume: db_kline.8,
+            close_time: db_kline.9 as u64,
+            quote_volume: db_kline.10,
+            count: db_kline.11 as u32,
+            taker_buy_volume: db_kline.12,
+            taker_buy_quote_volume: db_kline.13,
+        })
+    }
+}

subvt-persistence/src/postgres/network/mod.rs

@@ -12,6 +12,7 @@ pub mod era;
 pub mod error_log;
 pub mod event;
 pub mod extrinsic;
+pub mod kline;
 pub mod nft;
 pub mod notify;
 pub mod onekv;

subvt-persistence/src/postgres/network/onekv.rs

@@ -48,7 +48,7 @@ impl PostgreSQLNetworkStorage {
             .bind(candidate.inclusion)
             .bind(candidate.commission)
             .bind(candidate.is_active)
-            .bind(&candidate.unclaimed_eras.as_ref().map(|v| v.iter().map(|i| *i  as i64).collect::<Vec<i64>>()))
+            .bind(candidate.unclaimed_eras.as_ref().map(|v| v.iter().map(|i| *i  as i64).collect::<Vec<i64>>()))
             .bind(candidate.nominated_at.map(|last_valid| last_valid as i64))
             .bind(candidate.offline_accumulated)
             .bind(candidate.offline_since as i64)

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

@@ -1,7 +1,7 @@
 use crate::postgres::network::PostgreSQLNetworkStorage;
 use std::str::FromStr;
 use subvt_types::crypto::AccountId;
-use subvt_types::report::ValidatorTotalReward;
+use subvt_types::report::{Reward, ValidatorTotalReward};
 use subvt_types::substrate::{Balance, Era};
 
 impl PostgreSQLNetworkStorage {
@@ -70,4 +70,41 @@ impl PostgreSQLNetworkStorage {
         }
         Ok(result)
     }
+
+    pub async fn get_rewards_in_time_range(
+        &self,
+        rewardee_account_id: &AccountId,
+        start_timestamp: u64,
+        end_timestamp: u64,
+    ) -> anyhow::Result<Vec<Reward>> {
+        let db_rewards: Vec<(i32, String, i64, i64, i32, i32, String)> = sqlx::query_as(
+            r#"
+            SELECT E.id, E.block_hash, B.number, B.timestamp, E.extrinsic_index, E.event_index, E.amount
+            FROM sub_event_rewarded E
+            INNER JOIN sub_block B ON B.hash = E.block_hash
+            WHERE B.timestamp >= $1 AND B.timestamp < $2
+            AND E.rewardee_account_id = $3
+            ORDER BY B.timestamp ASC
+            "#,
+        )
+            .bind(start_timestamp as i64)
+            .bind(end_timestamp as i64)
+            .bind(rewardee_account_id.to_string())
+            .fetch_all(&self.connection_pool)
+            .await?;
+        let mut rewards = Vec::new();
+        for db_reward in db_rewards.iter() {
+            rewards.push(Reward {
+                id: db_reward.0 as u32,
+                block_hash: db_reward.1.clone(),
+                block_number: db_reward.2 as u64,
+                block_timestamp: db_reward.3 as u64,
+                extrinsic_index: db_reward.4 as u32,
+                event_index: db_reward.5 as u32,
+                rewardee_account_id: *rewardee_account_id,
+                amount: db_reward.6.parse()?,
+            })
+        }
+        Ok(rewards)
+    }
 }

subvt-persistence/src/postgres/network/telegram/mod.rs

@@ -253,7 +253,7 @@ impl PostgreSQLNetworkStorage {
         )
         .bind(app_user_id as i32)
         .bind(telegram_chat_id)
-        .bind(&state.to_string())
+        .bind(state.to_string())
         .execute(&self.connection_pool)
         .await?;
         Ok(())
@@ -284,7 +284,7 @@ impl PostgreSQLNetworkStorage {
             "#,
         )
         .bind(telegram_chat_id)
-        .bind(&validator_account_id.to_string())
+        .bind(validator_account_id.to_string())
         .fetch_one(&self.connection_pool)
         .await?;
         Ok(record_count.0 > 0)
@@ -307,7 +307,7 @@ impl PostgreSQLNetworkStorage {
             "#,
         )
         .bind(telegram_chat_id)
-        .bind(&account_id.to_string())
+        .bind(account_id.to_string())
         .bind(address)
         .bind(display)
         .fetch_one(&self.connection_pool)
@@ -348,7 +348,7 @@ impl PostgreSQLNetworkStorage {
             WHERE telegram_chat_id = $2
             "#,
         )
-        .bind(&state.to_string())
+        .bind(state.to_string())
         .bind(telegram_chat_id)
         .execute(&self.connection_pool)
         .await?;

subvt-persistence/src/redis.rs

@@ -28,7 +28,7 @@ impl Redis {
         block_summary: &BlockSummary,
     ) -> anyhow::Result<()> {
         let mut connection = self.client.get_multiplexed_async_connection().await?;
-        redis::cmd("MSET")
+        () = redis::cmd("MSET")
             .arg(format!(
                 "subvt:{}:validators:finalized_block_number",
                 CONFIG.substrate.chain
@@ -55,7 +55,7 @@ impl Redis {
         account_id: &AccountId,
     ) -> anyhow::Result<()> {
         let mut connection = self.client.get_multiplexed_async_connection().await?;
-        redis::cmd("SADD")
+        () = redis::cmd("SADD")
             .arg(format!(
                 "subvt:{}:validators:{finalized_block_number}:active:account_id_set",
                 CONFIG.substrate.chain,
@@ -73,7 +73,7 @@ impl Redis {
     ) -> anyhow::Result<()> {
         let mut connection = self.client.get_multiplexed_async_connection().await?;
         let validator_details_json = serde_json::to_string(validator_details)?;
-        redis::cmd("SET")
+        () = redis::cmd("SET")
             .arg(format!(
                 "subvt:{}:validators:{finalized_block_number}:active:validator:{}",
                 CONFIG.substrate.chain, validator_details.account.id,
@@ -171,7 +171,7 @@ impl Redis {
     pub async fn set_network_status(&self, network_status: &NetworkStatus) -> anyhow::Result<()> {
         let mut connection = self.client.get_multiplexed_async_connection().await?;
         let network_status_json = serde_json::to_string(network_status)?;
-        redis::cmd("SET")
+        () = redis::cmd("SET")
             .arg(format!("subvt:{}:network_status", CONFIG.substrate.chain))
             .arg(network_status_json)
             .query_async(&mut connection)

subvt-report-service/Cargo.toml

@@ -8,6 +8,7 @@ rust-version = "1.67.0"
 actix-web = "4.4"
 anyhow = { workspace = true }
 async-trait = "0.1"
+chrono = "0.4"
 futures-util = "0.3"
 hex = "0.4"
 lazy_static = { workspace = true }
@@ -17,6 +18,7 @@ redis = { version = "0.25", features = ["tokio-comp"] }
 rustc-hash = "1.1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
+sqlx = { git  = "https://github.com/helikon-labs/sqlx.git", branch = "helikon-increased-field-count", features = ["bigdecimal"] }
 subvt-config = { path = "../subvt-config" }
 subvt-metrics = { path = "../subvt-metrics" }
 subvt-persistence = { path = "../subvt-persistence" }