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

crates/bridge-exec/Cargo.toml

@@ -14,6 +14,7 @@ rustdoc.all = "warn"
 
 [dependencies]
 alpen-express-btcio = { workspace = true }
+alpen-express-db = { 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

@@ -3,9 +3,13 @@
 use std::sync::Arc;
 
 use alpen_express_btcio::rpc::traits::Signer;
+use alpen_express_db::entities::bridge_tx_state::BridgeTxState;
 use alpen_express_primitives::bridge::OperatorPartialSig;
 use alpen_express_rpc_api::AlpenApiClient;
-use bitcoin::secp256k1::{schnorr::Signature, SECP256K1};
+use bitcoin::{
+    secp256k1::{schnorr::Signature, SECP256K1},
+    Txid,
+};
 use express_bridge_sig_manager::manager::SignatureManager;
 use express_bridge_tx_builder::{deposit::DepositInfo, prelude::*, TxKind};
 use express_storage::ops::bridge::BridgeTxStateOps;
@@ -124,12 +128,86 @@ 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!()
+/// Aggregate the received signature with the ones already accumulated
+/// into a fully signed [`Signature`].
+///
+/// This is executed by the bridge operator that is assigned the given deposit.
+///
+/// Returns [`None`] if the transaction in [`BridgeTxStateOps`] is not fully signed
+/// by all remaining operators.
+pub async fn aggregate_signature(
+    txid: &Txid,
+    sig: &OperatorPartialSig,
+    db_ops: &Arc<BridgeTxStateOps>,
+) -> DepositExecResult<Option<Signature>> {
+    // setup logging
+    let span = span!(
+        Level::INFO,
+        "starting deposit transaction signature aggregation"
+    );
+    let _guard = span.enter();
+
+    // aggregates using MuSig2 the OperatorPartialSig into the BridgeStateOps
+    // checks if is fully complete
+    let mut tx_state = get_tx_state_by_txid(db_ops, txid).await?;
+
+    event!(
+        Level::DEBUG,
+        event = "got an updated transaction state",
+        %txid,
+        ?tx_state
+    );
+
+    let is_fully_signed = tx_state
+        .add_signature(*sig)
+        .map_err(|e| DepositExecError::Execution(e.to_string()))?;
+
+    event!(
+        Level::INFO,
+        event = "transaction is or isn't fully signed",
+        %is_fully_signed
+    );
+
+    if is_fully_signed {
+        // get a new up-to-date transaction state
+        let tx_state = get_tx_state_by_txid(db_ops, txid).await?;
+        let sig = tx_state
+            .aggregate_signature()
+            .map_err(|e| DepositExecError::Execution(e.to_string()))?;
+
+        event!(
+            Level::INFO,
+            event = "aggregated final signature",
+            %sig
+        );
+
+        Ok(Some(sig))
+    } else {
+        event!(
+            Level::WARN,
+            event = "could not aggregate final signature, missing partial sigs",
+        );
+        Ok(None)
+    }
+}
+
+/// Gets the [`BridgeTxState`] by [`Txid`].
+///
+/// Errors if cannot find the `txid` in the database,
+/// or if the database returns an error.
+async fn get_tx_state_by_txid(
+    db_ops: &Arc<BridgeTxStateOps>,
+    txid: &Txid,
+) -> DepositExecResult<BridgeTxState> {
+    let tx_state = db_ops
+        .get_tx_state_async(*txid)
+        .await
+        .map_err(|e| DepositExecError::Execution(e.to_string()))?;
+
+    match tx_state {
+        Some(tx_state) => Ok(tx_state),
+        None => Err(DepositExecError::InvalidRequest(
+            "could not find a transaction state for txid: {txid}".to_string(),
+        )),
+    }
 }

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

@@ -5,11 +5,12 @@
 use std::sync::Arc;
 
 use alpen_express_btcio::rpc::traits::{Signer, Wallet};
+use alpen_express_db::entities::bridge_tx_state::BridgeTxState;
 use alpen_express_primitives::{bridge::OperatorPartialSig, l1::BitcoinAmount};
 use alpen_express_rpc_api::AlpenApiClient;
 use bitcoin::{
     secp256k1::{schnorr::Signature, SECP256K1},
-    Address, OutPoint,
+    Address, OutPoint, Txid,
 };
 use express_bridge_sig_manager::manager::SignatureManager;
 use express_bridge_tx_builder::{prelude::*, withdrawal::CooperativeWithdrawalInfo, TxKind};
@@ -160,13 +161,86 @@ pub async fn sign_withdraw_tx(
     }
 }
 
-/// Aggregate the received signature with the ones already accumulated.
+/// Aggregate the received signature with the ones already accumulated
+/// into a fully signed [`Signature`].
 ///
 /// This is executed by the bridge operator that is assigned the given withdrawal.
-// TODO: pass in a database client once the database traits have been implemented.
-pub async fn aggregate_withdrawal_sig(
-    _withdrawal_info: &CooperativeWithdrawalInfo,
-    _sig: &OperatorPartialSig,
+///
+/// Returns [`None`] if the transaction in [`BridgeTxStateOps`] is not fully signed
+/// by all remaining operators.
+pub async fn aggregate_withdraw_sig(
+    txid: &Txid,
+    sig: &OperatorPartialSig,
+    db_ops: &Arc<BridgeTxStateOps>,
 ) -> WithdrawalExecResult<Option<Signature>> {
-    unimplemented!()
+    // setup logging
+    let span = span!(
+        Level::INFO,
+        "starting withdrawal transaction signature aggregation"
+    );
+    let _guard = span.enter();
+
+    // aggregates using MuSig2 the OperatorPartialSig into the BridgeStateOps
+    // checks if is fully complete
+    let mut tx_state = get_tx_state_by_txid(db_ops, txid).await?;
+
+    event!(
+        Level::DEBUG,
+        event = "got an updated transaction state",
+        %txid,
+        ?tx_state
+    );
+
+    let is_fully_signed = tx_state
+        .add_signature(*sig)
+        .map_err(|e| WithdrawalExecError::Execution(e.to_string()))?;
+
+    event!(
+        Level::INFO,
+        event = "transaction is or isn't fully signed",
+        %is_fully_signed
+    );
+
+    if is_fully_signed {
+        // get a new up-to-date transaction state
+        let tx_state = get_tx_state_by_txid(db_ops, txid).await?;
+        let sig = tx_state
+            .aggregate_signature()
+            .map_err(|e| WithdrawalExecError::Execution(e.to_string()))?;
+
+        event!(
+            Level::INFO,
+            event = "aggregated final signature",
+            %sig
+        );
+
+        Ok(Some(sig))
+    } else {
+        event!(
+            Level::WARN,
+            event = "could not aggregate final signature, missing partial sigs",
+        );
+        Ok(None)
+    }
+}
+
+/// Gets the [`BridgeTxState`] by [`Txid`].
+///
+/// Errors if cannot find the `txid` in the database,
+/// or if the database returns an error.
+async fn get_tx_state_by_txid(
+    db_ops: &Arc<BridgeTxStateOps>,
+    txid: &Txid,
+) -> WithdrawalExecResult<BridgeTxState> {
+    let tx_state = db_ops
+        .get_tx_state_async(*txid)
+        .await
+        .map_err(|e| WithdrawalExecError::Execution(e.to_string()))?;
+
+    match tx_state {
+        Some(tx_state) => Ok(tx_state),
+        None => Err(WithdrawalExecError::InvalidRequest(
+            "could not find a transaction state for txid: {txid}".to_string(),
+        )),
+    }
 }

crates/db/src/entities/bridge_tx_state.rs

@@ -11,9 +11,12 @@ use alpen_express_primitives::{
     l1::{BitcoinPsbt, TaprootSpendPath},
 };
 use arbitrary::Arbitrary;
-use bitcoin::{Transaction, TxOut, Txid};
+use bitcoin::{hashes::Hash, Transaction, TxOut, Txid};
 use borsh::{BorshDeserialize, BorshSerialize};
-use musig2::{PartialSignature, PubNonce};
+use musig2::{
+    aggregate_partial_signatures, secp256k1::schnorr::Signature, AggNonce, KeyAggContext,
+    PartialSignature, PubNonce,
+};
 
 use super::errors::{BridgeTxStateError, EntityResult};
 
@@ -210,6 +213,23 @@ impl BridgeTxState {
     pub fn ordered_sigs(&self) -> impl Iterator<Item = PartialSignature> + '_ {
         self.collected_sigs.values().map(|v| *v.inner())
     }
+
+    /// Aggregates and extracts the underlying partial signatures as a [`Signature`].
+    pub fn aggregate_signature(&self) -> EntityResult<Signature> {
+        let message = *self.compute_txid().as_byte_array();
+        let key_agg_context: KeyAggContext = self
+            .pubkey_table
+            .clone()
+            .try_into()
+            .map_err(|_| BridgeTxStateError::Unauthorized)?;
+        let pub_nonces = self.collected_nonces.values().map(|n| n.inner().clone());
+        let aggregate_nonce = AggNonce::sum(pub_nonces);
+        let partial_sigs = self.collected_sigs.values().map(|s| *s.inner());
+        let sig: Signature =
+            aggregate_partial_signatures(&key_agg_context, &aggregate_nonce, partial_sigs, message)
+                .map_err(|_| BridgeTxStateError::Unauthorized)?;
+        Ok(sig)
+    }
 }
 
 #[cfg(test)]

crates/primitives/src/bridge.rs

@@ -11,7 +11,10 @@ use bitcoin::{
     secp256k1::{PublicKey, SecretKey},
 };
 use borsh::{BorshDeserialize, BorshSerialize};
-use musig2::{BinaryEncoding, NonceSeed, PartialSignature, PubNonce, SecNonce};
+use musig2::{
+    errors::KeyAggError, BinaryEncoding, KeyAggContext, NonceSeed, PartialSignature, PubNonce,
+    SecNonce,
+};
 use serde::{Deserialize, Serialize};
 
 use crate::{
@@ -45,6 +48,14 @@ impl From<PublickeyTable> for Vec<PublicKey> {
     }
 }
 
+impl TryFrom<PublickeyTable> for KeyAggContext {
+    type Error = KeyAggError;
+
+    fn try_from(value: PublickeyTable) -> Result<Self, Self::Error> {
+        KeyAggContext::new(Into::<Vec<PublicKey>>::into(value))
+    }
+}
+
 impl BorshSerialize for PublickeyTable {
     fn serialize<W: Write>(&self, writer: &mut W) -> std::io::Result<()> {
         // Serialize the length of the BTreeMap