spark_rust/wallet/internal_handlers/implementations/

transfer.rs

use crate::common_types::types::frost::FrostSigningCommitments;
use crate::common_types::types::AbsoluteLockTime;
use crate::common_types::types::OutPoint;
use crate::common_types::types::PublicKey;
use crate::common_types::types::ScriptBuf;
use crate::common_types::types::Secp256k1;
use crate::common_types::types::Secp256k1Message;
use crate::common_types::types::SecretKey;
use crate::common_types::types::TransactionVersion;
use crate::common_types::types::TxIn;
use crate::common_types::types::Uuid;
use crate::common_types::types::Witness;
use crate::error::{validation::ValidationError, SparkSdkError, TransactionError};
use crate::signer::traits::SparkSigner;
use crate::wallet::handlers::transfer::Transfer;
use crate::wallet::internal_handlers::traits::transfer::LeafKeyTweak;
use crate::wallet::internal_handlers::traits::transfer::LeafRefundSigningData;
use crate::wallet::internal_handlers::traits::transfer::TransferInternalHandlers;
use crate::wallet::leaf_manager::SparkLeaf;
use crate::wallet::utils::bitcoin::bitcoin_tx_from_bytes;
use crate::wallet::utils::bitcoin::serialize_bitcoin_transaction;
use crate::wallet::utils::sequence::next_sequence;
use crate::SparkSdk;
use bitcoin::Sequence;
use spark_cryptography::secret_sharing::VerifiableSecretShare;
use spark_protos::common::SignatureIntent;
use spark_protos::common::SigningCommitment as ProtoSigningCommitment;
use spark_protos::spark::CancelSendTransferRequest;
use spark_protos::spark::CancelSendTransferResponse;
use spark_protos::spark::ClaimLeafKeyTweak;
use spark_protos::spark::ClaimTransferTweakKeysRequest;
use spark_protos::spark::CompleteSendTransferRequest;
use spark_protos::spark::FinalizeNodeSignaturesRequest;
use spark_protos::spark::LeafRefundTxSigningJob;
use spark_protos::spark::QueryAllTransfersRequest;
use spark_protos::spark::QueryAllTransfersResponse;
use spark_protos::spark::SendLeafKeyTweak;
use spark_protos::spark::SigningJob;
use spark_protos::spark::StartSendTransferRequest;
use spark_protos::spark::TreeNode;
use std::collections::HashMap;

// external crates
use sha256::digest;
use tonic::async_trait;

#[async_trait]
impl<S: SparkSigner + Send + Sync + Clone + 'static> TransferInternalHandlers<S> for SparkSdk<S> {
    /// Top-level transfer execution function, given the leaves to transfer. The function expects the leaves to be in [`LeafKeyTweak`] format.
    ///
    /// # Arguments
    ///
    /// * `leaves` - A vector of [`LeafKeyTweak`] objects representing the leaves to transfer.
    /// * `receiver_identity_pubkey` - The public key of the receiver.
    /// * `expiry_time` - The expiry time of the transfer.
    ///
    /// # Returns
    ///
    /// A [`Transfer`] object representing the transfer. This is an auto-converted protobuf object that is returned by the Spark API as the last step of the transfer for the sender.
    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn start_send_transfer(
        &self,
        leaves: &Vec<LeafKeyTweak>,
        receiver_identity_pubkey: &PublicKey,
        expiry_time: u64,
    ) -> Result<Transfer, SparkSdkError> {
        // Send the transfer, get the refund signatures, and aggregate by adding yours.
        let (transfer, refund_signatures) = self
            .send_transfer_sign_refunds(leaves, receiver_identity_pubkey, expiry_time)
            .await?;

        // Send the transfer, get the tweak keys, and send them.
        let transfer = self
            .send_transfer_tweak_key(&transfer, leaves, &refund_signatures)
            .await?;

        Ok(transfer)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn send_transfer_tweak_key(
        &self,
        transfer: &Transfer,
        leaves: &Vec<LeafKeyTweak>,
        refund_signature_map: &HashMap<String, Vec<u8>>,
    ) -> Result<Transfer, SparkSdkError> {
        // Prepare the key tweaks for the transfer.
        let key_tweak_input_map =
            match self.prepare_send_transfer_key_tweaks(transfer, leaves, refund_signature_map) {
                Ok(map) => map,
                Err(e) => {
                    // If preparation fails, cancel the transfer and propagate the error
                    if let Err(cancel_err) =
                        self.cancel_send_transfer(transfer.id.to_string()).await
                    {
                        // Log the cancellation error but return the original error
                        #[cfg(feature = "telemetry")]
                        tracing::error!(
                            "Failed to cancel transfer after key tweak preparation error: {}",
                            cancel_err
                        );
                    } else {
                        // Log the cancellation response
                        #[cfg(feature = "telemetry")]
                        tracing::error!("Cancelled transfer {} successfully", transfer.id);
                    }

                    return Err(e);
                }
            };

        let mut updated_transfer: Option<Transfer> = None;

        for operator in &self.config.spark_config.operator_pool.operators {
            let request = CompleteSendTransferRequest {
                transfer_id: transfer.id.to_string(),
                owner_identity_public_key: self.get_spark_address()?.serialize().to_vec(),
                leaves_to_send: key_tweak_input_map[&operator.frost_identifier_str()].clone(),
            };

            let complete_response: Option<Transfer> = match self
                .config
                .spark_config
                .call_with_retry(
                    request,
                    |mut client, req| {
                        Box::pin(async move { client.complete_send_transfer(req).await })
                    },
                    Some(operator.id),
                )
                .await
            {
                Ok(response) => response
                    .transfer
                    .map(|transfer| transfer.try_into())
                    .transpose()?,

                Err(status_err) => {
                    // If the request fails, cancel the transfer and propagate the error
                    if let Err(cancel_err) =
                        self.cancel_send_transfer(transfer.id.to_string()).await
                    {
                        // Log the cancellation error but return the original error
                        #[cfg(feature = "telemetry")]
                        tracing::error!(
                            "Failed to cancel transfer after complete_send_transfer error: {}",
                            cancel_err
                        );
                    }
                    return Err(status_err);
                }
            };

            match &updated_transfer {
                None => updated_transfer = complete_response,
                Some(existing) => {
                    // This should never happen. In this case, it should be too late to cancel the transfer.
                    if existing != &complete_response.unwrap() {
                        return Err(SparkSdkError::from(TransactionError::Transfer {
                            reason: "Inconsistent transfer responses from operators".into(),
                        }));
                    }
                }
            }
        }

        Ok(updated_transfer.unwrap())
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn send_transfer_sign_refunds(
        &self,
        leaves: &Vec<LeafKeyTweak>,
        receiver_identity_pubkey: &PublicKey,
        expiry_time: u64,
    ) -> Result<(Transfer, HashMap<String, Vec<u8>>), SparkSdkError> {
        // Generate a new transfer ID.
        let transfer_id = Uuid::now_v7().to_string();

        // Prepare signing data for each leaf
        let mut leaf_data_map = HashMap::new();
        for leaf_key in leaves {
            let commitments = self.signer.generate_frost_signing_commitments()?;
            let node_tx = bitcoin_tx_from_bytes(&leaf_key.leaf.node_tx)?;

            let secp = Secp256k1::new();
            let signing_public_key = leaf_key.old_signing_private_key.public_key(&secp);

            let leaf_refund_signing_data = LeafRefundSigningData {
                signing_public_key,
                receiving_pubkey: *receiver_identity_pubkey,
                commitment: frost_commitment_to_proto_commitment(&commitments)?,
                tx: node_tx,
                refund_tx: None,
                vout: leaf_key.leaf.vout,
            };
            leaf_data_map.insert(leaf_key.leaf.id.clone(), leaf_refund_signing_data);
        }

        // Create the signing jobs needed for the transfer.
        let signing_jobs = self.prepare_refund_so_signing_jobs(leaves, &mut leaf_data_map)?;

        // Send the first transfer request as the sender.
        let request_data = StartSendTransferRequest {
            transfer_id: transfer_id.clone(),
            leaves_to_send: signing_jobs,
            owner_identity_public_key: self.get_spark_address()?.serialize().to_vec(),
            receiver_identity_public_key: receiver_identity_pubkey.serialize().to_vec(),
            expiry_time: Some(prost_types::Timestamp {
                seconds: expiry_time as i64,
                nanos: 0,
            }),
            key_tweak_proofs: Default::default(),
        };

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

        let transfer: Transfer = response.transfer.unwrap().try_into()?;

        // Sign the refunds and aggregate signatures combining your and Spark Operators' signatures. This will give you the signature for the transfer transaction to be sent to the receiver.
        let signing_results = response.signing_results;
        let signatures =
            match self
                .signer
                .sign_transfer_refunds(&leaf_data_map, &signing_results, vec![])
            {
                Ok(sigs) => sigs,
                Err(e) => {
                    // If signing fails, cancel the transfer and propagate the error
                    if let Err(cancel_err) =
                        self.cancel_send_transfer(transfer.id.to_string()).await
                    {
                        // Log the cancellation error but return the original error
                        #[cfg(feature = "telemetry")]
                        tracing::error!(
                            "Failed to cancel transfer after signing error: {}",
                            cancel_err
                        );
                    }
                    return Err(e);
                }
            };

        // Create a map of node IDs to signatures.
        let mut signature_map = HashMap::new();
        for leaf_signature in signatures {
            signature_map.insert(leaf_signature.node_id, leaf_signature.refund_tx_signature);
        }

        // `transfer` is always Some.
        Ok((transfer, signature_map))
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    fn prepare_send_transfer_key_tweaks(
        &self,
        transfer: &Transfer,
        leaves: &Vec<LeafKeyTweak>,
        refund_signature_map: &HashMap<String, Vec<u8>>,
    ) -> Result<HashMap<String, Vec<SendLeafKeyTweak>>, SparkSdkError> {
        let mut leaves_tweaks_map = HashMap::new();
        for leaf in leaves {
            // Get the refund signature
            let leaf_refund_signature = refund_signature_map.get(&leaf.leaf.id).cloned();

            // Get the tweaks for this leaf by preparing the key tweak data
            let leaf_tweaks = self.prepare_single_send_transfer_key_tweak(
                &transfer.id.to_string(),
                leaf,
                &transfer.receiver_identity_public_key,
                leaf_refund_signature,
            )?;

            // Add the tweaks to the map, grouped by identifier
            for (identifier, leaf_tweak) in leaf_tweaks {
                leaves_tweaks_map
                    .entry(identifier)
                    .or_insert_with(Vec::new)
                    .push(leaf_tweak);
            }
        }

        Ok(leaves_tweaks_map)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    fn prepare_single_send_transfer_key_tweak(
        &self,
        transfer_id: &str,
        leaf: &LeafKeyTweak,
        receiver_pubkey: &PublicKey,
        refund_signature: Option<Vec<u8>>,
    ) -> Result<HashMap<String, SendLeafKeyTweak>, SparkSdkError> {
        let refund_signature = refund_signature.unwrap_or_default();

        let secp = Secp256k1::new();
        let publickey = &leaf.old_signing_private_key.public_key(&secp);
        let tweaked_public_key = self.signer.subtract_secret_keys_given_pubkeys(
            publickey,
            &leaf.new_signing_public_key,
            true,
        )?;

        // Split the secret key that belongs to the tweaked public key.
        let shares = self
            .signer
            .split_from_public_key_with_verifiable_secret_sharing(
                &tweaked_public_key,
                self.config.spark_config.threshold as usize,
                self.config.spark_config.operator_pool.operators.len(),
            )?;

        let mut pubkey_shares_tweak = HashMap::new();
        for operator in &self.config.spark_config.operator_pool.operators {
            let share = find_share(&shares, operator.id.into())?;

            let share_scalar = SecretKey::from_slice(&share.secret_share.share.to_bytes())?;
            let pubkey_tweak = PublicKey::from_secret_key(&Secp256k1::new(), &share_scalar);
            pubkey_shares_tweak.insert(
                operator.frost_identifier_str(),
                pubkey_tweak.serialize().to_vec(),
            );
        }

        // Generate signature over payload
        // First, let's encrypt the new signing private key using ECIES
        let secret_cipher = self
            .signer
            .encrypt_secret_key_with_ecies(receiver_pubkey, &leaf.new_signing_public_key)?;

        // Now we'll create the payload exactly as in the Go code by concatenating:
        // 1. leaf ID
        // 2. transfer ID
        // 3. encrypted secret (secret_cipher)
        let payload = [
            leaf.leaf.id.as_bytes(), // leaf ID bytes
            transfer_id.as_bytes(),  // transfer ID bytes
            &secret_cipher,          // encrypted secret bytes
        ]
        .concat();

        // Sign the hash using ECDSA with our identity private key
        // Note: We use the raw hash bytes directly, no hex encoding needed
        let network = self.config.spark_config.network.to_bitcoin_network();
        let signature = self
            .signer
            .sign_message_ecdsa_with_identity_key(payload, true, network)?;
        // let signature = signature.serialize_der().to_vec();

        let mut leaf_tweaks_map = HashMap::new();
        for operator in &self.config.spark_config.operator_pool.operators {
            let share = find_share(&shares, operator.id.into())?;
            leaf_tweaks_map.insert(
                operator.frost_identifier_str(),
                SendLeafKeyTweak {
                    leaf_id: leaf.leaf.id.clone(),
                    secret_share_tweak: Some(spark_protos::spark::SecretShare {
                        secret_share: share.secret_share.share.to_bytes().to_vec(),
                        proofs: share.proofs.clone(),
                    }),
                    pubkey_shares_tweak: pubkey_shares_tweak.clone(),
                    secret_cipher: secret_cipher.clone(),
                    signature: signature.serialize_der().to_vec(),
                    refund_signature: refund_signature.to_vec(),
                },
            );
        }

        Ok(leaf_tweaks_map)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    fn prepare_refund_so_signing_jobs(
        &self,
        leaves: &Vec<LeafKeyTweak>,
        leaf_data_map: &mut HashMap<String, LeafRefundSigningData>,
    ) -> Result<Vec<LeafRefundTxSigningJob>, SparkSdkError> {
        let mut signing_jobs = Vec::new();

        for leaf in leaves {
            let refund_signing_data = leaf_data_map.get_mut(&leaf.leaf.id).ok_or_else(|| {
                SparkSdkError::from(ValidationError::InvalidInput {
                    field: "Leaf data not found".to_string(),
                })
            })?;

            let signing_pubkey = &refund_signing_data.signing_public_key;
            let refund_tx =
                self.create_refund_tx(&leaf.leaf, &refund_signing_data.receiving_pubkey)?;
            let refund_bytes = serialize_bitcoin_transaction(&refund_tx)?;

            refund_signing_data.refund_tx = Some(refund_tx);

            let refund_commitment_proto = refund_signing_data.commitment.clone();

            signing_jobs.push(LeafRefundTxSigningJob {
                leaf_id: leaf.leaf.id.clone(),
                refund_tx_signing_job: Some(SigningJob {
                    signing_public_key: signing_pubkey.serialize().to_vec(),
                    raw_tx: refund_bytes,
                    signing_nonce_commitment: Some(refund_commitment_proto),
                }),
            });
        }

        Ok(signing_jobs)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    fn create_refund_tx(
        &self,
        leaf: &TreeNode,
        receiving_pubkey: &PublicKey,
    ) -> Result<bitcoin::Transaction, SparkSdkError> {
        let node_tx = bitcoin_tx_from_bytes(&leaf.node_tx)?;
        let refund_tx = bitcoin_tx_from_bytes(&leaf.refund_tx)?;

        let mut new_refund_tx = bitcoin::Transaction {
            version: TransactionVersion::TWO,
            lock_time: AbsoluteLockTime::ZERO,
            input: vec![],
            output: vec![],
        };

        let old_sequence = refund_tx.input[0].sequence.0;
        let sequence = Sequence(next_sequence(old_sequence));

        new_refund_tx.input.push(TxIn {
            previous_output: OutPoint {
                txid: node_tx.compute_txid(),
                vout: 0,
            },
            script_sig: ScriptBuf::default(),
            sequence,
            witness: Witness::default(),
        });

        let secp = Secp256k1::new();
        let addr = bitcoin::Address::p2tr(
            &secp,
            receiving_pubkey.x_only_public_key().0,
            None,
            self.config.spark_config.network.to_bitcoin_network(),
        );

        new_refund_tx.output.push(bitcoin::TxOut {
            value: node_tx.output[0].value,
            script_pubkey: addr.script_pubkey(),
        });

        Ok(new_refund_tx)
    }

    // Helper methods
    // fn compare_transfers(&self, t1: &Transfer, t2: &Transfer) -> bool {
    //     t1.id == t2.id
    //         && t1.receiver_identity_public_key == t2.receiver_identity_public_key
    //         && t1.status == t2.status
    //         && t1.total_value == t2.total_value
    //         && t1.expiry_time.as_ref().map(|t| t.seconds)
    //             == t2.expiry_time.as_ref().map(|t| t.seconds)
    //         && t1.leaves.len() == t2.leaves.len()
    // }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn claim_finalize_incoming_transfer(
        &self,
        transfer: &Transfer,
        leaves: &Vec<LeafKeyTweak>,
    ) -> Result<(), SparkSdkError> {
        // First tweak the keys
        self.claim_transfer_tweak_keys(transfer, leaves).await?;

        // Then sign the refunds
        let signatures = self.claim_transfer_sign_refunds(transfer, leaves).await?;

        // Finally, finalize the transfer
        self.finalize_transfer(&signatures).await?;

        let mut leaf_nodes = vec![];
        for (leaf, _) in leaves.iter().zip(signatures.iter()) {
            // Create a new refund transaction with decremented timelock
            let new_refund_tx = self.create_refund_tx(&leaf.leaf, &leaf.new_signing_public_key)?;
            let new_refund_tx_bytes = serialize_bitcoin_transaction(&new_refund_tx)?;

            let mut updated_leaf = leaf.leaf.clone();
            updated_leaf.refund_tx = new_refund_tx_bytes;
            leaf_nodes.push(SparkLeaf::Bitcoin(updated_leaf));
        }

        self.leaf_manager.insert_leaves(leaf_nodes, false)?;

        Ok(())
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn claim_transfer_tweak_keys(
        &self,
        transfer: &Transfer,
        leaves: &Vec<LeafKeyTweak>,
    ) -> Result<(), SparkSdkError> {
        let leaves_tweaks_map = self.prepare_claim_leaves_key_tweaks(leaves)?;

        for operator in &self.config.spark_config.operator_pool.operators {
            let request_data = ClaimTransferTweakKeysRequest {
                transfer_id: transfer.id.to_string(),
                owner_identity_public_key: self.get_spark_address()?.serialize().to_vec(),
                leaves_to_receive: leaves_tweaks_map[&operator.frost_identifier_str()].clone(),
            };

            self.config
                .spark_config
                .call_with_retry(
                    request_data,
                    |mut client, req| {
                        Box::pin(async move { client.claim_transfer_tweak_keys(req).await })
                    },
                    Some(operator.id),
                )
                .await?;
        }

        Ok(())
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    fn prepare_claim_leaves_key_tweaks(
        &self,
        leaves: &Vec<LeafKeyTweak>,
    ) -> Result<HashMap<String, Vec<ClaimLeafKeyTweak>>, SparkSdkError> {
        let mut leaves_tweaks_map = HashMap::new();

        for leaf in leaves {
            let leaf_tweaks = self.prepare_claim_leaf_key_tweaks(leaf)?;

            for (identifier, leaf_tweak) in leaf_tweaks {
                leaves_tweaks_map
                    .entry(identifier)
                    .or_insert_with(Vec::new)
                    .push(leaf_tweak);
            }
        }

        Ok(leaves_tweaks_map)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    fn prepare_claim_leaf_key_tweaks(
        &self,
        leaf: &LeafKeyTweak,
    ) -> Result<HashMap<String, ClaimLeafKeyTweak>, SparkSdkError> {
        let secp = Secp256k1::new();
        let tweaked_public_key = self.signer.subtract_secret_keys_given_pubkeys(
            &leaf.old_signing_private_key.public_key(&secp),
            &leaf.new_signing_public_key,
            true,
        )?;

        let shares = self
            .signer
            .split_from_public_key_with_verifiable_secret_sharing(
                &tweaked_public_key,
                // &minus_one.to_be_bytes().to_vec(),
                self.config.spark_config.threshold as usize,
                self.config.spark_config.operator_pool.operators.len(),
            )
            .unwrap();

        let mut pubkey_shares_tweak = HashMap::new();
        for operator in &self.config.spark_config.operator_pool.operators {
            let share = find_share(&shares, operator.id.into())?;

            // This part comes from Spark cryptography, so the secret key is in the code.
            let share_scalar = SecretKey::from_slice(&share.secret_share.share.to_bytes())?;
            let pubkey_tweak = PublicKey::from_secret_key(&Secp256k1::new(), &share_scalar);
            pubkey_shares_tweak.insert(
                operator.frost_identifier_str(),
                pubkey_tweak.serialize().to_vec(),
            );
        }

        let mut leaf_tweaks_map = HashMap::new();
        for operator in &self.config.spark_config.operator_pool.operators {
            let share = find_share(&shares, operator.id.into())?;

            leaf_tweaks_map.insert(
                operator.frost_identifier_str(),
                ClaimLeafKeyTweak {
                    leaf_id: leaf.leaf.id.clone(),
                    secret_share_tweak: Some(spark_protos::spark::SecretShare {
                        secret_share: share.secret_share.share.to_bytes().to_vec(),
                        proofs: share.proofs.clone(),
                    }),
                    pubkey_shares_tweak: pubkey_shares_tweak.clone(),
                },
            );
        }

        Ok(leaf_tweaks_map)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn claim_transfer_sign_refunds(
        &self,
        transfer: &Transfer,
        leaf_keys: &Vec<LeafKeyTweak>,
    ) -> Result<Vec<spark_protos::spark::NodeSignatures>, SparkSdkError> {
        // Create a map to store refund signing data for each leaf
        let mut leaf_data_map = HashMap::new();

        for leaf_key in leaf_keys {
            // For each leaf key, we create the signing data using the new private key
            let commitments = self.signer.generate_frost_signing_commitments()?;

            // Deserialize the transaction from raw bytes
            let tx = match bitcoin::consensus::deserialize(&leaf_key.leaf.node_tx) {
                Ok(tx) => tx,
                Err(e) => {
                    return Err(SparkSdkError::from(ValidationError::InvalidInput {
                        field: format!("Failed to deserialize transaction: {}", e),
                    }));
                }
            };

            // Store all the necessary data for signing
            leaf_data_map.insert(
                leaf_key.leaf.id.clone(),
                LeafRefundSigningData {
                    signing_public_key: leaf_key.new_signing_public_key,
                    receiving_pubkey: leaf_key.new_signing_public_key,
                    commitment: frost_commitment_to_proto_commitment(&commitments)?,
                    tx,
                    refund_tx: None,
                    vout: leaf_key.leaf.vout,
                },
            );
        }

        // Prepare the signing jobs for each leaf
        let signing_jobs = self.prepare_refund_so_signing_jobs(leaf_keys, &mut leaf_data_map)?;

        // Request signing of refunds
        let request = spark_protos::spark::ClaimTransferSignRefundsRequest {
            transfer_id: transfer.id.to_string(),
            owner_identity_public_key: self.get_spark_address()?.serialize().to_vec(),
            signing_jobs,
            key_tweak_proofs: Default::default(),
        };

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

        // Process the signing results and generate final signatures
        self.signer
            .sign_transfer_refunds(&leaf_data_map, &response.signing_results, vec![])
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn finalize_transfer(
        &self,
        signatures: &[spark_protos::spark::NodeSignatures],
    ) -> Result<(), SparkSdkError> {
        let request = FinalizeNodeSignaturesRequest {
            intent: SignatureIntent::Transfer as i32,
            node_signatures: signatures.to_vec(),
        };

        self.config
            .spark_config
            .call_with_retry(
                request,
                |mut client, req| {
                    Box::pin(async move { client.finalize_node_signatures(req).await })
                },
                None,
            )
            .await?;

        Ok(())
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn verify_pending_transfer(
        &self,
        transfer: &Transfer,
    ) -> Result<HashMap<String, SecretKey>, SparkSdkError> {
        // Create the map for leafID -> decrypted leaf private key
        let mut leaf_privkey_map = HashMap::new();

        // 1) Parse the sender's public key
        let secp = Secp256k1::new();

        // 2) For each leaf, verify the signature, then decrypt the secret
        for leaf in &transfer.leaves {
            // Create the message to verify: leaf.Leaf.Id + transfer.Id + leaf.SecretCipher
            let mut payload = leaf
                .leaf
                .as_ref()
                .map(|l| l.id.to_string())
                .unwrap_or_default()
                .into_bytes();
            payload.extend_from_slice(transfer.id.to_string().as_bytes());
            payload.extend_from_slice(&leaf.secret_cipher);

            // Hash the payload
            let payload_hash_hex = digest(&payload);
            let payload_hash_bytes = hex::decode(payload_hash_hex).map_err(|e| {
                SparkSdkError::from(ValidationError::InvalidInput {
                    field: format!("Failed to decode hex payload hash: {e}"),
                })
            })?;

            // Build secp256k1 message and verify ECDSA signature
            let msg = Secp256k1Message::from_digest_slice(&payload_hash_bytes).map_err(|e| {
                SparkSdkError::from(ValidationError::InvalidInput {
                    field: format!("Failed to create message for signature verify: {e}"),
                })
            })?;

            let signature =
                leaf.signature
                    .ok_or(SparkSdkError::from(ValidationError::InvalidInput {
                        field: "Missing signature".to_string(),
                    }))?;

            secp.verify_ecdsa(&msg, &signature, &transfer.sender_identity_public_key)
                .map_err(|e| {
                    SparkSdkError::from(ValidationError::InvalidInput {
                        field: format!("Failed to verify signature: {e}"),
                    })
                })?;

            // Decrypt secret cipher with our identity key (assuming the SparkSigner can do ECIES decryption)
            let network = self.config.spark_config.network.to_bitcoin_network();
            let leaf_secret = self
                .signer
                .decrypt_secret_key_with_ecies(&leaf.secret_cipher, network)?;

            // Record the decrypted leaf secret (private key) in the map
            if let Some(leaf_node) = &leaf.leaf {
                leaf_privkey_map.insert(leaf_node.id.to_string().clone(), leaf_secret);
            }
        }

        Ok(leaf_privkey_map)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn query_all_transfers(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<QueryAllTransfersResponse, SparkSdkError> {
        let identity_public_key = self.get_spark_address()?;
        let request = QueryAllTransfersRequest {
            limit: limit as i64,
            offset: offset as i64,
            identity_public_key: identity_public_key.serialize().to_vec(),
        };

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

        Ok(response)
    }

    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
    async fn cancel_send_transfer(
        &self,
        transfer_id: String,
    ) -> Result<Option<Transfer>, SparkSdkError> {
        let request_data = CancelSendTransferRequest {
            transfer_id: transfer_id.to_string(),
            sender_identity_public_key: self.get_spark_address()?.serialize().to_vec(),
        };

        let futures = self
            .config
            .spark_config
            .operator_pool
            .operators
            .iter()
            .map(|operator| {
                let operator_id = operator.id;
                let request_clone = request_data.clone();

                async move {
                    let response = self
                        .config
                        .spark_config
                        .call_with_retry(
                            request_clone,
                            |mut client, req| {
                                Box::pin(async move { client.cancel_send_transfer(req).await })
                            },
                            Some(operator_id),
                        )
                        .await?;

                    Ok(response)
                }
            });

        let results: Vec<Result<CancelSendTransferResponse, SparkSdkError>> =
            futures::future::join_all(futures).await;

        // Return the first successful response, if any
        if let Some(result) = results.into_iter().flatten().next() {
            return result.transfer.map(|tr| tr.try_into()).transpose();
        }

        Ok(None)
    }
}

fn find_share(
    shares: &[VerifiableSecretShare],
    operator_id: u64,
) -> Result<VerifiableSecretShare, SparkSdkError> {
    let target_index = k256::Scalar::from(operator_id + 1);
    shares
        .iter()
        .find(|s| s.secret_share.index == target_index)
        .cloned()
        .ok_or_else(|| {
            SparkSdkError::from(ValidationError::InvalidInput {
                field: "Share not found".to_string(),
            })
        })
}

fn frost_commitment_to_proto_commitment(
    commitments: &FrostSigningCommitments,
) -> Result<ProtoSigningCommitment, SparkSdkError> {
    let hiding = commitments.hiding().serialize().unwrap();
    let binding = commitments.binding().serialize().unwrap();

    Ok(ProtoSigningCommitment { hiding, binding })
}