feat: Introduce 'utxo by nft' reducer

src/reducers/mod.rs

@@ -20,6 +20,8 @@ pub mod address_by_ada_handle;
 #[cfg(feature = "unstable")]
 pub mod address_by_txo;
 #[cfg(feature = "unstable")]
+pub mod asset_holders_by_asset_id;
+#[cfg(feature = "unstable")]
 pub mod balance_by_address;
 #[cfg(feature = "unstable")]
 pub mod block_header_by_hash;
@@ -32,7 +34,7 @@ pub mod tx_count_by_address;
 #[cfg(feature = "unstable")]
 pub mod tx_count_by_native_token_policy_id;
 #[cfg(feature = "unstable")]
-pub mod asset_holders_by_asset_id;
+pub mod utxo_by_nft;
 
 #[derive(Deserialize)]
 #[serde(tag = "type")]
@@ -59,6 +61,8 @@ pub enum Config {
     TxCountByNativeTokenPolicyId(tx_count_by_native_token_policy_id::Config),
     #[cfg(feature = "unstable")]
     AssetHoldersByAsset(asset_holders_by_asset_id::Config),
+    #[cfg(feature = "unstable")]
+    UtxoByNft(utxo_by_nft::Config),
 }
 
 impl Config {
@@ -90,6 +94,8 @@ impl Config {
             Config::TxCountByNativeTokenPolicyId(c) => c.plugin(chain),
             #[cfg(feature = "unstable")]
             Config::AssetHoldersByAsset(c) => c.plugin(chain, policy),
+            #[cfg(feature = "unstable")]
+            Config::UtxoByNft(c) => c.plugin(policy),
         }
     }
 }
@@ -162,6 +168,8 @@ pub enum Reducer {
     TxCountByNativeTokenPolicyId(tx_count_by_native_token_policy_id::Reducer),
     #[cfg(feature = "unstable")]
     AssetHoldersByAssetId(asset_holders_by_asset_id::Reducer),
+    #[cfg(feature = "unstable")]
+    UtxoByNft(utxo_by_nft::Reducer),
 }
 
 impl Reducer {
@@ -194,6 +202,8 @@ impl Reducer {
             Reducer::TxCountByNativeTokenPolicyId(x) => x.reduce_block(block, output),
             #[cfg(feature = "unstable")]
             Reducer::AssetHoldersByAssetId(x) => x.reduce_block(block, ctx, output),
+            #[cfg(feature = "unstable")]
+            Reducer::UtxoByNft(x) => x.reduce_block(block, ctx, output),
         }
-}
+    }
 }

src/reducers/utxo_by_nft.rs

@@ -0,0 +1,67 @@
+use pallas::crypto::hash::Hash;
+use pallas::ledger::traverse::{Asset, MultiEraOutput};
+use pallas::ledger::traverse::{MultiEraBlock, MultiEraTx};
+use serde::Deserialize;
+
+use crate::{crosscut, model};
+
+#[derive(Deserialize)]
+pub struct Config {
+    pub key_prefix: Option<String>,
+}
+
+pub struct Reducer {
+    config: Config,
+}
+
+impl Reducer {
+    fn process_received_asset(
+        &mut self,
+        tx: &MultiEraTx,
+        txo_idx: usize,
+        policy: Hash<28>,
+        asset: Vec<u8>,
+        output: &mut super::OutputPort,
+    ) -> Result<(), gasket::error::Error> {
+        let tx_hash = tx.hash();
+
+        let crdt = model::CRDTCommand::any_write_wins(
+            self.config.key_prefix.as_deref(),
+            format!("{}{}", policy, hex::encode(asset)),
+            format!("{}#{}", tx_hash, txo_idx),
+        );
+
+        output.send(crdt.into())
+    }
+
+    pub fn reduce_block<'b>(
+        &mut self,
+        block: &'b MultiEraBlock<'b>,
+        _ctx: &model::BlockContext,
+        output: &mut super::OutputPort,
+    ) -> Result<(), gasket::error::Error> {
+        for tx in block.txs().into_iter() {
+            for (idx, txo) in tx.produces() {
+                for asset in txo.assets() {
+                    if let Asset::NativeAsset(policy, asset, quantity) = asset {
+                        // This check is to avoid indexing fungible tokens. This method will have
+                        // several false positives, but it's fast and provides a good approximation.
+                        if quantity == 1 {
+                            self.process_received_asset(&tx, idx, policy, asset, output)?;
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+}
+
+impl Config {
+    pub fn plugin(self, _policy: &crosscut::policies::RuntimePolicy) -> super::Reducer {
+        let reducer = Reducer { config: self };
+
+        super::Reducer::UtxoByNft(reducer)
+    }
+}