Merge pull request #373 from input-output-hk/whankinsiv/address-asset-utxos-endpoint

feat: address UTxOs of a specific asset endpoint

Author: whankinsiv

modules/rest_blockfrost/src/handlers/addresses.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use crate::types::{AddressInfoExtended, AddressTotalsREST, TransactionInfoREST, UTxOREST};
+use crate::utils::split_policy_and_asset;
 use crate::{handlers_config::HandlersConfig, types::AddressInfoREST};
 use acropolis_common::queries::assets::{AssetsStateQuery, AssetsStateQueryResponse};
 use acropolis_common::queries::blocks::{BlocksStateQuery, BlocksStateQueryResponse};
@@ -379,11 +380,116 @@ pub async fn handle_address_utxos_blockfrost(
 
 /// Handle `/addresses/{address}/utxos/{asset}` Blockfrost-compatible endpoint
 pub async fn handle_address_asset_utxos_blockfrost(
-    _context: Arc<Context<Message>>,
-    _params: Vec<String>,
-    _handlers_config: Arc<HandlersConfig>,
+    context: Arc<Context<Message>>,
+    params: Vec<String>,
+    handlers_config: Arc<HandlersConfig>,
 ) -> Result<RESTResponse, RESTError> {
-    Err(RESTError::not_implemented("Address asset UTxOs endpoint"))
+    let address = parse_address(&params)?;
+    let address_str = address.to_string()?;
+    let (target_policy, target_name) = split_policy_and_asset(&params[1])?;
+
+    // Get utxos from address state
+    let msg = Arc::new(Message::StateQuery(StateQuery::Addresses(
+        AddressStateQuery::GetAddressUTxOs { address },
+    )));
+    let utxo_identifiers = query_state(
+        &context,
+        &handlers_config.addresses_query_topic,
+        msg,
+        |message| match message {
+            Message::StateQueryResponse(StateQueryResponse::Addresses(
+                AddressStateQueryResponse::AddressUTxOs(utxos),
+            )) => Ok(utxos),
+            Message::StateQueryResponse(StateQueryResponse::Addresses(
+                AddressStateQueryResponse::Error(e),
+            )) => Err(e),
+            _ => Err(QueryError::internal_error(
+                "Unexpected message type while retrieving address UTxOs",
+            )),
+        },
+    )
+    .await?;
+
+    // Get UTxO balances from utxo state
+    let msg = Arc::new(Message::StateQuery(StateQuery::UTxOs(
+        UTxOStateQuery::GetUTxOs {
+            utxo_identifiers: utxo_identifiers.clone(),
+        },
+    )));
+    let entries = query_state(
+        &context,
+        &handlers_config.utxos_query_topic,
+        msg,
+        |message| match message {
+            Message::StateQueryResponse(StateQueryResponse::UTxOs(
+                UTxOStateQueryResponse::UTxOs(utxos),
+            )) => Ok(utxos),
+            Message::StateQueryResponse(StateQueryResponse::UTxOs(
+                UTxOStateQueryResponse::Error(e),
+            )) => Err(e),
+            _ => Err(QueryError::internal_error(
+                "Unexpected message type while retrieving UTxO entries",
+            )),
+        },
+    )
+    .await?;
+
+    // Filter for UTxOs which contain the asset
+    let mut filtered_identifiers = Vec::new();
+    let mut filtered_entries = Vec::new();
+
+    for (i, entry) in entries.iter().enumerate() {
+        let matches = entry.value.assets.iter().any(|(policy, assets)| {
+            policy == &target_policy && assets.iter().any(|asset| asset.name == target_name)
+        });
+
+        if matches {
+            filtered_identifiers.push(utxo_identifiers[i]);
+            filtered_entries.push(entry);
+        }
+    }
+
+    if filtered_identifiers.is_empty() {
+        return Ok(RESTResponse::with_json(200, "[]"));
+    }
+
+    // Get TxHashes and BlockHashes from subset of UTxOIdentifiers with specific asset balances
+    let msg = Arc::new(Message::StateQuery(StateQuery::Blocks(
+        BlocksStateQuery::GetUTxOHashes {
+            utxo_ids: filtered_identifiers.clone(),
+        },
+    )));
+    let hashes = query_state(
+        &context,
+        &handlers_config.blocks_query_topic,
+        msg,
+        |message| match message {
+            Message::StateQueryResponse(StateQueryResponse::Blocks(
+                BlocksStateQueryResponse::UTxOHashes(hashes),
+            )) => Ok(hashes),
+            Message::StateQueryResponse(StateQueryResponse::Blocks(
+                BlocksStateQueryResponse::Error(e),
+            )) => Err(e),
+            _ => Err(QueryError::internal_error(
+                "Unexpected message type while retrieving UTxO hashes",
+            )),
+        },
+    )
+    .await?;
+
+    let mut rest_response = Vec::with_capacity(filtered_entries.len());
+    for (i, entry) in filtered_entries.into_iter().enumerate() {
+        rest_response.push(UTxOREST::new(
+            address_str.clone(),
+            &filtered_identifiers[i],
+            entry,
+            hashes.tx_hashes[i].as_ref(),
+            hashes.block_hashes[i].as_ref(),
+        ))
+    }
+
+    let json = serde_json::to_string_pretty(&rest_response)?;
+    Ok(RESTResponse::with_json(200, &json))
 }
 
 /// Handle `/addresses/{address}/transactions` Blockfrost-compatible endpoint

modules/rest_blockfrost/src/handlers/assets.rs

@@ -4,6 +4,7 @@ use crate::{
         AssetAddressRest, AssetInfoRest, AssetMetadataREST, AssetMintRecordRest,
         AssetTransactionRest, PolicyAssetRest,
     },
+    utils::split_policy_and_asset,
 };
 use acropolis_common::queries::errors::QueryError;
 use acropolis_common::rest_error::RESTError;
@@ -14,7 +15,7 @@ use acropolis_common::{
         utils::query_state,
     },
     serialization::Bech32WithHrp,
-    AssetName, PolicyId,
+    PolicyId,
 };
 use blake2::{digest::consts::U20, Blake2b, Digest};
 use caryatid_sdk::Context;
@@ -298,28 +299,6 @@ pub async fn handle_policy_assets_blockfrost(
     Ok(RESTResponse::with_json(200, &json))
 }
 
-fn split_policy_and_asset(hex_str: &str) -> Result<(PolicyId, AssetName), RESTError> {
-    let decoded = hex::decode(hex_str)?;
-
-    if decoded.len() < 28 {
-        return Err(RESTError::BadRequest(
-            "Asset identifier must be at least 28 bytes".to_string(),
-        ));
-    }
-
-    let (policy_part, asset_part) = decoded.split_at(28);
-
-    let policy_id: PolicyId = policy_part
-        .try_into()
-        .map_err(|_| RESTError::BadRequest("Policy id must be 28 bytes".to_string()))?;
-
-    let asset_name = AssetName::new(asset_part).ok_or_else(|| {
-        RESTError::BadRequest("Asset name must be less than 32 bytes".to_string())
-    })?;
-
-    Ok((policy_id, asset_name))
-}
-
 pub async fn fetch_asset_metadata(
     asset: &str,
     offchain_registry_url: &str,
@@ -448,59 +427,3 @@ fn cbor_to_json(val: CborValue) -> Value {
         _ => Value::Null,
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use crate::handlers::assets::split_policy_and_asset;
-    use hex;
-
-    fn policy_bytes() -> [u8; 28] {
-        [0u8; 28]
-    }
-
-    #[test]
-    fn invalid_hex_string() {
-        let result = split_policy_and_asset("zzzz");
-        assert!(result.is_err());
-        let err = result.unwrap_err();
-        assert_eq!(err.status_code(), 400);
-        assert_eq!(
-            err.message(),
-            "Invalid hex string: Invalid character 'z' at position 0"
-        );
-    }
-
-    #[test]
-    fn too_short_input() {
-        let hex_str = hex::encode([1u8, 2, 3]);
-        let result = split_policy_and_asset(&hex_str);
-        assert!(result.is_err());
-        let err = result.unwrap_err();
-        assert_eq!(err.status_code(), 400);
-        assert_eq!(err.message(), "Asset identifier must be at least 28 bytes");
-    }
-
-    #[test]
-    fn invalid_asset_name_too_long() {
-        let mut bytes = policy_bytes().to_vec();
-        bytes.extend(vec![0u8; 33]);
-        let hex_str = hex::encode(bytes);
-        let result = split_policy_and_asset(&hex_str);
-        assert!(result.is_err());
-        let err = result.unwrap_err();
-        assert_eq!(err.status_code(), 400);
-        assert_eq!(err.message(), "Asset name must be less than 32 bytes");
-    }
-
-    #[test]
-    fn valid_policy_and_asset() {
-        let mut bytes = policy_bytes().to_vec();
-        bytes.extend_from_slice(b"MyToken");
-        let hex_str = hex::encode(bytes);
-        let result = split_policy_and_asset(&hex_str);
-        assert!(result.is_ok());
-        let (policy, name) = result.unwrap();
-        assert_eq!(policy, policy_bytes());
-        assert_eq!(name.as_slice(), b"MyToken");
-    }
-}

modules/rest_blockfrost/src/utils.rs

@@ -1,5 +1,6 @@
 use std::time::Duration;
 
+use acropolis_common::{rest_error::RESTError, AssetName, PolicyId};
 use anyhow::Result;
 use blake2::digest::{Update, VariableOutput};
 use reqwest::Client;
@@ -83,6 +84,29 @@ pub fn verify_pool_metadata_hash(
 fn invalid_size_desc<T: std::fmt::Display>(e: T) -> String {
     format!("Invalid size for hashing pool metadata json {e}")
 }
+
+pub fn split_policy_and_asset(hex_str: &str) -> Result<(PolicyId, AssetName), RESTError> {
+    let decoded = hex::decode(hex_str)?;
+
+    if decoded.len() < 28 {
+        return Err(RESTError::BadRequest(
+            "Asset identifier must be at least 28 bytes".to_string(),
+        ));
+    }
+
+    let (policy_part, asset_part) = decoded.split_at(28);
+
+    let policy_id: PolicyId = policy_part
+        .try_into()
+        .map_err(|_| RESTError::BadRequest("Policy id must be 28 bytes".to_string()))?;
+
+    let asset_name = AssetName::new(asset_part).ok_or_else(|| {
+        RESTError::BadRequest("Asset name must be less than 32 bytes".to_string())
+    })?;
+
+    Ok((policy_id, asset_name))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -142,4 +166,54 @@ mod tests {
             Ok(())
         );
     }
+
+    fn policy_bytes() -> [u8; 28] {
+        [0u8; 28]
+    }
+
+    #[test]
+    fn invalid_hex_string() {
+        let result = split_policy_and_asset("zzzz");
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.status_code(), 400);
+        assert_eq!(
+            err.message(),
+            "Invalid hex string: Invalid character 'z' at position 0"
+        );
+    }
+
+    #[test]
+    fn too_short_input() {
+        let hex_str = hex::encode([1u8, 2, 3]);
+        let result = split_policy_and_asset(&hex_str);
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.status_code(), 400);
+        assert_eq!(err.message(), "Asset identifier must be at least 28 bytes");
+    }
+
+    #[test]
+    fn invalid_asset_name_too_long() {
+        let mut bytes = policy_bytes().to_vec();
+        bytes.extend(vec![0u8; 33]);
+        let hex_str = hex::encode(bytes);
+        let result = split_policy_and_asset(&hex_str);
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.status_code(), 400);
+        assert_eq!(err.message(), "Asset name must be less than 32 bytes");
+    }
+
+    #[test]
+    fn valid_policy_and_asset() {
+        let mut bytes = policy_bytes().to_vec();
+        bytes.extend_from_slice(b"MyToken");
+        let hex_str = hex::encode(bytes);
+        let result = split_policy_and_asset(&hex_str);
+        assert!(result.is_ok());
+        let (policy, name) = result.unwrap();
+        assert_eq!(policy, policy_bytes());
+        assert_eq!(name.as_slice(), b"MyToken");
+    }
 }