feat: address UTxOs of a specific asset endpoint

modules/rest_blockfrost/src/handlers/addresses.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use crate::types::{AddressTotalsREST, TransactionInfoREST, UTxOREST};
+use crate::utils::split_policy_and_asset;
 use crate::{handlers_config::HandlersConfig, types::AddressInfoREST};
 use acropolis_common::queries::blocks::{BlocksStateQuery, BlocksStateQueryResponse};
 use acropolis_common::queries::errors::QueryError;
@@ -261,11 +262,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, AssetMetadata, 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,
-    AssetMetadataStandard, AssetName, PolicyId,
+    AssetMetadataStandard, PolicyId,
 };
 use blake2::{digest::consts::U20, Blake2b, Digest};
 use caryatid_sdk::Context;
@@ -295,28 +296,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,

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::*;