feat: make pools history endpoint

common/src/messages.rs

@@ -168,6 +168,15 @@ pub struct SPOStakeDistributionMessage {
     pub spos: Vec<(KeyHash, DelegatedStake)>,
 }
 
+#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
+pub struct SPORewardsMessage {
+    /// Epoch which has ended
+    pub epoch: u64,
+
+    /// SPO rewards by operator ID (total rewards before distribution, pool operator's rewards)
+    pub spos: Vec<(KeyHash, SPORewards)>,
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ProtocolParamsMessage {
     pub params: ProtocolParams,
@@ -224,6 +233,7 @@ pub enum CardanoMessage {
     // Stake distribution info
     DRepStakeDistribution(DRepStakeDistributionMessage), // Info about drep stake
     SPOStakeDistribution(SPOStakeDistributionMessage),   // SPO delegation distribution (SPDD)
+    SPORewards(SPORewardsMessage),                       // SPO rewards distribution (SPRD)
     StakeAddressDeltas(StakeAddressDeltasMessage),       // Stake part of address deltas
 }
 

common/src/queries/pools.rs

@@ -1,4 +1,4 @@
-use crate::PoolRegistration;
+use crate::{PoolEpochState, PoolRegistration};
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum PoolsStateQuery {
@@ -94,7 +94,9 @@ pub struct PoolsTotalBlocksMinted {
 pub struct PoolInfo {}
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct PoolHistory {}
+pub struct PoolHistory {
+    pub history: Vec<PoolEpochState>,
+}
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct PoolMetadata {}

common/src/types.rs

@@ -471,6 +471,18 @@ pub struct PoolRetirement {
     pub epoch: u64,
 }
 
+/// Pool Epoch History Data
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct PoolEpochState {
+    pub epoch: u64,
+    pub blocks_minted: u64,
+    pub active_stake: u64,
+    pub active_size: RationalNumber,
+    pub delegators_count: u64,
+    pub pool_reward: u64,
+    pub spo_reward: u64,
+}
+
 /// Stake delegation data
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct StakeDelegation {
@@ -487,10 +499,23 @@ pub struct DelegatedStake {
     /// Active stake - UTXO values only (used for reward calcs)
     pub active: Lovelace,
 
+    /// Active delegators count - delegators making active stakes (used for pool history)
+    pub active_delegators_count: u64,
+
     /// Total 'live' stake - UTXO values and rewards (used for VRF)
     pub live: Lovelace,
 }
 
+/// SPO rewards data (for SPORewardsMessage)
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct SPORewards {
+    /// Total rewards before distribution
+    pub total_rewards: Lovelace,
+
+    /// Pool operator's rewards
+    pub operator_rewards: Lovelace,
+}
+
 /// Genesis key delegation
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct GenesisKeyDelegation {

modules/accounts_state/src/accounts_state.rs

@@ -19,6 +19,8 @@ mod drep_distribution_publisher;
 use drep_distribution_publisher::DRepDistributionPublisher;
 mod spo_distribution_publisher;
 use spo_distribution_publisher::SPODistributionPublisher;
+mod spo_rewards_publisher;
+use spo_rewards_publisher::SPORewardsPublisher;
 mod state;
 use state::State;
 mod monetary;
@@ -39,6 +41,7 @@ const DEFAULT_STAKE_DELTAS_TOPIC: &str = "cardano.stake.deltas";
 const DEFAULT_DREP_STATE_TOPIC: &str = "cardano.drep.state";
 const DEFAULT_DREP_DISTRIBUTION_TOPIC: &str = "cardano.drep.distribution";
 const DEFAULT_SPO_DISTRIBUTION_TOPIC: &str = "cardano.spo.distribution";
+const DEFAULT_SPO_REWARDS_TOPIC: &str = "cardano.spo.rewards";
 const DEFAULT_PROTOCOL_PARAMETERS_TOPIC: &str = "cardano.protocol.parameters";
 
 const DEFAULT_HANDLE_SPDD_TOPIC: (&str, &str) = ("handle-topic-spdd", "rest.get.spdd");
@@ -61,6 +64,7 @@ impl AccountsState {
         history: Arc<Mutex<StateHistory<State>>>,
         mut drep_publisher: DRepDistributionPublisher,
         mut spo_publisher: SPODistributionPublisher,
+        mut spo_rewards_publisher: SPORewardsPublisher,
         mut spos_subscription: Box<dyn Subscription<Message>>,
         mut ea_subscription: Box<dyn Subscription<Message>>,
         mut certs_subscription: Box<dyn Subscription<Message>>,
@@ -196,11 +200,20 @@ impl AccountsState {
                         );
                         async {
                             Self::check_sync(&current_block, &block_info);
-                            state
+                            let spo_rewards = state
                                 .handle_epoch_activity(ea_msg)
                                 .await
                                 .inspect_err(|e| error!("EpochActivity handling error: {e:#}"))
                                 .ok();
+                            // SPO rewards is for previous epoch
+                            if let Some(spo_rewards) = spo_rewards {
+                                if let Err(e) = spo_rewards_publisher
+                                    .publish_spo_rewards(block_info, spo_rewards)
+                                    .await
+                                {
+                                    error!("Error publishing SPO rewards: {e:#}")
+                                }
+                            }
                         }
                         .instrument(span)
                         .await;
@@ -374,6 +387,10 @@ impl AccountsState {
             .get_string("publish-spo-distribution-topic")
             .unwrap_or(DEFAULT_SPO_DISTRIBUTION_TOPIC.to_string());
 
+        let spo_rewards_topic = config
+            .get_string("publish-spo-rewards-topic")
+            .unwrap_or(DEFAULT_SPO_REWARDS_TOPIC.to_string());
+
         // REST handler topics
         let handle_spdd_topic = config
             .get_string(DEFAULT_HANDLE_SPDD_TOPIC.0)
@@ -489,6 +506,7 @@ impl AccountsState {
         let drep_publisher =
             DRepDistributionPublisher::new(context.clone(), drep_distribution_topic);
         let spo_publisher = SPODistributionPublisher::new(context.clone(), spo_distribution_topic);
+        let spo_rewards_publisher = SPORewardsPublisher::new(context.clone(), spo_rewards_topic);
 
         // Subscribe
         let spos_subscription = context.subscribe(&spo_state_topic).await?;
@@ -506,6 +524,7 @@ impl AccountsState {
                 history,
                 drep_publisher,
                 spo_publisher,
+                spo_rewards_publisher,
                 spos_subscription,
                 ea_subscription,
                 certs_subscription,

modules/accounts_state/src/rewards.rs

@@ -3,7 +3,7 @@
 use crate::snapshot::{Snapshot, SnapshotSPO};
 use acropolis_common::{
     protocol_params::ShelleyParams, rational_number::RationalNumber, KeyHash, Lovelace,
-    RewardAccount,
+    RewardAccount, SPORewards,
 };
 use anyhow::{bail, Result};
 use bigdecimal::{BigDecimal, One, ToPrimitive, Zero};
@@ -19,6 +19,9 @@ pub struct RewardsResult {
 
     /// Rewards to be paid
     pub rewards: Vec<(RewardAccount, Lovelace)>,
+
+    /// SPO rewards
+    pub spo_rewards: Vec<(KeyHash, SPORewards)>,
 }
 
 /// State for rewards calculation
@@ -241,6 +244,13 @@ impl RewardsState {
             costs.to_u64().unwrap_or(0)
         };
         result.rewards.push((spo.reward_account.clone(), spo_benefit));
+        result.spo_rewards.push((
+            operator_id.clone(),
+            SPORewards {
+                total_rewards: pool_rewards.to_u64().unwrap_or(0),
+                operator_rewards: spo_benefit,
+            },
+        ));
         result.total_paid += spo_benefit;
         *total_paid_to_pools += spo_benefit;
     }

modules/accounts_state/src/spo_rewards_publisher.rs

@@ -0,0 +1,41 @@
+use acropolis_common::messages::{CardanoMessage, Message, SPORewardsMessage};
+use acropolis_common::{BlockInfo, KeyHash, SPORewards};
+use caryatid_sdk::Context;
+use std::sync::Arc;
+
+/// Message publisher for Stake Pool Delegation Distribution (SPDD)
+pub struct SPORewardsPublisher {
+    /// Module context
+    context: Arc<Context<Message>>,
+
+    /// Topic to publish on
+    topic: String,
+}
+
+impl SPORewardsPublisher {
+    /// Construct with context and topic to publish on
+    pub fn new(context: Arc<Context<Message>>, topic: String) -> Self {
+        Self { context, topic }
+    }
+
+    /// Publish the SPO rewards
+    pub async fn publish_spo_rewards(
+        &mut self,
+        block: &BlockInfo,
+        spo_rewards: Vec<(KeyHash, SPORewards)>,
+    ) -> anyhow::Result<()> {
+        self.context
+            .message_bus
+            .publish(
+                &self.topic,
+                Arc::new(Message::Cardano((
+                    block.clone(),
+                    CardanoMessage::SPORewards(SPORewardsMessage {
+                        epoch: block.epoch - 1, // End of previous epoch
+                        spos: spo_rewards.into_iter().collect(),
+                    }),
+                ))),
+            )
+            .await
+    }
+}

modules/accounts_state/src/state.rs

@@ -2,6 +2,7 @@
 use crate::monetary::calculate_monetary_change;
 use crate::rewards::{RewardsResult, RewardsState};
 use crate::snapshot::Snapshot;
+use acropolis_common::SPORewards;
 use acropolis_common::{
     messages::{
         DRepStateMessage, EpochActivityMessage, PotDeltasMessage, ProtocolParamsMessage,
@@ -382,10 +383,11 @@ impl State {
 
     /// Derive the Stake Pool Delegation Distribution (SPDD) - a map of total stake values
     /// (both with and without rewards) for each active SPO
+    /// And Stake Pool Reward State (rewards and delegators_count for each pool)
     /// Key of returned map is the SPO 'operator' ID
     pub fn generate_spdd(&self) -> BTreeMap<KeyHash, DelegatedStake> {
         // Shareable Dashmap with referenced keys
-        let spo_stakes = Arc::new(DashMap::<KeyHash, DelegatedStake>::new());
+        let spo_stakes = DashMap::<KeyHash, DelegatedStake>::new();
 
         // Total stake across all addresses in parallel, first collecting into a vector
         // because imbl::OrdMap doesn't work in Rayon
@@ -402,20 +404,20 @@ impl State {
         // Parallel sum all the stakes into the spo_stake map
         sas_data
             .par_iter() // Rayon multi-threaded iterator
-            .for_each_init(
-                || Arc::clone(&spo_stakes),
-                |map, (spo, (utxo_value, rewards))| {
-                    map.entry(spo.clone())
-                        .and_modify(|v| {
-                            v.active += *utxo_value;
-                            v.live += *utxo_value + *rewards;
-                        })
-                        .or_insert(DelegatedStake {
-                            active: *utxo_value,
-                            live: *utxo_value + *rewards,
-                        });
-                },
-            );
+            .for_each(|(spo, (utxo_value, rewards))| {
+                spo_stakes
+                    .entry(spo.clone())
+                    .and_modify(|v| {
+                        v.active += *utxo_value;
+                        v.active_delegators_count += 1;
+                        v.live += *utxo_value + *rewards;
+                    })
+                    .or_insert(DelegatedStake {
+                        active: *utxo_value,
+                        active_delegators_count: 1,
+                        live: *utxo_value + *rewards,
+                    });
+            });
 
         // Collect into a plain BTreeMap, so that it is ordered on output
         spo_stakes.iter().map(|entry| (entry.key().clone(), entry.value().clone())).collect()
@@ -491,7 +493,12 @@ impl State {
 
     /// Handle an EpochActivityMessage giving total fees and block counts by VRF key for
     /// the just-ended epoch
-    pub async fn handle_epoch_activity(&mut self, ea_msg: &EpochActivityMessage) -> Result<()> {
+    /// This also returns SPO rewards for publishing to the SPDD topic (For epoch N)
+    pub async fn handle_epoch_activity(
+        &mut self,
+        ea_msg: &EpochActivityMessage,
+    ) -> Result<Vec<(KeyHash, SPORewards)>> {
+        let mut spo_rewards: Vec<(KeyHash, SPORewards)> = Vec::new();
         // Reverse map of VRF key to SPO operator ID
         let vrf_to_operator: HashMap<KeyHash, KeyHash> =
             self.spos.iter().map(|(id, spo)| (spo.vrf_key_hash.clone(), id.clone())).collect();
@@ -524,14 +531,19 @@ impl State {
                         self.add_to_reward(&account, amount);
                     }
 
+                    // save SPO rewards
+                    spo_rewards = reward_result.spo_rewards.into_iter().collect();
+
                     // Adjust the reserves for next time with amount actually paid
                     self.pots.reserves -= reward_result.total_paid;
                 }
                 _ => (),
             }
         };
         // Enter epoch - note the message specifies the epoch that has just *ended*
-        self.enter_epoch(ea_msg.epoch + 1, ea_msg.total_fees, spo_block_counts)
+        self.enter_epoch(ea_msg.epoch + 1, ea_msg.total_fees, spo_block_counts)?;
+
+        Ok(spo_rewards)
     }
 
     /// Handle an SPOStateMessage with the full set of SPOs valid at the end of the last