cli: Add receiver support to cancel command

Author: spacebear21

payjoin-cli/src/app/mod.rs

@@ -11,6 +11,8 @@ pub mod wallet;
 use crate::app::config::Config;
 use crate::app::wallet::BitcoindWallet;
 #[cfg(feature = "v2")]
+use crate::cli::Role;
+#[cfg(feature = "v2")]
 use crate::db::v2::SessionId;
 
 #[cfg(feature = "v1")]
@@ -31,7 +33,12 @@ pub trait App: Send + Sync {
     #[cfg(feature = "v2")]
     async fn history(&self) -> Result<()>;
     #[cfg(feature = "v2")]
-    async fn cancel_sender(&self, session_id: SessionId, no_broadcast: bool) -> Result<()>;
+    async fn cancel(
+        &self,
+        session_id: SessionId,
+        no_broadcast: bool,
+        role: Option<Role>,
+    ) -> Result<()>;
 
     fn create_original_psbt(
         &self,

payjoin-cli/src/app/v1.rs

@@ -129,10 +129,11 @@ impl AppTrait for App {
     }
 
     #[cfg(feature = "v2")]
-    async fn cancel_sender(
+    async fn cancel(
         &self,
         _session_id: crate::db::v2::SessionId,
         _no_broadcast: bool,
+        _role: Option<crate::cli::Role>,
     ) -> Result<()> {
         anyhow::bail!("cancel is only supported for v2 (BIP77) sessions")
     }

payjoin-cli/src/app/v2/mod.rs

@@ -8,13 +8,14 @@ use payjoin::persist::{OptionalTransitionOutcome, SessionPersister};
 use payjoin::receive::v2::{
     replay_event_log as replay_receiver_event_log, HasReplyableError, Initialized,
     MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, PayjoinProposal,
-    ProvisionalProposal, ReceiveSession, Receiver, ReceiverBuilder,
-    SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload, WantsFeeRange, WantsInputs,
-    WantsOutputs,
+    PendingFallback as ReceiverPendingFallback, ProvisionalProposal, ReceiveSession, Receiver,
+    ReceiverBuilder, SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload,
+    WantsFeeRange, WantsInputs, WantsOutputs,
 };
 use payjoin::send::v2::{
-    replay_event_log as replay_sender_event_log, PendingFallback, PollingForProposal, SendSession,
-    Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome, WithReplyKey,
+    replay_event_log as replay_sender_event_log, PendingFallback as SenderPendingFallback,
+    PollingForProposal, SendSession, Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome,
+    WithReplyKey,
 };
 use payjoin::{ImplementationError, PjParam, Uri};
 use tokio::sync::watch;
@@ -24,6 +25,7 @@ use super::wallet::BitcoindWallet;
 use super::App as AppTrait;
 use crate::app::v2::ohttp::{unwrap_ohttp_keys_or_else_fetch, RelayManager};
 use crate::app::{handle_interrupt, http_agent};
+use crate::cli::Role as CliRole;
 use crate::db::v2::{ReceiverPersister, SenderPersister, SessionId};
 use crate::db::Database;
 
@@ -486,11 +488,42 @@ impl AppTrait for App {
         Ok(())
     }
 
-    async fn cancel_sender(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> {
+    async fn cancel(
+        &self,
+        session_id: SessionId,
+        no_broadcast: bool,
+        role: Option<CliRole>,
+    ) -> Result<()> {
+        if let Some(role) = role {
+            return match role {
+                CliRole::Sender => self.cancel_sender_session(session_id, no_broadcast),
+                CliRole::Receiver => self.cancel_receiver_session(session_id, no_broadcast),
+            };
+        }
+
+        let send_ids = self.db.get_send_session_ids()?;
+        let recv_ids = self.db.get_recv_session_ids()?;
+        let is_sender = send_ids.iter().any(|id| id.0 == session_id.0);
+        let is_receiver = recv_ids.iter().any(|id| id.0 == session_id.0);
+
+        match (is_sender, is_receiver) {
+            (true, false) => self.cancel_sender_session(session_id, no_broadcast),
+            (false, true) => self.cancel_receiver_session(session_id, no_broadcast),
+            (true, true) => anyhow::bail!(
+                "Session {session_id} exists as both a sender and receiver session. \
+                 Pass `--role sender` or `--role receiver`."
+            ),
+            (false, false) => anyhow::bail!("Session {session_id} not found"),
+        }
+    }
+}
+
+impl App {
+    fn cancel_sender_session(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> {
         let persister = SenderPersister::from_id(self.db.clone(), session_id.clone());
         let (session, _history) = replay_sender_event_log(&persister)?;
 
-        let pending: Sender<PendingFallback> = match session {
+        let pending: Sender<SenderPendingFallback> = match session {
             SendSession::WithReplyKey(sender) => sender.cancel().save(&persister)?,
             SendSession::PollingForProposal(sender) => sender.cancel().save(&persister)?,
             SendSession::PendingFallback(sender) => sender,
@@ -523,9 +556,72 @@ impl AppTrait for App {
         pending.close().save(&persister)?;
         Ok(())
     }
-}
 
-impl App {
+    fn cancel_receiver_session(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> {
+        let persister = ReceiverPersister::from_id(self.db.clone(), session_id.clone());
+        let (session, _history) = replay_receiver_event_log(&persister)?;
+
+        let pending: Receiver<ReceiverPendingFallback> = match session {
+            ReceiveSession::Initialized(receiver) => {
+                receiver.cancel().save(&persister)?;
+                println!("Session {session_id} cancelled. No fallback transaction to broadcast.");
+                return Ok(());
+            }
+            ReceiveSession::UncheckedOriginalPayload(receiver) => {
+                receiver.cancel().save(&persister)?;
+                println!("Session {session_id} cancelled. No fallback transaction to broadcast.");
+                return Ok(());
+            }
+            ReceiveSession::MaybeInputsOwned(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::MaybeInputsSeen(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::OutputsUnknown(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::WantsOutputs(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::WantsInputs(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::WantsFeeRange(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::ProvisionalProposal(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::PayjoinProposal(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::Monitor(receiver) => receiver.cancel().save(&persister)?,
+            ReceiveSession::HasReplyableError(receiver) => match receiver
+                .cancel()
+                .save(&persister)?
+            {
+                Some(pending) => pending,
+                None => {
+                    println!("Session {session_id} cancelled. No fallback transaction available.");
+                    return Ok(());
+                }
+            },
+            ReceiveSession::PendingFallback(receiver) => receiver,
+            ReceiveSession::Closed(
+                ReceiverSessionOutcome::Success(_)
+                | ReceiverSessionOutcome::FallbackBroadcasted
+                | ReceiverSessionOutcome::PayjoinProposalSent,
+            ) => {
+                println!("Session {session_id} already completed successfully. Cannot cancel.");
+                return Ok(());
+            }
+            ReceiveSession::Closed(_) => {
+                println!("Session {session_id} is already closed. Nothing left to do.");
+                return Ok(());
+            }
+        };
+
+        if no_broadcast {
+            println!(
+                "Session {session_id} cancelled. Broadcast the fallback transaction manually:\n{}",
+                serialize_hex(pending.fallback_tx())
+            );
+        } else {
+            self.wallet().broadcast_tx(pending.fallback_tx())?;
+            println!(
+                "Broadcasted fallback transaction txid: {}",
+                pending.fallback_tx().compute_txid()
+            );
+        }
+        pending.close().save(&persister)?;
+        Ok(())
+    }
+
     fn close_failed_session<P>(persister: &P, session_id: &SessionId, role: &str)
     where
         P: SessionPersister,
@@ -559,7 +655,7 @@ impl App {
             SendSession::PendingFallback(_) => {
                 let id = persister.session_id();
                 println!(
-                    "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the original transaction."
+                    "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction."
                 );
                 return Ok(());
             }
@@ -674,8 +770,13 @@ impl App {
                     self.handle_error(error, persister).await,
                 ReceiveSession::Monitor(proposal) =>
                     self.monitor_payjoin_proposal(proposal, persister).await,
-                ReceiveSession::PendingFallback(_) =>
-                    return Err(anyhow!("PendingFallback session handling not yet implemented")),
+                ReceiveSession::PendingFallback(_) => {
+                    let id = persister.session_id();
+                    println!(
+                        "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction."
+                    );
+                    return Ok(());
+                }
                 ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")),
             }
         };

payjoin-cli/src/cli/mod.rs

@@ -1,6 +1,6 @@
 use std::path::PathBuf;
 
-use clap::{value_parser, Parser, Subcommand};
+use clap::{value_parser, Parser, Subcommand, ValueEnum};
 use payjoin::bitcoin::amount::ParseAmountError;
 use payjoin::bitcoin::{Amount, FeeRate};
 use payjoin::Url;
@@ -133,7 +133,7 @@ pub enum Commands {
     /// Show payjoin session history
     History,
     #[cfg(feature = "v2")]
-    /// Cancel a sender session, broadcasting the fallback transaction by default (BIP77/v2 only)
+    /// Cancel a sender or receiver session, broadcasting the fallback transaction by default (BIP77/v2 only)
     Cancel {
         /// The session ID to cancel
         #[arg(required = true)]
@@ -142,9 +142,19 @@ pub enum Commands {
         /// Cancel without broadcasting the fallback transaction
         #[arg(long = "no-broadcast")]
         no_broadcast: bool,
+
+        /// The session role to cancel
+        #[arg(long = "role", value_enum)]
+        role: Option<Role>,
     },
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
+pub enum Role {
+    Sender,
+    Receiver,
+}
+
 pub fn parse_amount_in_sat(s: &str) -> Result<Amount, ParseAmountError> {
     Amount::from_str_in(s, payjoin::bitcoin::Denomination::Satoshi)
 }

payjoin-cli/src/db/v2.rs

@@ -141,6 +141,8 @@ impl ReceiverPersister {
     }
 
     pub fn from_id(db: Arc<Database>, id: SessionId) -> Self { Self { db, session_id: id } }
+
+    pub fn session_id(&self) -> SessionId { self.session_id.clone() }
 }
 
 impl SessionPersister for ReceiverPersister {

payjoin-cli/src/main.rs

@@ -79,8 +79,8 @@ async fn main() -> Result<()> {
             app.history().await?;
         }
         #[cfg(feature = "v2")]
-        Commands::Cancel { session_id, no_broadcast } => {
-            app.cancel_sender(SessionId(*session_id), *no_broadcast).await?;
+        Commands::Cancel { session_id, no_broadcast, role } => {
+            app.cancel(SessionId(*session_id), *no_broadcast, *role).await?;
         }
     };
 

payjoin-cli/tests/e2e.rs

@@ -754,4 +754,115 @@ mod e2e {
 
         Ok(())
     }
+
+    #[cfg(feature = "v2")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn receiver_cancel_v2() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        use payjoin_test_utils::{init_tracing, TestServices};
+        use tempfile::TempDir;
+
+        type Result<T> = std::result::Result<T, BoxError>;
+
+        init_tracing();
+        let mut services = TestServices::initialize().await?;
+        let temp_dir = tempdir()?;
+
+        let result = tokio::select! {
+            res = services.take_ohttp_relay_handle() => Err(format!("Ohttp relay is long running: {res:?}").into()),
+            res = services.take_directory_handle() => Err(format!("Directory server is long running: {res:?}").into()),
+            res = cancel_cli_async(&services, &temp_dir) => res,
+        };
+
+        assert!(result.is_ok(), "receiver_cancel_v2 failed: {:#?}", result.unwrap_err());
+
+        async fn cancel_cli_async(services: &TestServices, temp_dir: &TempDir) -> Result<()> {
+            let receiver_db_path = temp_dir.path().join("receiver_db");
+            let (bitcoind, _sender, _receiver) = init_bitcoind_sender_receiver(None, None)?;
+            let cert_path = &temp_dir.path().join("localhost.der");
+            tokio::fs::write(cert_path, services.cert()).await?;
+            services.wait_for_services_ready().await?;
+            let ohttp_keys = services.fetch_ohttp_keys().await?;
+            let ohttp_keys_path = temp_dir.path().join("ohttp_keys");
+            tokio::fs::write(&ohttp_keys_path, ohttp_keys.encode()?).await?;
+
+            let receiver_rpchost = format!("http://{}/wallet/receiver", bitcoind.params.rpc_socket);
+            let cookie_file = &bitcoind.params.cookie_file;
+            let payjoin_cli = env!("CARGO_BIN_EXE_payjoin-cli");
+            let directory = &services.directory_url();
+            let ohttp_relay = &services.ohttp_relay_url();
+
+            // Start a receiver and capture its BIP21 so a session is persisted,
+            // then leave it parked at Initialized waiting for a proposal.
+            let cli_receiver = Command::new(payjoin_cli)
+                .arg("--root-certificate")
+                .arg(cert_path)
+                .arg("--rpchost")
+                .arg(&receiver_rpchost)
+                .arg("--cookie-file")
+                .arg(cookie_file)
+                .arg("--db-path")
+                .arg(&receiver_db_path)
+                .arg("--ohttp-relays")
+                .arg(ohttp_relay)
+                .arg("receive")
+                .arg(RECEIVE_SATS)
+                .arg("--pj-directory")
+                .arg(directory)
+                .arg("--ohttp-keys")
+                .arg(&ohttp_keys_path)
+                .stdout(Stdio::piped())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .expect("Failed to execute payjoin-cli receiver");
+            let _bip21 = get_bip21_from_receiver(cli_receiver).await;
+
+            // There is only one receiver session in progress.
+            let session_id = 1i64;
+
+            // Run `payjoin-cli cancel <session-id> --role receiver`: the session is at
+            // Initialized so there is no fallback transaction to broadcast.
+            let mut cli_cancel = Command::new(payjoin_cli)
+                .arg("--root-certificate")
+                .arg(cert_path)
+                .arg("--rpchost")
+                .arg(&receiver_rpchost)
+                .arg("--cookie-file")
+                .arg(cookie_file)
+                .arg("--db-path")
+                .arg(&receiver_db_path)
+                .arg("--ohttp-relays")
+                .arg(ohttp_relay)
+                .arg("cancel")
+                .arg(session_id.to_string())
+                .arg("--role")
+                .arg("receiver")
+                .stdout(Stdio::piped())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .expect("Failed to execute payjoin-cli cancel");
+
+            let mut cancel_stdout =
+                cli_cancel.stdout.take().expect("failed to take stdout of cancel");
+            let timeout = tokio::time::Duration::from_secs(10);
+            let cancel_line = tokio::time::timeout(
+                timeout,
+                wait_for_stdout_match(&mut cancel_stdout, |l| {
+                    l.contains("No fallback transaction to broadcast")
+                }),
+            )
+            .await?;
+            terminate(cli_cancel).await.expect("Failed to kill payjoin-cli cancel");
+            let subcommand_output =
+                cancel_line.expect("cancel should report no fallback transaction");
+
+            assert!(
+                subcommand_output.contains(&format!("Session {session_id} cancelled")),
+                "cancel should reference the cancelled session id"
+            );
+
+            Ok(())
+        }
+
+        Ok(())
+    }
 }