spark_rust/wallet/handlers/

transfer.rs

use crate::with_handler_lock;
use crate::{
    constants::spark::DEFAULT_TRANSFER_EXPIRY,
    error::{SparkSdkError, ValidationError},
    signer::traits::{derivation_path::SparkKeyType, SparkSigner},
    wallet::{
        internal_handlers::traits::leaves::LeavesInternalHandlers,
        internal_handlers::traits::transfer::{LeafKeyTweak, TransferInternalHandlers},
        leaf_manager::SparkNodeStatus,
    },
    SparkSdk,
};
use bitcoin::secp256k1::PublicKey;
use spark_protos::spark::{
    query_pending_transfers_request::Participant, QueryAllTransfersResponse,
    QueryPendingTransfersRequest, Transfer, TransferStatus,
};

impl<S: SparkSigner + Send + Sync + Clone + 'static> SparkSdk<S> {
    /// Queries all pending transfers where the current user is the receiver.
    ///
    /// This function retrieves all pending transfers that are waiting to be accepted by the current user.
    /// A pending transfer represents funds that have been sent to the user but have not yet been claimed.
    /// The transfers remain in a pending state until the receiver claims them, at which point the funds
    /// become available in their wallet.
    ///
    /// # Returns
    ///
    /// * `Ok(Vec<Transfer>)` - A vector of pending [`Transfer`] objects if successful
    /// * `Err(SparkSdkError)` - If there was an error querying the transfers
    ///
    /// # Example
    ///
    /// ```
    /// # use spark_rust::SparkSdk;
    /// # async fn example(sdk: SparkSdk) -> Result<(), Box<dyn std::error::Error>> {
    /// let pending = sdk.query_pending_transfers().await?;
    /// for transfer in pending {
    ///     println!("Pending transfer: {:?} satoshis", transfer.total_value);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub async fn query_pending_transfers(&self) -> Result<Vec<Transfer>, SparkSdkError> {
        with_handler_lock!(self, async {
            self.query_pending_transfers_internal().await
        })
        .await
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub(crate) async fn query_pending_transfers_internal(
        &self,
    ) -> Result<Vec<Transfer>, SparkSdkError> {
        let request_data = QueryPendingTransfersRequest {
            transfer_ids: vec![],
            participant: Some(Participant::ReceiverIdentityPublicKey(
                self.get_spark_address()?.serialize().to_vec(),
            )),
            network: self.config.spark_config.network.marshal_proto(),
        };

        let response = self
            .config
            .spark_config
            .call_with_retry(
                request_data,
                |mut client, req| {
                    Box::pin(async move { client.query_pending_transfers(req).await })
                },
                None,
            )
            .await?;

        Ok(response.transfers)
    }

    /// Initiates a transfer of funds to another user.
    ///
    /// This function handles the process of transferring funds from the current user's wallet to another user,
    /// identified by their public key. The transfer process involves several steps:
    ///
    /// 1. Selecting appropriate leaves (UTXOs) that contain sufficient funds for the transfer
    /// 2. Locking the selected leaves to prevent concurrent usage
    /// 3. Generating new signing keys for the transfer
    /// 4. Creating and signing the transfer transaction
    /// 5. Removing the used leaves from the wallet
    ///
    /// The transfer remains in a pending state until the receiver claims it. The expiry time is set to
    /// 30 days by default (see `DEFAULT_TRANSFER_EXPIRY`).
    ///
    /// # Arguments
    ///
    /// * `amount` - The amount to transfer in satoshis. Must be greater than the dust limit and the wallet
    ///             must have a leaf with exactly this amount.
    /// * `receiver_spark_address` - The Spark address identifying the receiver of the transfer. This should
    ///                               be the receiver's Spark address, not a regular Bitcoin public key.
    ///
    /// # Returns
    ///
    /// * `Ok(String)` - The transfer ID if successful. This ID can be used to track the transfer status.
    /// * `Err(SparkSdkError)` - If the transfer fails. Common error cases include:
    ///   - No leaf with exact amount available
    ///   - Failed to lock leaves
    ///   - Failed to generate new signing keys
    ///   - Network errors when communicating with Spark operators
    ///
    /// # Example
    ///
    /// ```
    /// # use spark_rust::SparkSdk;
    /// # use bitcoin::secp256k1::PublicKey;
    /// # use std::str::FromStr;
    /// # use uuid::Uuid;
    /// # async fn example(sdk: &mut SparkSdk) -> Result<(), Box<dyn std::error::Error>> {
    /// let amount = 100_000;
    ///
    /// // Currently, a user's Spark address is their public key.
    /// let receiver_spark_address = PublicKey::from_str("02782d7ba8764306bd324e23082f785f7c880b7202cb10c85a2cb96496aedcaba7").unwrap();
    ///
    /// let transfer_id_string = sdk.transfer(amount, &receiver_spark_address).await?;
    /// let transfer_id = Uuid::parse_str(&transfer_id_string).unwrap();
    /// println!("Transfer ID is {}", transfer_id);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Notes
    ///
    /// Currently, the leaf selection algorithm only supports selecting a single leaf with the exact
    /// transfer amount. Future versions will support combining multiple leaves and handling change outputs.
    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub async fn transfer(
        &self,
        amount: u64,
        receiver_spark_address: &PublicKey,
    ) -> Result<String, SparkSdkError> {
        with_handler_lock!(self, async {
            let expiry_time = chrono::Utc::now().timestamp() as u64 + DEFAULT_TRANSFER_EXPIRY;

            self.refresh_timelock_nodes(None).await?;

            // do leaf selection
            // if not sufficient leaves are found, a swap with the SSP will be requested
            let leaf_selection_response = self.prepare_leaves_for_amount(amount).await?;
            let unlocking_id = leaf_selection_response.unlocking_id.unwrap();

            // TODO: expect that at this point, leaf_selection_response.total_value == amount, because a swap should happen between the SSP and the wallet.
            let selected_leaves = leaf_selection_response.leaves;
            let leaf_ids = selected_leaves
                .iter()
                .map(|leaf| leaf.get_id().clone())
                .collect::<Vec<String>>();

            let mut leaves_to_transfer = Vec::new();
            for leaf in selected_leaves {
                // get new
                let new_signing_public_key = self.signer.new_ephemeral_keypair()?;

                // get old
                let old_signing_private_key = self.signer.expose_leaf_secret_key_for_transfer(
                    leaf.get_id().clone(),
                    SparkKeyType::BaseSigning,
                    0,
                    self.config.spark_config.network.to_bitcoin_network(),
                )?;

                leaves_to_transfer.push(LeafKeyTweak {
                    leaf: leaf.get_tree_node()?,
                    old_signing_private_key,
                    new_signing_public_key,
                });
            }

            // TODO - when add actual leaf selection, this might be an array of length > 1.
            let transfer = self
                .start_send_transfer(&leaves_to_transfer, receiver_spark_address, expiry_time)
                .await?;

            // unlock and remove leaves from the leaf manager
            self.leaf_manager
                .unlock_leaves(unlocking_id.clone(), &leaf_ids, true)?;

            Ok(transfer.id)
        })
        .await
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub async fn transfer_leaf_ids(
        &self,
        leaf_ids: Vec<String>,
        receiver_identity_pubkey: &PublicKey,
    ) -> Result<String, SparkSdkError> {
        with_handler_lock!(self, async {
            self.transfer_leaf_ids_internal(leaf_ids, receiver_identity_pubkey)
                .await
        })
        .await
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub(crate) async fn transfer_leaf_ids_internal(
        &self,
        leaf_ids: Vec<String>,
        receiver_identity_pubkey: &PublicKey,
    ) -> Result<String, SparkSdkError> {
        let expiry_time = chrono::Utc::now().timestamp() as u64 + DEFAULT_TRANSFER_EXPIRY;

        let leaf_selection_response = self
            .leaf_manager
            .lock_leaf_ids(&leaf_ids, SparkNodeStatus::Transfer)?;

        let unlocking_id = leaf_selection_response.unlocking_id.unwrap();

        let selected_leaves = leaf_selection_response.leaves;

        let mut leaves_to_transfer = Vec::new();
        for leaf in selected_leaves {
            // get new
            let new_signing_public_key = self.signer.new_ephemeral_keypair()?;

            let network = self.config.spark_config.network.to_bitcoin_network();
            let old_signing_pubkey = self.signer.get_deposit_signing_key(network)?;
            let old_signing_private_key = self
                .signer
                .sensitive_expose_secret_key_from_pubkey(&old_signing_pubkey, false)?;

            leaves_to_transfer.push(LeafKeyTweak {
                leaf: leaf.get_tree_node()?,
                old_signing_private_key,
                new_signing_public_key,
            });
        }

        // TODO - when add actual leaf selection, this might be an array of length > 1.
        let transfer = self
            .start_send_transfer(&leaves_to_transfer, receiver_identity_pubkey, expiry_time)
            .await?;

        // unlock and remove leaves from the leaf manager
        self.leaf_manager
            .unlock_leaves(unlocking_id.clone(), &leaf_ids, true)?;

        Ok(transfer.id)
    }

    /// Claims a pending transfer that was sent to this wallet.
    ///
    /// This function processes a pending transfer and claims the funds into the wallet. It performs the following steps:
    /// 1. Verifies the transfer is in the correct state (SenderKeyTweaked)
    /// 2. Verifies and decrypts the leaf private keys using the wallet's identity key
    /// 3. Generates new signing keys for the claimed leaves
    /// 4. Finalizes the transfer by:
    ///    - Tweaking the leaf keys
    ///    - Signing refund transactions
    ///    - Submitting the signatures to the Spark network
    ///    - Storing the claimed leaves in the wallet's database
    ///
    /// # Arguments
    ///
    /// * `transfer` - The pending transfer to claim, must be in SenderKeyTweaked status
    ///
    /// # Returns
    ///
    /// * `Ok(())` - If the transfer was successfully claimed
    /// * `Err(SparkSdkError)` - If there was an error during the claim process
    ///
    /// # Errors
    ///
    /// Returns [`SparkSdkError::InvalidInput`] if:
    /// - The transfer is not in SenderKeyTweaked status
    ///
    /// May also return other `SparkSdkError` variants for network, signing or storage errors.
    ///
    /// # Example
    ///
    /// ```
    /// # use spark_rust::{SparkSdk, SparkNetwork, signer::default_signer::DefaultSigner, signer::traits::SparkSigner};
    ///
    /// async fn example() {
    ///     let mnemonic = "abandon ability able about above absent absorb abstract absurd abuse access accident";
    ///     let network = SparkNetwork::Regtest;
    ///     let signer = DefaultSigner::from_mnemonic(mnemonic, network.clone()).await.unwrap();
    ///     let sdk = SparkSdk::new(network, signer).await.unwrap();
    ///     let pending = sdk.query_pending_transfers().await.unwrap();
    ///     for transfer in pending {
    ///         sdk.claim_transfer(transfer).await.unwrap();
    ///     }
    /// }
    /// ```
    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub async fn claim_transfer(&self, transfer: Transfer) -> Result<(), SparkSdkError> {
        with_handler_lock!(self, async { self.claim_transfer_internal(transfer).await }).await
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn claim_transfer_internal(&self, transfer: Transfer) -> Result<(), SparkSdkError> {
        // Validate the request
        for transfer_leaf in &transfer.leaves {
            if transfer_leaf.leaf.is_none() {
                return Err(SparkSdkError::from(ValidationError::InvalidInput {
                    field: "Transfer leaf is not found".to_string(),
                }));
            }
        }

        if transfer.status != TransferStatus::SenderKeyTweaked as i32 {
            return Err(SparkSdkError::from(ValidationError::InvalidInput {
                field: "Transfer is not in the correct status".to_string(),
            }));
        }

        let mut leaves_to_claim = Vec::new();
        for leaf in &transfer.leaves {
            let leaf_private_key_map = self.verify_pending_transfer(&transfer).await?;

            let leaf_id = leaf.leaf.as_ref().unwrap().id.clone();
            let new_pubkey = self.signer.new_secp256k1_keypair(
                leaf_id.clone(),
                SparkKeyType::BaseSigning,
                0,
                self.config.spark_config.network.to_bitcoin_network(),
            )?;

            self.signer
                .insert_secp256k1_keypair_from_secret_key(&leaf_private_key_map[&leaf_id])
                .unwrap();

            let claim_node = LeafKeyTweak {
                leaf: leaf.leaf.as_ref().unwrap().clone(),
                old_signing_private_key: leaf_private_key_map[&leaf_id],
                new_signing_public_key: new_pubkey,
            };

            leaves_to_claim.push(claim_node);
        }

        self.claim_finalize_incoming_transfer(&transfer, &leaves_to_claim)
            .await?;

        // refresh timelock nodes
        self.refresh_timelock_nodes(None).await?;

        Ok(())
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub async fn claim_transfers(&self) -> Result<(), SparkSdkError> {
        with_handler_lock!(self, async { self.claim_transfers_internal().await }).await
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub async fn claim_transfers_internal(&self) -> Result<(), SparkSdkError> {
        let pending = self.query_pending_transfers_internal().await?;

        let pending_len = pending.len();
        let claim_futures = pending
            .into_iter()
            .map(|transfer| self.claim_transfer_internal(transfer.clone()));
        futures::future::try_join_all(claim_futures).await?;

        #[cfg(feature = "telemetry")]
        tracing::debug!("Claimed {:?} pending transfers", pending_len);

        Ok(())
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    pub async fn get_all_transfers(
        &self,
        limit: Option<u32>,
        offset: Option<u32>,
    ) -> Result<QueryAllTransfersResponse, SparkSdkError> {
        let limit = limit.unwrap_or(20);
        let offset = offset.unwrap_or(0);

        let response = self.query_all_transfers(limit, offset).await?;
        Ok(response)
    }
}