feat: add total reward for epoch

orm/migrations/2024-12-16-231256_add_epoch_pos_rewards_table/down.sql

@@ -0,0 +1 @@
+ALTER TABLE pos_rewards DROP COLUMN epoch;

orm/migrations/2024-12-16-231256_add_epoch_pos_rewards_table/up.sql

@@ -0,0 +1,6 @@
+-- Alters the pos_rewards table to add a fourth column:
+-- epoch: INTEGER, representing the epoch number
+-- Populate existing records with epoch = 0
+ALTER TABLE pos_rewards ADD COLUMN epoch INTEGER NOT NULL DEFAULT 0;
+-- Now we can safely drop the default
+ALTER TABLE pos_rewards ALTER COLUMN epoch DROP DEFAULT;

orm/src/pos_rewards.rs

@@ -1,29 +1,26 @@
-use std::str::FromStr;
-
 use bigdecimal::BigDecimal;
 use diesel::{Insertable, Queryable, Selectable};
-use shared::rewards::Reward;
-
 use crate::schema::pos_rewards;
+use shared::rewards::Reward;
 
-#[derive(Insertable, Clone, Queryable, Selectable)]
+#[derive(Insertable, Queryable, Selectable, Clone)]
 #[diesel(table_name = pos_rewards)]
-#[diesel(check_for_backend(diesel::pg::Pg))]
 pub struct PosRewardInsertDb {
     pub owner: String,
     pub validator_id: i32,
     pub raw_amount: BigDecimal,
+    pub epoch: i32,
 }
 
 pub type PoSRewardDb = PosRewardInsertDb;
 
 impl PosRewardInsertDb {
-    pub fn from_reward(reward: Reward, validator_id: i32) -> Self {
-        Self {
+    pub fn from_reward(reward: Reward, validator_id: i32, epoch: i32) -> Self {
+        PosRewardInsertDb {
             owner: reward.delegation_pair.delegator_address.to_string(),
-            raw_amount: BigDecimal::from_str(&reward.amount.to_string())
-                .expect("Invalid amount"),
             validator_id,
+            raw_amount: BigDecimal::from(reward.amount),
+            epoch,
         }
     }
 }

orm/src/schema.rs

@@ -1,91 +1,47 @@
 // @generated automatically by Diesel CLI.
 
 pub mod sql_types {
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "crawler_name"))]
     pub struct CrawlerName;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "governance_kind"))]
     pub struct GovernanceKind;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "governance_result"))]
     pub struct GovernanceResult;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "governance_tally_type"))]
     pub struct GovernanceTallyType;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "payment_kind"))]
     pub struct PaymentKind;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "payment_recurrence"))]
     pub struct PaymentRecurrence;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "token_type"))]
     pub struct TokenType;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "transaction_kind"))]
     pub struct TransactionKind;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "transaction_result"))]
     pub struct TransactionResult;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "validator_state"))]
     pub struct ValidatorState;
 
-    #[derive(
-        diesel::query_builder::QueryId,
-        std::fmt::Debug,
-        diesel::sql_types::SqlType,
-    )]
+    #[derive(diesel::query_builder::QueryId, std::fmt::Debug, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "vote_kind"))]
     pub struct VoteKind;
 }
@@ -227,6 +183,7 @@ diesel::table! {
         owner -> Varchar,
         validator_id -> Int4,
         raw_amount -> Numeric,
+        epoch -> Int4,
     }
 }
 

rewards/src/main.rs

@@ -83,6 +83,9 @@ async fn crawling_fn(
 
     tracing::info!("Starting to update proposals...");
 
+    let current_epoch = namada_service::get_current_epoch(&client)
+    .await
+    .into_rpc_error()?;
     // TODO: change this by querying all the pairs in the database
     let delegations_pairs = namada_service::query_delegation_pairs(&client)
         .await
@@ -100,18 +103,19 @@ async fn crawling_fn(
 
     conn.interact(move |conn| {
         conn.build_transaction().read_write().run(
-            |transaction_conn: &mut diesel::prelude::PgConnection| {
+            |transaction_conn: &mut diesel::pg::PgConnection| {
                 repository::pos_rewards::upsert_rewards(
                     transaction_conn,
                     non_zero_rewards,
+                    current_epoch as i32,
                 )?;
 
                 repository::crawler_state::upsert_crawler_state(
                     transaction_conn,
                     (CrawlerName::Rewards, crawler_state).into(),
                 )?;
 
-                anyhow::Ok(())
+                Ok(())
             },
         )
     })

rewards/src/repository/pos_rewards.rs

@@ -7,30 +7,29 @@ use shared::rewards::Reward;
 pub fn upsert_rewards(
     transaction_conn: &mut PgConnection,
     rewards: Vec<Reward>,
+    epoch: i32, // Add an epoch parameter
 ) -> anyhow::Result<()> {
     diesel::insert_into(pos_rewards::table)
         .values::<Vec<PosRewardInsertDb>>(
             rewards
                 .into_iter()
                 .map(|reward| {
                     let validator_id: i32 = validators::table
-                        .filter(
-                            validators::namada_address.eq(&reward
-                                .delegation_pair
-                                .validator_address
-                                .to_string()),
-                        )
+                        .filter(validators::namada_address.eq(
+                            &reward.delegation_pair.validator_address.to_string(),
+                        ))
                         .select(validators::id)
                         .first(transaction_conn)
                         .expect("Failed to get validator");
 
-                    PosRewardInsertDb::from_reward(reward, validator_id)
+                    PosRewardInsertDb::from_reward(reward, validator_id, epoch)
                 })
                 .collect::<Vec<_>>(),
         )
         .on_conflict((
             pos_rewards::columns::owner,
             pos_rewards::columns::validator_id,
+            pos_rewards::columns::epoch, // Add epoch to conflict target
         ))
         .do_update()
         .set(

rewards/src/services/namada.rs

@@ -37,6 +37,9 @@ pub async fn query_rewards(
     client: &HttpClient,
     delegation_pairs: HashSet<DelegationPair>,
 ) -> anyhow::Result<Vec<Reward>> {
+    let epoch = rpc::query_epoch(client)
+        .await
+        .context("Failed to query Namada's current epoch")?;
     Ok(futures::stream::iter(delegation_pairs)
         .filter_map(|delegation| async move {
             tracing::info!(
@@ -64,6 +67,7 @@ pub async fn query_rewards(
             Some(Reward {
                 delegation_pair: delegation,
                 amount: Amount::from(reward),
+                epoch: epoch.0 as i32,
             })
         })
         .map(futures::future::ready)

seeder/src/main.rs

@@ -178,7 +178,8 @@ async fn main() -> anyhow::Result<(), MainError> {
                             .into_iter()
                             .map(|reward| {
                                 let validator_id = reward.delegation_pair.validator_address.to_string().parse::<i32>().unwrap();
-                                PosRewardInsertDb::from_reward(reward, validator_id)
+                                let epoch = reward.epoch;
+                                PosRewardInsertDb::from_reward(reward, validator_id, epoch)
                             })
                             .collect::<Vec<_>>(),
                     )

shared/src/balance.rs

@@ -1,4 +1,5 @@
 use std::fmt::Display;
+use std::str::FromStr;
 
 use bigdecimal::BigDecimal;
 use fake::Fake;
@@ -28,6 +29,13 @@ impl From<BigDecimal> for Amount {
     }
 }
 
+impl From<Amount> for BigDecimal {
+    fn from(amount: Amount) -> BigDecimal {
+        BigDecimal::from_str(&amount.0.to_string_native())
+            .expect("Invalid amount")
+    }
+}
+
 impl Display for Amount {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.0)
@@ -114,3 +122,41 @@ impl Balance {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::balance::{Amount, NamadaAmount};
+    use bigdecimal::BigDecimal;
+    use namada_sdk::token::NATIVE_MAX_DECIMAL_PLACES;
+    use std::str::FromStr;
+    #[test]
+    fn test_bigquery_amount_round_trip_integer() {
+        let bigdecimal =
+            BigDecimal::from(100).with_scale(NATIVE_MAX_DECIMAL_PLACES.into());
+        let amount = Amount::from(bigdecimal.clone());
+        let bigdecimal_from_amount = BigDecimal::from(amount.clone());
+        assert_eq!(amount, Amount::from(bigdecimal_from_amount.clone()));
+        assert_eq!(bigdecimal, bigdecimal_from_amount);
+    }
+    #[test]
+    fn test_bigquery_amount_round_trip_decimal() {
+        let bigdecimal = BigDecimal::from_str("100.123456").unwrap();
+        let amount = Amount::from(bigdecimal.clone());
+        let bigdecimal_from_amount = BigDecimal::from(amount.clone());
+        assert_eq!(amount, Amount::from(bigdecimal_from_amount.clone()));
+        assert_eq!(bigdecimal, bigdecimal_from_amount);
+    }
+    #[test]
+    fn test_amount_same_as_namada_amount_integer() {
+        let amount = Amount::from(BigDecimal::from(100));
+        let namada_amount = NamadaAmount::from_u64(100);
+        assert_eq!(amount.0, namada_amount);
+    }
+    #[test]
+    fn test_amount_same_as_namada_amount_decimal() {
+        let amount = Amount::from(BigDecimal::from_str("100.123").unwrap());
+        let namada_amount = NamadaAmount::from_string_precise("100.123")
+            .expect("Invalid amount");
+        assert_eq!(amount.0, namada_amount);
+    }
+}

shared/src/rewards.rs

@@ -6,6 +6,7 @@ use crate::utils::DelegationPair;
 pub struct Reward {
     pub delegation_pair: DelegationPair,
     pub amount: Amount,
+    pub epoch: i32,
 }
 
 impl Reward {
@@ -19,6 +20,7 @@ impl Reward {
                 delegator_address: Id::Account(delegator_address.to_string()),
             },
             amount: Amount::fake(),
+            epoch: 0,
         }
     }
 }