feat: reward delta processing in historical accounts state

Signed-off-by: William Hankins <[email protected]>

Author: whankinsiv

common/src/queries/accounts.rs

@@ -1,6 +1,8 @@
 use std::collections::HashMap;
 
-use crate::{DRepChoice, KeyHash, PoolId, PoolLiveStakeInfo, StakeAddress, TxIdentifier};
+use crate::{
+    DRepChoice, KeyHash, PoolId, PoolLiveStakeInfo, RewardType, StakeAddress, TxIdentifier,
+};
 
 pub const DEFAULT_ACCOUNTS_QUERY_TOPIC: (&str, &str) =
     ("accounts-state-query-topic", "cardano.query.accounts");
@@ -158,7 +160,7 @@ pub struct RewardHistory {
     #[n(2)]
     pub pool: PoolId,
     #[n(3)]
-    pub is_owner: bool,
+    pub reward_type: RewardType,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

common/src/types.rs

@@ -217,10 +217,15 @@ pub struct StakeRewardDelta {
 }
 
 /// Type of reward being given
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(
+    Debug, Clone, minicbor::Encode, minicbor::Decode, serde::Serialize, serde::Deserialize,
+)]
 pub enum RewardType {
+    #[n(0)]
     Leader,
+    #[n(1)]
     Member,
+    #[n(2)]
     PoolRefund,
 }
 

modules/historical_accounts_state/src/historical_accounts_state.rs

@@ -126,10 +126,7 @@ impl HistoricalAccountsState {
                 {
                     Self::check_sync(&current_block, block_info);
                     let mut state = state_mutex.lock().await;
-                    state
-                        .handle_rewards(rewards_msg)
-                        .inspect_err(|e| error!("Reward deltas handling error: {e:#}"))
-                        .ok();
+                    state.handle_rewards(rewards_msg, block_info.epoch as u32);
                 }
             }
 

modules/historical_accounts_state/src/state.rs

@@ -104,8 +104,18 @@ impl State {
             && block_info.number > self.volatile.epoch_start_block + self.volatile.security_param_k
     }
 
-    pub fn handle_rewards(&mut self, _reward_deltas: &StakeRewardDeltasMessage) -> Result<()> {
-        Ok(())
+    pub fn handle_rewards(&mut self, reward_deltas: &StakeRewardDeltasMessage, epoch: u32) {
+        let volatile = self.volatile.window.back_mut().expect("window should never be empty");
+        for reward in reward_deltas.deltas.iter() {
+            let entry = volatile.entry(reward.stake_address.clone()).or_default();
+            let update = RewardHistory {
+                epoch,
+                amount: reward.delta,
+                pool: Vec::new(),
+                reward_type: reward.reward_type.clone(),
+            };
+            entry.reward_history.get_or_insert_with(Vec::new).push(update);
+        }
     }
 
     pub fn handle_tx_certificates(&mut self, tx_certs: &TxCertificatesMessage, epoch: u32) {