feat(bridge-exec)!: aggregate_[deposit|withdraw]_sig

crates/bridge-exec/Cargo.toml

@@ -13,6 +13,7 @@ rust.unused_must_use = "deny"
 rustdoc.all = "warn"
 
 [dependencies]
+alpen-express-btcio = { workspace = true }
 alpen-express-primitives = { workspace = true }
 alpen-express-rpc-api = { workspace = true, features = ["client"] }
 express-bridge-sig-manager = { workspace = true }

crates/bridge-exec/src/deposit_handler/handler.rs

@@ -8,7 +8,7 @@ use alpen_express_primitives::{
     relay::{types::Scope, util::MessageSigner},
 };
 use alpen_express_rpc_api::AlpenApiClient;
-use bitcoin::{secp256k1::SECP256K1, Txid};
+use bitcoin::{secp256k1::SECP256K1, Transaction, Txid};
 use express_bridge_sig_manager::manager::SignatureManager;
 use express_bridge_tx_builder::{deposit::DepositInfo, prelude::*, TxKind};
 use jsonrpsee::tokio::time::{sleep, Duration};
@@ -180,12 +180,175 @@ pub async fn sign_deposit_tx(
     }
 }
 
-/// Add the signature to the already accumulated set of signatures for a deposit transaction and
-/// produce the aggregated signature if all operators have signed. Also update the database
-/// entry with the signatures accumulated so far.
-//
-// TODO: this method will also accept a `BridgeMessage` that holds the signature attached to a
-// particular deposit info by other operators.
-pub async fn aggregate_signature() -> DepositExecResult<Option<Signature>> {
-    unimplemented!()
+/// Pools and aggregates signatures for the deposit transaction
+/// into a fully-signed ready-to-be-broadcasted Bitcoin [`Transaction`].
+///
+/// Also broadcasts to the bridge transaction database.
+///
+/// # Arguments
+///
+/// - `txid`: a [`Txid`] that is in the bridge transaction database.
+/// - `l1_rpc_client`: anything that is able to sign transactions and messages; i.e. holds private
+///   keys.
+/// - `l2_rpc_client`: anything that can communicate with the [`AlpenApiClient`].
+/// - `sig_manager`: a stateful [`SignatureManager`].
+/// - `tx_build_context`: stateful [`TxBuildContext`].
+///
+/// # Notes
+///
+/// Both the [`SignatureManager`] and the [`TxBuildContext`] can be reused
+/// for multiple signing sessions if the operators'
+/// [`PublickeyTable`](alpen_express_primitives::bridge::PublickeyTable)
+/// does _not_ change.
+///
+/// We don't need mutexes since all functions to [`SignatureManager`] and
+/// [`TxBuildContext`] takes non-mutable references.
+pub async fn aggregate_deposit_sig(
+    txid: &Txid,
+    l1_rpc_client: &Arc<impl Signer>,
+    l2_rpc_client: &Arc<impl AlpenApiClient + Sync>,
+    sig_manager: &Arc<SignatureManager>,
+    tx_build_context: &Arc<TxBuildContext>,
+) -> DepositExecResult<Transaction> {
+    // setup logging
+    let span = span!(
+        Level::INFO,
+        "starting deposit transaction signature aggregation"
+    );
+    let _guard = span.enter();
+
+    let operator_pubkeys = tx_build_context.pubkey_table();
+    let own_index = tx_build_context.own_index();
+    let own_pubkey = operator_pubkeys
+        .0
+        .get(&own_index)
+        .expect("could not find operator's pubkey in public key table");
+
+    event!(
+        Level::INFO,
+        event = "got the basic self information",
+        %txid,
+        %own_index,
+        %own_pubkey,
+    );
+
+    let tx_state = sig_manager
+        .get_tx_state(txid)
+        .await
+        .map_err(|e| DepositExecError::Signing(e.to_string()))?;
+
+    event!(
+        Level::DEBUG,
+        event = "got transaction state from bridge database",
+        %txid,
+        ?tx_state,
+    );
+
+    // Fully signed and in the database, nothing to do here...
+    if tx_state.is_fully_signed() {
+        event!(
+            Level::INFO,
+            event = "deposit transaction fully signed and in the bridge database",
+            %txid,
+        );
+        let tx = sig_manager
+            .finalize_transaction(txid)
+            .await
+            .map_err(|e| DepositExecError::Signing(e.to_string()))?;
+        return Ok(tx);
+    }
+
+    // Not fully signed, then partially sign transaction, construct, and sign a message
+    let xpriv = l1_rpc_client.get_xpriv().await?;
+    if let Some(xpriv) = xpriv {
+        let keypair = xpriv.to_keypair(SECP256K1);
+
+        // First check if it needs this operator's partial signature
+        let needs_our_sig = tx_state.collected_sigs().get(&own_index).is_none();
+        if needs_our_sig {
+            sig_manager
+                .add_own_partial_sig(txid)
+                .await
+                .map_err(|e| DepositExecError::Signing(e.to_string()))?;
+            event!(
+                Level::INFO,
+                event = "added own's partial signature to the bridge transaction database",
+                %txid,
+            );
+        }
+
+        // Now, get this operator's partial sig
+        let partial_sig = sig_manager
+            .get_own_partial_sig(txid)
+            .await
+            .map_err(|e| DepositExecError::Signing(e.to_string()))?
+            .expect("should've been signed");
+
+        event!(
+            Level::INFO,
+            event = "got own's partial signature from the bridge transaction database",
+            ?partial_sig,
+        );
+
+        // submit_message RPC call
+        let bitcoin_txid: BitcoinTxid = (*txid).into();
+        let scope = Scope::V0DepositSig(bitcoin_txid);
+        event!(
+            Level::DEBUG,
+            event = "create the deposit partial signature scope",
+            ?scope
+        );
+        let message = MessageSigner::new(own_index, keypair.secret_key().into())
+            .sign_scope(&scope, &partial_sig)
+            .map_err(|e| DepositExecError::Signing(e.to_string()))?;
+        event!(
+            Level::DEBUG,
+            event = "create the deposit partial signature message",
+            ?message,
+        );
+        let raw_message: Vec<u8> = message
+            .try_into()
+            .expect("could not serialize bridge message into raw bytes");
+
+        l2_rpc_client.submit_bridge_msg(raw_message.into()).await?;
+        event!(
+            Level::INFO,
+            event = "broadcasted the withdrawal partial signature message",
+        );
+
+        // Wait for all the partial signatures to be broadcasted by other operators.
+        // Collect all partial signature.
+        loop {
+            event!(
+                Level::DEBUG,
+                event = "trying to get all partial signatures from the bridge transaction database, waiting for other operators' signatures",
+            );
+            let got_all_sigs = sig_manager
+                .get_tx_state(txid)
+                .await
+                .map_err(|e| DepositExecError::Signing(e.to_string()))?
+                .is_fully_signed();
+            if got_all_sigs {
+                event!(
+                    Level::INFO,
+                    event = "got all partial signatures from the bridge transaction database",
+                    %got_all_sigs
+                );
+                break;
+            } else {
+                // TODO: this is hardcoded, maybe move this to a user-configurable Config
+                sleep(Duration::from_millis(100)).await;
+                continue;
+            }
+        }
+        let tx = sig_manager
+            .finalize_transaction(txid)
+            .await
+            .map_err(|e| DepositExecError::Signing(e.to_string()))?;
+        Ok(tx)
+    } else {
+        Err(DepositExecError::Signing(
+            "Could not get wallet's xpriv".to_string(),
+        ))
+    }
 }