Add handling for fetching nft asset

Author: gemcoder21

apps/daemon/src/consumers/store/store_transactions_consumer.rs

@@ -2,8 +2,8 @@ use std::collections::HashSet;
 use std::{collections::HashMap, error::Error};
 
 use async_trait::async_trait;
-use primitives::{AssetAddress, DeviceSubscription, Transaction, TransactionId, TransactionState, TransactionType};
-use storage::{AssetsAddressesRepository, AssetsRepository, Database, TransactionsRepository, WalletsRepository};
+use primitives::{AssetAddress, DeviceSubscription, NFTAssetId, Transaction, TransactionId, TransactionState, TransactionType};
+use storage::{AssetsAddressesRepository, AssetsRepository, Database, NftAssetFilter, NftRepository, TransactionsRepository, WalletsRepository};
 use streamer::{AssetId, NotificationsPayload, StreamProducer, StreamProducerQueue, TransactionsPayload, WalletStreamEvent, WalletStreamPayload, consumer::MessageConsumer};
 use swapper::cross_chain::{self, DepositAddressMap, SendAddressMap};
 
@@ -64,6 +64,17 @@ impl MessageConsumer<TransactionsPayload, usize> for StoreTransactionsConsumer {
 
         let _ = self.stream_producer.publish_fetch_assets(missing_assets).await;
 
+        let nft_asset_ids: Vec<NFTAssetId> = transactions
+            .iter()
+            .filter(|x| x.addresses().iter().any(|addr| subscription_addresses.contains(addr)))
+            .filter_map(|x| x.nft_asset_id())
+            .collect::<HashSet<_>>()
+            .into_iter()
+            .collect();
+
+        let missing_nft_assets = self.get_missing_nft_assets(nft_asset_ids).await?;
+        let _ = self.stream_producer.publish_fetch_nft_assets(missing_nft_assets).await;
+
         let mut transactions_map: HashMap<TransactionId, Transaction> = HashMap::new();
         let mut assets_addresses = HashSet::new();
         let mut notifications: Vec<NotificationsPayload> = Vec::new();
@@ -225,6 +236,16 @@ impl StoreTransactionsConsumer {
         Ok((assets_with_prices, missing_assets_ids))
     }
 
+    async fn get_missing_nft_assets(&self, nft_asset_ids: Vec<NFTAssetId>) -> Result<Vec<NFTAssetId>, Box<dyn Error + Send + Sync>> {
+        if nft_asset_ids.is_empty() {
+            return Ok(Vec::new());
+        }
+        let identifiers: Vec<String> = nft_asset_ids.iter().map(|id| id.to_string()).collect();
+        let existing = self.database.nft()?.get_nft_assets_by_filter(vec![NftAssetFilter::Identifiers(identifiers)])?;
+        let existing_ids: HashSet<NFTAssetId> = existing.into_iter().map(|row| row.identifier.0).collect();
+        Ok(nft_asset_ids.into_iter().filter(|id| !existing_ids.contains(id)).collect())
+    }
+
     async fn store_transactions(&self, transactions: Vec<Transaction>) -> Result<usize, Box<dyn Error + Send + Sync>> {
         if transactions.is_empty() {
             return Ok(0);

crates/gem_ton/src/provider/transactions_mapper.rs

@@ -59,79 +59,103 @@ pub fn map_trace_transactions(traces: Vec<Trace>) -> Vec<Transaction> {
     traces.into_iter().filter_map(map_root_trace_transaction).collect()
 }
 
+struct TransferDetails {
+    asset_id: AssetId,
+    from: String,
+    to: String,
+    value: String,
+    transaction_type: TransactionType,
+    memo: Option<String>,
+    metadata: Option<serde_json::Value>,
+}
+
 fn map_root_trace_transaction(trace: Trace) -> Option<Transaction> {
     let state = if trace.is_incomplete || trace.has_actions() { Some(trace.action_state()) } else { None };
-    let swap = jetton_swap(&trace.actions);
-    let jetton_transfer = jetton_transfer(&trace.actions);
     let mut transactions = trace.transactions;
     let root_hash = trace.transactions_order.into_iter().next()?;
     let root = transactions.remove(&root_hash)?;
 
-    if let Some(details) = jetton_transfer {
-        return map_jetton_transfer_transaction(&root, state, details);
-    }
-
-    let mut transaction = map_transaction_message_with_state(root, state)?;
-    if let Some((sender, metadata)) = swap
-        && let Ok(value) = serde_json::to_value(metadata)
-    {
-        transaction.transaction_type = TransactionType::Swap;
-        transaction.from = sender.clone();
-        transaction.to = sender;
-        transaction.metadata = Some(value);
-    }
-    Some(transaction)
-}
+    let details = if let Some(d) = jetton_transfer_details(&trace.actions) {
+        d
+    } else {
+        let mut d = simple_transfer_details(&root)?;
+        apply_jetton_swap(&mut d, &trace.actions);
+        d
+    };
 
-fn jetton_transfer(actions: &[TraceAction]) -> Option<JettonTransferDetails> {
-    let action = actions
-        .iter()
-        .find(|action| action.action_type.as_deref() == Some(TRACE_ACTION_JETTON_TRANSFER) && action.success == Some(true))?;
-    serde_json::from_value(action.details.clone()?).ok()
+    build_transaction(&root, state, details)
 }
 
-fn map_jetton_transfer_transaction(transaction: &TransactionMessage, state: Option<TransactionState>, details: JettonTransferDetails) -> Option<Transaction> {
+fn build_transaction(message: &TransactionMessage, state: Option<TransactionState>, details: TransferDetails) -> Option<Transaction> {
     let fee_asset_id = Chain::Ton.as_asset_id();
-    let state = state.unwrap_or_else(|| map_transaction_state(transaction));
-    let created_at = DateTime::from_timestamp(transaction.now, 0)?;
-    let hash = base64_hash_to_hex(&transaction.hash)?;
-    let token_id = hex_to_base64_address(&details.asset)?;
-    let asset_id = AssetId::from_token(Chain::Ton, &token_id);
-    let from = parse_address(&details.sender)?;
-    let to = parse_address(&details.receiver)?;
-    let memo = details.comment.filter(|comment| !comment.is_empty());
+    let state = state.unwrap_or_else(|| map_transaction_state(message));
+    let created_at = DateTime::from_timestamp(message.now, 0)?;
+    let hash = base64_hash_to_hex(&message.hash)?;
 
     Some(Transaction::new(
         hash,
-        asset_id,
-        from,
-        to,
+        details.asset_id,
+        details.from,
+        details.to,
         None,
-        TransactionType::Transfer,
+        details.transaction_type,
         state,
-        transaction.total_fees.to_string(),
+        message.total_fees.to_string(),
         fee_asset_id,
-        details.amount,
-        memo,
-        None,
+        details.value,
+        details.memo,
+        details.metadata,
         created_at,
     ))
 }
 
-fn jetton_swap(actions: &[TraceAction]) -> Option<(String, TransactionSwapMetadata)> {
-    let action = actions
+fn find_action<'a>(actions: &'a [TraceAction], action_type: &str) -> Option<&'a TraceAction> {
+    actions
         .iter()
-        .find(|action| action.action_type.as_deref() == Some(TRACE_ACTION_JETTON_SWAP) && action.success == Some(true))?;
-    let details: JettonSwapDetails = serde_json::from_value(action.details.clone()?).ok()?;
-    let sender = parse_address(&details.sender)?;
+        .find(|action| action.action_type.as_deref() == Some(action_type) && action.success == Some(true))
+}
+
+fn jetton_transfer_details(actions: &[TraceAction]) -> Option<TransferDetails> {
+    let details: JettonTransferDetails = serde_json::from_value(find_action(actions, TRACE_ACTION_JETTON_TRANSFER)?.details.clone()?).ok()?;
+    let token_id = hex_to_base64_address(&details.asset)?;
+    Some(TransferDetails {
+        asset_id: AssetId::from_token(Chain::Ton, &token_id),
+        from: parse_address(&details.sender)?,
+        to: parse_address(&details.receiver)?,
+        value: details.amount,
+        transaction_type: TransactionType::Transfer,
+        memo: details.comment.filter(|comment| !comment.is_empty()),
+        metadata: None,
+    })
+}
+
+fn apply_jetton_swap(details: &mut TransferDetails, actions: &[TraceAction]) {
+    let Some(action) = find_action(actions, TRACE_ACTION_JETTON_SWAP) else {
+        return;
+    };
+    let Some(swap): Option<JettonSwapDetails> = action.details.clone().and_then(|value| serde_json::from_value(value).ok()) else {
+        return;
+    };
+    let Some(sender) = parse_address(&swap.sender) else {
+        return;
+    };
+    let (Some(from_asset), Some(to_asset)) = (ton_asset_id(swap.asset_in.as_deref()), ton_asset_id(swap.asset_out.as_deref())) else {
+        return;
+    };
     let metadata = TransactionSwapMetadata {
-        from_asset: ton_asset_id(details.asset_in.as_deref())?,
-        from_value: details.dex_incoming_transfer.amount,
-        to_asset: ton_asset_id(details.asset_out.as_deref())?,
-        to_value: details.dex_outgoing_transfer.amount,
-        provider: details.dex,
+        from_asset,
+        from_value: swap.dex_incoming_transfer.amount,
+        to_asset,
+        to_value: swap.dex_outgoing_transfer.amount,
+        provider: swap.dex,
     };
-    Some((sender, metadata))
+    let Ok(metadata) = serde_json::to_value(metadata) else {
+        return;
+    };
+    details.transaction_type = TransactionType::Swap;
+    details.from = sender.clone();
+    details.to = sender;
+    details.metadata = Some(metadata);
 }
 
 fn ton_asset_id(raw_address: Option<&str>) -> Option<AssetId> {
@@ -141,63 +165,38 @@ fn ton_asset_id(raw_address: Option<&str>) -> Option<AssetId> {
     }
 }
 
-fn map_transaction_message_with_state(transaction: TransactionMessage, state: Option<TransactionState>) -> Option<Transaction> {
+fn simple_transfer_details(message: &TransactionMessage) -> Option<TransferDetails> {
     let asset_id = Chain::Ton.as_asset_id();
-    let state = state.unwrap_or_else(|| map_transaction_state(&transaction));
-    let created_at = DateTime::from_timestamp(transaction.now, 0)?;
-    let hash = base64_hash_to_hex(&transaction.hash)?;
-
-    if transaction.out_msgs.len() == 1 && is_simple_transfer(transaction.out_msgs.first()?) {
-        let out_message = transaction.out_msgs.first()?;
-        let from = parse_address(&out_message.source)?;
-        let to = match &out_message.destination {
-            Some(destination) => parse_address(destination)?,
-            None => return None,
-        };
-        let value = out_message.value.as_ref().unwrap_or(&"0".to_string()).clone();
-        let memo = extract_memo(out_message);
-
-        return Some(Transaction::new(
-            hash,
-            asset_id.clone(),
-            from,
-            to,
-            None,
-            TransactionType::Transfer,
-            state,
-            transaction.total_fees.to_string(),
+
+    if message.out_msgs.len() == 1 && is_simple_transfer(message.out_msgs.first()?) {
+        let out_message = message.out_msgs.first()?;
+        let to = parse_address(out_message.destination.as_deref()?)?;
+        return Some(TransferDetails {
             asset_id,
-            value,
-            memo,
-            None,
-            created_at,
-        ));
+            from: parse_address(&out_message.source)?,
+            to,
+            value: out_message.value.clone().unwrap_or_else(|| "0".to_string()),
+            transaction_type: TransactionType::Transfer,
+            memo: extract_memo(out_message),
+            metadata: None,
+        });
     }
 
-    if transaction.out_msgs.is_empty()
-        && let Some(in_msg) = &transaction.in_msg
+    if message.out_msgs.is_empty()
+        && let Some(in_msg) = &message.in_msg
         && let (Some(value), Some(source)) = (&in_msg.value, &in_msg.source)
         && let Ok(value_int) = value.parse::<i64>()
         && value_int > 0
     {
-        let from = parse_address(source)?;
-        let to = parse_address(&in_msg.destination)?;
-
-        return Some(Transaction::new(
-            hash,
-            asset_id.clone(),
-            from,
-            to,
-            None,
-            TransactionType::Transfer,
-            state,
-            transaction.total_fees.to_string(),
+        return Some(TransferDetails {
             asset_id,
-            value.clone(),
-            None,
-            None,
-            created_at,
-        ));
+            from: parse_address(source)?,
+            to: parse_address(&in_msg.destination)?,
+            value: value.clone(),
+            transaction_type: TransactionType::Transfer,
+            memo: None,
+            metadata: None,
+        });
     }
 
     None

crates/primitives/src/transaction.rs

@@ -1,6 +1,6 @@
 use crate::{
-    AddressName, AssetAddress, Chain, TransactionId, TransactionSwapMetadata, asset_id::AssetId, transaction_direction::TransactionDirection, transaction_state::TransactionState,
-    transaction_type::TransactionType, transaction_utxo::TransactionUtxoInput,
+    AddressName, AssetAddress, Chain, NFTAssetId, TransactionId, TransactionNFTTransferMetadata, TransactionSwapMetadata, asset_id::AssetId,
+    transaction_direction::TransactionDirection, transaction_state::TransactionState, transaction_type::TransactionType, transaction_utxo::TransactionUtxoInput,
 };
 
 use chrono::{DateTime, Utc};
@@ -254,6 +254,16 @@ impl Transaction {
         self.metadata.as_ref().and_then(|value| TransactionSwapMetadata::deserialize(value).ok())
     }
 
+    pub fn nft_asset_id(&self) -> Option<NFTAssetId> {
+        if self.transaction_type != TransactionType::TransferNFT {
+            return None;
+        }
+        self.metadata
+            .as_ref()
+            .and_then(|value| TransactionNFTTransferMetadata::deserialize(value).ok())
+            .map(|metadata| metadata.asset_id)
+    }
+
     pub fn asset_ids(&self) -> Vec<AssetId> {
         match self.transaction_type {
             TransactionType::Transfer

crates/storage/src/lib.rs

@@ -19,7 +19,7 @@ pub use self::database::{
     assets::{AssetFilter, AssetUpdate},
     charts::ChartFilter,
     fiat::FiatAssetFilter,
-    nft::NftCollectionFilter,
+    nft::{NftAssetFilter, NftCollectionFilter},
     perpetuals::PerpetualFilter,
     prices::{AssetsWithPricesFilter, PriceUpdate},
     referrals::{AbusePatterns, ReferralUpdate},

crates/streamer/src/steam_producer_queue.rs

@@ -11,6 +11,7 @@ use crate::{
 pub trait StreamProducerQueue {
     async fn publish_fetch_assets(&self, asset_ids: Vec<AssetId>) -> Result<bool, Box<dyn Error + Send + Sync>>;
     async fn publish_fetch_nft_asset(&self, asset_id: NFTAssetId) -> Result<bool, Box<dyn Error + Send + Sync>>;
+    async fn publish_fetch_nft_assets(&self, asset_ids: Vec<NFTAssetId>) -> Result<bool, Box<dyn Error + Send + Sync>>;
     async fn publish_fetch_prices(&self, payload: FetchPricesPayload) -> Result<bool, Box<dyn Error + Send + Sync>>;
     async fn publish_fetch_prices_assets(&self, asset_ids: Vec<AssetId>) -> Result<bool, Box<dyn Error + Send + Sync>>;
     async fn publish_transactions(&self, payload: TransactionsPayload) -> Result<bool, Box<dyn Error + Send + Sync>>;
@@ -44,6 +45,13 @@ impl StreamProducerQueue for StreamProducer {
         self.publish(QueueName::FetchNFTCollectionAssets, &FetchNFTAssetPayload::new(asset_id)).await
     }
 
+    async fn publish_fetch_nft_assets(&self, asset_ids: Vec<NFTAssetId>) -> Result<bool, Box<dyn Error + Send + Sync>> {
+        for asset_id in asset_ids {
+            self.publish_fetch_nft_asset(asset_id).await?;
+        }
+        Ok(true)
+    }
+
     async fn publish_fetch_prices(&self, payload: FetchPricesPayload) -> Result<bool, Box<dyn Error + Send + Sync>> {
         self.publish(QueueName::FetchPrices, &payload).await
     }