Merge pull request #6 from helium/madninja/rewads_validators

Madninja/rewads validators

Cargo.toml

@@ -19,3 +19,4 @@ futures = "0"
 tokio = { version = "1", features = ["full"] }
 sqlx = {version = "0", features = [ "runtime-tokio-rustls", "postgres", "chrono" ] }
 h3ron = "0"
+csv = "1"

src/blockspan.rs

@@ -24,6 +24,15 @@ impl BlockSpan {
         Self::for_timespan(pool, &timespan).await
     }
 
+    pub async fn for_date_range<S: ToDateTimeUtc, E: ToDateTimeUtc>(
+        pool: &PgPool,
+        start: S,
+        end: E,
+    ) -> Result<Self> {
+        let timespan = TimeSpan::for_date_range(start, end);
+        Self::for_timespan(pool, &timespan).await
+    }
+
     pub async fn for_timespan(pool: &PgPool, timespan: &TimeSpan) -> Result<Self> {
         let span: BlockSpan = sqlx::query_as(BLOCKSPAN_QUERY)
             .bind(timespan.high)

src/cmd/mod.rs

@@ -1,11 +1,12 @@
-pub mod blocks;
-pub mod hotspots;
-pub mod rewards;
-
 use crate::Result;
+use futures::stream::{BoxStream, TryStreamExt};
+use std::path::PathBuf;
 use structopt::StructOpt;
 
-use std::path::PathBuf;
+pub mod blocks;
+// pub mod flow;
+pub mod hotspots;
+pub mod rewards;
 
 /// Common options for most commands
 #[derive(Debug, StructOpt)]
@@ -18,3 +19,63 @@ pub fn print_json<T: ?Sized + serde::Serialize>(value: &T) -> Result {
     println!("{}", serde_json::to_string_pretty(value)?);
     Ok(())
 }
+
+#[derive(Debug, Default)]
+pub enum Format {
+    #[default]
+    Json,
+    Csv,
+}
+
+impl Format {
+    pub async fn output<'a, W, E>(
+        &self,
+        output: W,
+        mut rows: BoxStream<'a, std::result::Result<E, sqlx::Error>>,
+    ) -> Result
+    where
+        W: std::io::Write,
+        E: serde::Serialize,
+    {
+        match self {
+            Self::Json => {
+                use serde::{ser::SerializeSeq, Serializer};
+                let mut serializer = serde_json::Serializer::pretty(output);
+                let mut entries = serializer.serialize_seq(None)?;
+                while let Some(row) = rows.try_next().await? {
+                    entries.serialize_element(&row)?;
+                }
+                entries.end()?;
+            }
+            Self::Csv => {
+                let mut serializer = csv::Writer::from_writer(output);
+                while let Some(row) = rows.try_next().await? {
+                    serializer.serialize(&row)?;
+                }
+                serializer.flush()?;
+            }
+        }
+        Ok(())
+    }
+}
+
+impl std::str::FromStr for Format {
+    type Err = crate::Error;
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        use anyhow::anyhow;
+        match s.to_ascii_lowercase().as_str() {
+            "csv" => Ok(Self::Csv),
+            "json" => Ok(Self::Json),
+            _ => Err(anyhow!("invalid format {s}")),
+        }
+    }
+}
+
+impl ToString for Format {
+    fn to_string(&self) -> String {
+        match self {
+            Self::Csv => "csv".to_string(),
+            Self::Json => "json".to_string(),
+        }   
+    }
+}

src/cmd/rewards/account.rs

@@ -0,0 +1,68 @@
+use crate::{
+    cmd::{Format, Opts},
+    BlockSpan, Result,
+};
+use chrono::{DateTime, NaiveDate, Utc};
+use sqlx::postgres::PgPool;
+use structopt::StructOpt;
+
+#[derive(Debug, StructOpt)]
+/// Generates CSV  or JSON output with HNT rewards for all reward entries for
+/// validators and emitted securities owned by a given wallet.  
+pub struct Cmd {
+    /// The wallet address to look up validators for
+    account: String,
+
+    /// The start day (inclusive) to run the report over (in UTC). The start
+    /// time is at the beginning midnight of the given date (00:00:00).
+    start: NaiveDate,
+
+    /// The end day (exclusive) to run the report over (in UTC). The end time is
+    /// at the beginning midnight of the given date (00:00:00).
+    end: NaiveDate,
+
+    #[structopt(long, default_value)]
+    format: Format,
+}
+
+#[derive(Debug, serde::Serialize, sqlx::FromRow)]
+pub struct ValidatorReward {
+    block: i64,
+    timestamp: DateTime<Utc>,
+    reward_type: String,
+    transaction_hash: String,
+    validator: Option<String>,
+    hnt: f64,
+    usd_oracle_price: f64,
+    usd_amount: f64,
+}
+
+const VALIDATOR_REWRDS_QUERY: &str = r#"
+    select 
+        t.block,
+        to_timestamp(t.time) as timestamp,
+        case when gateway = '1Wh4bh' then 'securities' else 'validator' end as reward_type,
+        t.transaction_hash,
+        (case when gateway = '1Wh4bh' then null else gateway end) as validator,
+        amount::float8 / 100000000 as hnt,
+        o.price::float8 / 100000000 as usd_oracle_price,
+        (amount::float8 / 100000000) * (o.price::float8 / 100000000) as usd_amount
+    from rewards t
+    left join oracle_prices o on o.block = (select max(o2.block) from oracle_prices o2 where o2.block <= t.block)
+    where t.block between $1 and $2
+    and account = $3
+    order by t.block asc;
+"#;
+
+impl Cmd {
+    pub async fn run(&self, pool: &PgPool, _opts: Opts) -> Result {
+        let blockspan = BlockSpan::for_date_range(pool, self.start, self.end).await?;
+        let rows = sqlx::query_as::<_, ValidatorReward>(VALIDATOR_REWRDS_QUERY)
+            .bind(blockspan.low)
+            .bind(blockspan.high)
+            .bind(&self.account)
+            .fetch(pool);
+        self.format.output(std::io::stdout(), rows).await?;
+        Ok(())
+    }
+}

src/cmd/rewards/mod.rs

@@ -2,20 +2,23 @@ use crate::{cmd::Opts, Result};
 use sqlx::PgPool;
 use structopt::StructOpt;
 
+mod account;
 mod hex;
 mod network;
 
 #[derive(Debug, StructOpt)]
 pub enum Cmd {
     Hex(hex::Cmd),
     Network(network::Cmd),
+    Account(account::Cmd),
 }
 
 impl Cmd {
     pub async fn run(&self, pool: &PgPool, opts: Opts) -> Result {
         match self {
             Self::Hex(cmd) => cmd.run(pool, opts).await,
             Self::Network(cmd) => cmd.run(pool, opts).await,
+            Self::Account(cmd) => cmd.run(pool, opts).await,
         }
     }
 }

src/timespan.rs

@@ -15,6 +15,12 @@ impl TimeSpan {
     pub fn new<S: ToDateTimeUtc>(start: S, days: i64) -> Self {
         let start = start.to_datetime();
         let end = start + Duration::days(days);
+        Self::for_date_range(start, end)
+    }
+
+    pub fn for_date_range<S: ToDateTimeUtc, E: ToDateTimeUtc>(start: S, end: E) -> Self {
+        let start = start.to_datetime();
+        let end = end.to_datetime();
         Self {
             low: std::cmp::min(start, end),
             high: std::cmp::max(start, end),