diff --git a/src/reducers/mod.rs b/src/reducers/mod.rs index 66140e39..2cac97bf 100644 --- a/src/reducers/mod.rs +++ b/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), } -} + } } diff --git a/src/reducers/utxo_by_nft.rs b/src/reducers/utxo_by_nft.rs new file mode 100644 index 00000000..57533396 --- /dev/null +++ b/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, +} + +pub struct Reducer { + config: Config, +} + +impl Reducer { + fn process_received_asset( + &mut self, + tx: &MultiEraTx, + txo_idx: usize, + policy: Hash<28>, + asset: Vec, + 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) + } +}