feat(json_rpc): implement limits on queryStakes

Author: guidiaz

node/src/actors/chain_manager/handlers.rs

@@ -9,6 +9,7 @@ use std::{
 use actix::{prelude::*, ActorFutureExt, WrapFuture};
 use futures::future::Either;
 
+use itertools::Itertools;
 use witnet_data_structures::{
     chain::{
         tapi::ActiveWips, Block, ChainInfo, ChainState, CheckpointBeacon, ConsensusConstantsWit2,
@@ -1542,10 +1543,39 @@ impl Handler<QueryStakes> for ChainManager {
     type Result = <QueryStakes as Message>::Result;
 
     fn handle(&mut self, msg: QueryStakes, _ctx: &mut Self::Context) -> Self::Result {
-        // build address from public key hash
-        let stakes = self.chain_state.stakes.query_stakes(msg.filter);
+        // query stakes from current chain state
+        let mut stakes = self
+            .chain_state
+            .stakes
+            .query_stakes(msg.filter)
+            .map_err(StakesError::from)?;
+        // filter out stake entries whose last mining and witnessing epochs are previous to `epoch`
+        let epoch: u32 = if msg.limits.since >= 0 {
+            msg.limits.since.try_into().inspect_err(|&e| {
+                log::warn!("Invalid 'since' limit on QueryStakes: {}", e);
+            })?
+        } else {
+            (self.current_epoch.unwrap() as i64 + msg.limits.since)
+                .try_into()
+                .unwrap_or_default()
+        };
+        stakes.retain(|stake| {
+            let nonce: u32 = stake.value.nonce.try_into().unwrap_or_default();
+            stake.value.epochs.mining >= epoch && stake.value.epochs.mining != nonce
+                || (stake.value.epochs.witnessing >= epoch
+                    && stake.value.epochs.witnessing != nonce)
+        });
+        if msg.limits.distinct {
+            // if only distinct validators are required, retain only first appearence in the stakes array:
+            let stakes = stakes
+                .into_iter()
+                .unique_by(|stake| stake.key.validator)
+                .collect();
 
-        stakes.map_err(StakesError::from).map_err(Into::into)
+            Ok(stakes)
+        } else {
+            Ok(stakes)
+        }
     }
 }
 

node/src/actors/json_rpc/api.rs

@@ -2247,32 +2247,10 @@ pub async fn authorize_stake(params: Result<AuthorizeStake, Error>) -> JsonRpcRe
 pub async fn query_stakes(params: Result<Option<QueryStakes>, Error>) -> JsonRpcResult {
     // Short-circuit if parameters are wrong
     let params = params?;
-    // If a withdrawer address is not specified, default to local node address
-    let filter: QueryStakesFilter = if let Some(address) = params {
-        match address {
-            QueryStakesParams::All(_) => QueryStakesFilter::All,
-            QueryStakesParams::Validator(validator) => QueryStakesFilter::Validator(
-                PublicKeyHash::from_bech32(get_environment(), &validator)
-                    .map_err(internal_error)?,
-            ),
-            QueryStakesParams::Withdrawer(withdrawer) => QueryStakesFilter::Withdrawer(
-                PublicKeyHash::from_bech32(get_environment(), &withdrawer)
-                    .map_err(internal_error)?,
-            ),
-            QueryStakesParams::Key((validator, withdrawer)) => QueryStakesFilter::Key((
-                PublicKeyHash::from_bech32(get_environment(), &validator)
-                    .map_err(internal_error)?,
-                PublicKeyHash::from_bech32(get_environment(), &withdrawer)
-                    .map_err(internal_error)?,
-            )),
-        }
-    } else {
-        let pk = signature_mngr::public_key().await.map_err(internal_error)?;
-
-        QueryStakesFilter::Validator(PublicKeyHash::from_public_key(&pk))
-    };
+    // Parse params or defaults:
+    let msg = params.unwrap_or(QueryStakes::default());
     ChainManager::from_registry()
-        .send(QueryStakes { filter })
+        .send(msg)
         .map(|res| match res {
             Ok(Ok(stakes)) => serde_json::to_value(stakes).map_err(internal_error),
             Ok(Err(e)) => {