refactor: Move business logic outside of operator::deposit_sign rpc c…

…all.

core/src/operator.rs

@@ -1,16 +1,22 @@
 use crate::actor::{Actor, WinternitzDerivationPath};
+use crate::builder::sighash::create_operator_sighash_stream;
 use crate::config::BridgeConfig;
 use crate::database::Database;
 use crate::errors::BridgeError;
 use crate::extended_rpc::ExtendedRpc;
 use crate::musig2::AggregateFromPublicKeys;
 use crate::utils::ALL_BITVM_INTERMEDIATE_VARIABLES;
-use bitcoin::{Amount, OutPoint, Txid, XOnlyPublicKey};
+use crate::EVMAddress;
+use bitcoin::address::NetworkUnchecked;
+use bitcoin::secp256k1::schnorr;
+use bitcoin::{Address, Amount, OutPoint, Txid, XOnlyPublicKey};
 use bitvm::signatures::winternitz;
 use jsonrpsee::core::client::ClientT;
 use jsonrpsee::http_client::HttpClientBuilder;
 use jsonrpsee::rpc_params;
 use serde_json::json;
+use tokio::sync::mpsc;
+use tokio_stream::StreamExt;
 
 pub type SecretPreimage = [u8; 20];
 pub type PublicHash = [u8; 20]; // TODO: Make sure these are 20 bytes and maybe do this a struct?
@@ -112,6 +118,51 @@ impl Operator {
         })
     }
 
+    pub async fn deposit_sign(
+        &self,
+        deposit_outpoint: OutPoint,
+        evm_address: EVMAddress,
+        recovery_taproot_address: Address<NetworkUnchecked>,
+        user_takes_after: u16,
+    ) -> Result<mpsc::Receiver<schnorr::Signature>, BridgeError> {
+        let (sig_tx, sig_rx) = mpsc::channel(1280);
+
+        let mut sighash_stream = Box::pin(create_operator_sighash_stream(
+            self.db.clone(),
+            self.idx,
+            self.collateral_funding_txid,
+            self.signer.xonly_public_key,
+            self.config.clone(),
+            deposit_outpoint,
+            evm_address,
+            recovery_taproot_address,
+            self.nofn_xonly_pk,
+            user_takes_after,
+            Amount::from_sat(200_000_000), // TODO: Fix this.
+            6,
+            100,
+            self.config.bridge_amount_sats,
+            self.config.network,
+        ));
+
+        let operator = self.clone();
+        tokio::spawn(async move {
+            while let Some(sighash) = sighash_stream.next().await {
+                // None because utxos that operators need to sign do not have scripts
+                let sig = operator
+                    .signer
+                    .sign_with_tweak(sighash.expect("Expected sig"), None)
+                    .expect("Error while sigbing with tweak");
+
+                if sig_tx.send(sig).await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        Ok(sig_rx)
+    }
+
     // /// Public endpoint for every depositor to call.
     // ///
     // /// It will get signatures from all verifiers:

core/src/rpc/operator.rs

@@ -4,12 +4,9 @@ use super::clementine::{
     WithdrawalFinalizedParams,
 };
 use super::error::*;
-use crate::builder::sighash::create_operator_sighash_stream;
 use crate::rpc::parser;
 use crate::{errors::BridgeError, operator::Operator};
-use bitcoin::{Amount, OutPoint};
-use futures::StreamExt;
-use std::pin::pin;
+use bitcoin::OutPoint;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::{async_trait, Request, Response, Status};
@@ -59,48 +56,30 @@ impl ClementineOperator for Operator {
         &self,
         request: Request<DepositSignSession>,
     ) -> Result<Response<Self::DepositSignStream>, Status> {
-        let operator = self.clone();
         let (tx, rx) = mpsc::channel(1280);
         let deposit_sign_session = request.into_inner();
 
         let (deposit_outpoint, evm_address, recovery_taproot_address, user_takes_after) =
             parser::parse_deposit_params(deposit_sign_session.try_into()?)?;
 
-        tokio::spawn(async move {
-            let mut sighash_stream = pin!(create_operator_sighash_stream(
-                operator.db,
-                operator.idx,
-                operator.collateral_funding_txid,
-                operator.signer.xonly_public_key,
-                operator.config.clone(),
+        let mut deposit_signatures_channel = self
+            .deposit_sign(
                 deposit_outpoint,
                 evm_address,
                 recovery_taproot_address,
-                operator.nofn_xonly_pk,
                 user_takes_after,
-                Amount::from_sat(200_000_000), // TODO: Fix this.
-                6,
-                100,
-                operator.config.bridge_amount_sats,
-                operator.config.network,
-            ));
-
-            while let Some(sighash) = sighash_stream.next().await {
-                let sighash = sighash?;
-
-                // None because utxos that operators need to sign do not have scripts
-                let sig = operator.signer.sign_with_tweak(sighash, None)?;
-                let operator_burn_sig = OperatorBurnSig {
-                    schnorr_sig: sig.serialize().to_vec(),
-                };
-
-                if tx.send(Ok(operator_burn_sig)).await.is_err() {
-                    break;
-                }
-            }
+            )
+            .await?;
 
-            Ok::<_, BridgeError>(())
-        });
+        while let Some(sig) = deposit_signatures_channel.recv().await {
+            let operator_burn_sig = OperatorBurnSig {
+                schnorr_sig: sig.serialize().to_vec(),
+            };
+
+            if tx.send(Ok(operator_burn_sig)).await.is_err() {
+                break;
+            }
+        }
 
         Ok(Response::new(ReceiverStream::new(rx)))
     }