[WIP] AddressByStake reducer

src/reducers/address_by_stake.rs

@@ -0,0 +1,58 @@
+use pallas::ledger::addresses::Address::Shelley;
+use pallas::ledger::addresses::{Error, StakeAddress};
+use pallas::ledger::traverse::MultiEraBlock;
+use serde::Deserialize;
+
+use crate::{model, prelude::*};
+
+#[derive(Deserialize)]
+pub struct Config {
+    pub key_prefix: Option<String>,
+}
+
+pub struct Reducer {
+    config: Config,
+}
+
+impl Reducer {
+    pub fn reduce_block<'b>(
+        &mut self,
+        block: &'b MultiEraBlock<'b>,
+        output: &mut super::OutputPort,
+    ) -> Result<(), gasket::error::Error> {
+        for tx in block.txs().into_iter() {
+            for (_, out) in tx.produces().into_iter() {
+                let address = out.address().or_panic()?;
+                match address {
+                    Shelley(shelly_address) => {
+                        let stake_address_result: Result<StakeAddress, Error> =
+                            shelly_address.clone().try_into();
+                        if stake_address_result.is_ok() {
+                            let stake_address =
+                                stake_address_result.unwrap().to_bech32().or_panic()?;
+                            let payment_address = shelly_address.to_bech32().or_panic()?;
+
+                            let crdt = model::CRDTCommand::set_add(
+                                self.config.key_prefix.as_deref(),
+                                &stake_address,
+                                payment_address,
+                            );
+                            output.send(crdt.into());
+                        }
+                    }
+                    _ => (),
+                };
+            }
+        }
+
+        Ok(())
+    }
+}
+
+impl Config {
+    pub fn plugin(self) -> super::Reducer {
+        let reducer = Reducer { config: self };
+
+        super::Reducer::AddressByStake(reducer)
+    }
+}

src/reducers/mod.rs

@@ -35,6 +35,8 @@ pub mod tx_count_by_address;
 pub mod tx_count_by_native_token_policy_id;
 #[cfg(feature = "unstable")]
 pub mod utxo_by_nft;
+#[cfg(feature = "unstable")]
+pub mod address_by_stake;
 
 #[derive(Deserialize)]
 #[serde(tag = "type")]
@@ -63,6 +65,8 @@ pub enum Config {
     AssetHoldersByAsset(asset_holders_by_asset_id::Config),
     #[cfg(feature = "unstable")]
     UtxoByNft(utxo_by_nft::Config),
+    #[cfg(feature = "unstable")]
+    AddressByStake(address_by_stake::Config)
 }
 
 impl Config {
@@ -96,6 +100,8 @@ impl Config {
             Config::AssetHoldersByAsset(c) => c.plugin(chain, policy),
             #[cfg(feature = "unstable")]
             Config::UtxoByNft(c) => c.plugin(policy),
+            #[cfg(feature = "unstable")]
+            Config::AddressByStake(c) => c.plugin(),
         }
     }
 }
@@ -170,6 +176,8 @@ pub enum Reducer {
     AssetHoldersByAssetId(asset_holders_by_asset_id::Reducer),
     #[cfg(feature = "unstable")]
     UtxoByNft(utxo_by_nft::Reducer),
+    #[cfg(feature = "unstable")]
+    AddressByStake(address_by_stake::Reducer),
 }
 
 impl Reducer {
@@ -204,6 +212,8 @@ impl Reducer {
             Reducer::AssetHoldersByAssetId(x) => x.reduce_block(block, ctx, output),
             #[cfg(feature = "unstable")]
             Reducer::UtxoByNft(x) => x.reduce_block(block, ctx, output),
+            #[cfg(feature = "unstable")]
+            Reducer::AddressByStake(x) => x.reduce_block(block, output),
         }
     }
 }