feat: add batch drop

migrations/20250209015007_add_batch_cancel_ledger_tx_id_to_batch_wallet_summaries.down.sql

@@ -0,0 +1 @@
+-- Add down migration script here

migrations/20250209015007_add_batch_cancel_ledger_tx_id_to_batch_wallet_summaries.up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE bria_batch_wallet_summaries
+ADD COLUMN batch_cancel_ledger_tx_id UUID;

proto/api/bria.proto

@@ -41,6 +41,7 @@ service BriaService {
   rpc CancelPayout(CancelPayoutRequest) returns (CancelPayoutResponse) {}
 
   rpc GetBatch (GetBatchRequest) returns (GetBatchResponse) {}
+  rpc CancelBatch (CancelBatchRequest) returns (CancelBatchResponse) {}
 
   rpc GetAccountBalanceSummary (GetAccountBalanceSummaryRequest) returns (GetAccountBalanceSummaryResponse) {}
 
@@ -434,6 +435,12 @@ message GetBatchResponse {
   repeated SigningSession signing_sessions = 6;
 }
 
+message CancelBatchRequest {
+  string id = 1;
+}
+
+message CancelBatchResponse {}
+
 message BatchWalletSummary {
   string wallet_id = 1;
   uint64 total_spent_sats = 2;

src/api/server/mod.rs

@@ -744,6 +744,7 @@ impl BriaService for Bria {
                     &profile,
                     id.parse()
                         .map_err(ApplicationError::CouldNotParseIncomingUuid)?,
+                    false,
                 )
                 .await?;
             Ok(Response::new(CancelPayoutResponse {}))
@@ -877,6 +878,29 @@ impl BriaService for Bria {
         .await
     }
 
+    #[instrument(name = "bria.cancel_batch", skip_all, fields(error, error.level, error.message), err)]
+    async fn cancel_batch(
+        &self,
+        request: Request<CancelBatchRequest>,
+    ) -> Result<Response<CancelBatchResponse>, Status> {
+        crate::tracing::record_error(|| async move {
+            extract_tracing(&request);
+            let key = extract_api_token(&request)?;
+            let profile = self.app.authenticate(key).await?;
+            let request = request.into_inner();
+            let CancelBatchRequest { id } = request;
+            self.app
+                .cancel_batch(
+                    &profile,
+                    id.parse()
+                        .map_err(ApplicationError::CouldNotParseIncomingUuid)?,
+                )
+                .await?;
+            Ok(Response::new(CancelBatchResponse {}))
+        })
+        .await
+    }
+
     type SubscribeAllStream = std::pin::Pin<
         Box<dyn futures::Stream<Item = Result<BriaEvent, Status>> + Send + Sync + 'static>,
     >;

src/app/mod.rs

@@ -12,7 +12,7 @@ use error::*;
 use crate::{
     account::balance::AccountBalanceSummary,
     address::*,
-    batch::*,
+    batch::{error::BatchError, *},
     batch_inclusion::*,
     descriptor::*,
     fees::{self, *},
@@ -891,13 +891,14 @@ impl App {
         &self,
         profile: &Profile,
         id: PayoutId,
+        skip_committed_check: bool,
     ) -> Result<(), ApplicationError> {
         let mut tx = self.pool.begin().await?;
         let mut payout = self
             .payouts
             .find_by_id_for_cancellation(&mut tx, profile.account_id, id)
             .await?;
-        payout.cancel_payout(profile.id)?;
+        payout.cancel_payout(profile.id, skip_committed_check)?;
         self.payouts.update(&mut tx, payout).await?;
         self.ledger
             .payout_cancelled(tx, LedgerTransactionId::new(), id)
@@ -1025,6 +1026,50 @@ impl App {
         Ok((batch, payouts, signing_sessions))
     }
 
+    #[allow(clippy::type_complexity)]
+    #[instrument(name = "app.cancel_batch", skip_all, err)]
+    pub async fn cancel_batch(
+        &self,
+        profile: &Profile,
+        batch_id: BatchId,
+    ) -> Result<(), ApplicationError> {
+        let batch = self
+            .batches
+            .find_by_id(profile.account_id, batch_id)
+            .await?;
+
+        batch
+            .validate_cancellation()
+            .map_err(ApplicationError::BatchError)?;
+
+        let (tx, batch_info, tx_id) = self
+            .batches
+            .set_batch_cancel_ledger_tx_id(batch.id)
+            .await?
+            .ok_or_else(|| {
+                ApplicationError::BatchError(BatchError::BatchIdNotFound(batch_id.to_string()))
+            })?;
+
+        self.ledger
+            .batch_dropped(tx, tx_id, batch_info.created_ledger_tx_id)
+            .await?;
+
+        let payouts = self
+            .payouts
+            .list_for_batch(profile.account_id, batch_id)
+            .await?;
+
+        for (_wallet_id, wallet_payouts) in payouts.iter() {
+            for payout in wallet_payouts {
+                // Skip committed check since we're cancelling the whole batch
+                // and we've already verified the batch isn't signed
+                self.cancel_payout(profile, payout.id, true).await.ok();
+            }
+        }
+
+        Ok(())
+    }
+
     #[instrument(name = "app.subscribe_all", skip(self), err)]
     pub async fn subscribe_all(
         &self,

src/batch/entity.rs

@@ -4,6 +4,8 @@ use std::collections::HashMap;
 
 use crate::primitives::*;
 
+use super::error::BatchError;
+
 pub struct Batch {
     pub id: BatchId,
     pub account_id: AccountId,
@@ -15,11 +17,41 @@ pub struct Batch {
 }
 
 impl Batch {
+    pub fn get_tx_to_broadcast(&self) -> Option<bitcoin::Transaction> {
+        if self.accounting_complete() && self.is_signed() && !self.is_cancelled() {
+            self.signed_tx.clone()
+        } else {
+            None
+        }
+    }
+
     pub fn accounting_complete(&self) -> bool {
         self.wallet_summaries
             .values()
             .all(|s| s.batch_created_ledger_tx_id.is_some())
     }
+
+    pub fn validate_cancellation(&self) -> Result<(), BatchError> {
+        if self.is_cancelled() {
+            return Err(BatchError::BatchAlreadyCancelled);
+        }
+
+        if self.is_signed() {
+            return Err(BatchError::BatchAlreadySigned);
+        }
+
+        Ok(())
+    }
+
+    pub fn is_cancelled(&self) -> bool {
+        self.wallet_summaries
+            .values()
+            .all(|s| s.batch_cancel_ledger_tx_id.is_some())
+    }
+
+    pub fn is_signed(&self) -> bool {
+        self.signed_tx.is_some()
+    }
 }
 
 #[derive(Builder, Clone)]
@@ -61,4 +93,5 @@ pub struct WalletSummary {
     pub change_outpoint: Option<bitcoin::OutPoint>,
     pub batch_created_ledger_tx_id: Option<LedgerTransactionId>,
     pub batch_broadcast_ledger_tx_id: Option<LedgerTransactionId>,
+    pub batch_cancel_ledger_tx_id: Option<LedgerTransactionId>,
 }

src/batch/error.rs

@@ -12,4 +12,8 @@ pub enum BatchError {
     BitcoinConsensusEncodeError(#[from] crate::primitives::bitcoin::consensus::encode::Error),
     #[error("BatchError - Could not deserialize PSBT: {0}")]
     PsbtDeserializationError(#[from] crate::primitives::bitcoin::psbt::Error),
+    #[error("BatchError - Batch is already cancelled")]
+    BatchAlreadyCancelled,
+    #[error("BatchError - Batch is already signed and can't be cancelled")]
+    BatchAlreadySigned,
 }

src/batch/repo.rs

@@ -45,7 +45,7 @@ impl Batches {
             r#"INSERT INTO bria_batch_wallet_summaries (
                    batch_id, wallet_id, current_keychain_id, signing_keychains, total_in_sats,
                    total_spent_sats, change_sats, change_address, change_vout, total_fee_sats,
-                   cpfp_fee_sats, cpfp_details, batch_created_ledger_tx_id, batch_broadcast_ledger_tx_id
+                   cpfp_fee_sats, cpfp_details, batch_created_ledger_tx_id, batch_broadcast_ledger_tx_id, batch_cancel_ledger_tx_id
                )"#,
         );
         query_builder.push_values(
@@ -71,6 +71,7 @@ impl Batches {
                 builder.push_bind(serde_json::to_value(summary.cpfp_details).unwrap());
                 builder.push_bind(summary.batch_created_ledger_tx_id);
                 builder.push_bind(summary.batch_broadcast_ledger_tx_id);
+                builder.push_bind(summary.batch_cancel_ledger_tx_id);
             },
         );
         let query = query_builder.build();
@@ -90,7 +91,7 @@ impl Batches {
                     payout_queue_id, unsigned_psbt, signed_tx, bitcoin_tx_id, s.batch_id,
                     s.wallet_id, s.current_keychain_id, s.signing_keychains, total_in_sats,
                     total_spent_sats, change_sats, change_address, change_vout, s.total_fee_sats,
-                    cpfp_fee_sats, cpfp_details, batch_created_ledger_tx_id, batch_broadcast_ledger_tx_id
+                    cpfp_fee_sats, cpfp_details, batch_created_ledger_tx_id, batch_broadcast_ledger_tx_id, batch_cancel_ledger_tx_id
             FROM bria_batch_wallet_summaries s
             LEFT JOIN bria_batches b ON b.id = s.batch_id
             WHERE s.batch_id = $1 AND b.account_id = $2"#,
@@ -148,6 +149,7 @@ impl Batches {
                     batch_broadcast_ledger_tx_id: row
                         .batch_broadcast_ledger_tx_id
                         .map(LedgerTxId::from),
+                    batch_cancel_ledger_tx_id: row.batch_cancel_ledger_tx_id.map(LedgerTxId::from),
                 },
             );
         }
@@ -274,4 +276,64 @@ impl Batches {
             ledger_transaction_id,
         )))
     }
+
+    #[instrument(name = "batches.set_batch_cancel_ledger_tx_id", skip(self))]
+    pub async fn set_batch_cancel_ledger_tx_id(
+        &self,
+        batch_id: BatchId,
+    ) -> Result<Option<(Transaction<'_, Postgres>, BatchInfo, LedgerTxId)>, BatchError> {
+        let mut tx = self.pool.begin().await?;
+        let row = sqlx::query!(
+            r#"SELECT
+                bb.id,
+                bb.payout_queue_id,
+                bbws.batch_cancel_ledger_tx_id as "ledger_id?",
+                bbws.batch_created_ledger_tx_id
+            FROM bria_batches bb
+            INNER JOIN bria_batch_wallet_summaries bbws ON bb.id = bbws.batch_id
+            WHERE bb.id = $1
+            FOR UPDATE"#,
+            batch_id as BatchId,
+        )
+        .fetch_optional(&mut *tx)
+        .await?;
+        if row.is_none() || row.as_ref().unwrap().batch_created_ledger_tx_id.is_none() {
+            return Ok(None);
+        }
+        let row = row.unwrap();
+        let created_ledger_tx_id = LedgerTxId::from(row.batch_created_ledger_tx_id.unwrap());
+        let batch_id = BatchId::from(row.id);
+        let payout_queue_id = PayoutQueueId::from(row.payout_queue_id);
+        if row.ledger_id.is_some() {
+            return Ok(Some((
+                tx,
+                BatchInfo {
+                    id: batch_id,
+                    payout_queue_id,
+                    created_ledger_tx_id,
+                },
+                LedgerTxId::from(row.ledger_id.unwrap()),
+            )));
+        }
+        let ledger_transaction_id = LedgerTxId::new();
+        sqlx::query!(
+            r#"UPDATE bria_batch_wallet_summaries
+               SET batch_cancel_ledger_tx_id = $1
+               WHERE bria_batch_wallet_summaries.batch_id = $2"#,
+            ledger_transaction_id as LedgerTxId,
+            batch_id as BatchId,
+        )
+        .execute(&mut *tx)
+        .await?;
+
+        Ok(Some((
+            tx,
+            BatchInfo {
+                id: batch_id,
+                payout_queue_id,
+                created_ledger_tx_id,
+            },
+            ledger_transaction_id,
+        )))
+    }
 }