Update payjoin-cli to non-blocking receive interface

Migrate both v1 and v2 receiver flows in payjoin-cli from
the callback-based validation API to the two-phase
extract/apply non-blocking API.

Author: xstoicunicornx

payjoin-cli/src/app/v1.rs

@@ -13,7 +13,7 @@ use hyper_util::rt::TokioIo;
 use payjoin::bitcoin::consensus::encode::serialize_hex;
 use payjoin::bitcoin::{Amount, FeeRate};
 use payjoin::receive::v1::{PayjoinProposal, UncheckedOriginalPayload};
-use payjoin::receive::Error;
+use payjoin::receive::{check_references, Error};
 use payjoin::send::v1::SenderBuilder;
 use payjoin::{ImplementationError, IntoUrl, Uri, UriExt};
 use tokio::net::TcpListener;
@@ -347,33 +347,37 @@ impl App {
         let wallet = self.wallet();
 
         // Receive Check 1: Can Broadcast
-        let proposal = proposal.check_broadcast_suitability(None, |tx| {
-            wallet
-                .can_broadcast(tx)
-                .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
-        })?;
+        let is_broadcast_suitable = wallet
+            .can_broadcast(&proposal.extract_tx_to_check_broadcast_suitability())
+            .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))?;
+        let proposal = proposal.apply_broadcast_suitability(None, is_broadcast_suitable)?;
         tracing::trace!("check1");
 
         // in a payment processor where the sender could go offline, this is where you schedule to broadcast the original_tx
         let _to_broadcast_in_failure_case = proposal.extract_tx_to_schedule_broadcast();
 
         // Receive Check 2: receiver can't sign for proposal inputs
-        let proposal = proposal.check_inputs_not_owned(&mut |input| {
+        let refs = proposal.get_input_script_refs()?;
+        let checked_refs = check_references(refs, &mut |input| {
             wallet.is_mine(input).map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
         })?;
+        let proposal = proposal.apply_input_owned_checks(checked_refs)?;
         tracing::trace!("check2");
 
         // Receive Check 3: have we seen this input before? More of a check for non-interactive i.e. payment processor receivers.
-        let payjoin = proposal.check_no_inputs_seen_before(&mut |input| {
-            Ok(self.db.insert_input_seen_before(*input)?)
-        })?;
+        let refs = proposal.get_input_outpoint_refs();
+        let checked_refs =
+            check_references(refs, &mut |input| Ok(self.db.insert_input_seen_before(*input)?))?;
+        let payjoin = proposal.apply_input_seen_checks(checked_refs)?;
         tracing::trace!("check3");
 
-        let payjoin = payjoin.identify_receiver_outputs(&mut |output_script| {
+        let refs = payjoin.get_output_script_refs();
+        let checked_refs = check_references(refs, &mut |output_script| {
             wallet
                 .is_mine(output_script)
                 .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
         })?;
+        let payjoin = payjoin.apply_output_owned_checks(checked_refs)?;
 
         let payjoin = payjoin
             .substitute_receiver_script(

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

@@ -5,6 +5,7 @@ use anyhow::{anyhow, Context, Result};
 use payjoin::bitcoin::consensus::encode::serialize_hex;
 use payjoin::bitcoin::{Amount, FeeRate};
 use payjoin::persist::{OptionalTransitionOutcome, SessionPersister};
+use payjoin::receive::check_references;
 use payjoin::receive::v2::{
     replay_event_log as replay_receiver_event_log, HasReplyableError, Initialized,
     MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, PayjoinProposal,
@@ -702,13 +703,11 @@ impl App {
         persister: &ReceiverPersister,
     ) -> Result<()> {
         let wallet = self.wallet();
-        let proposal = proposal
-            .check_broadcast_suitability(None, |tx| {
-                wallet
-                    .can_broadcast(tx)
-                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
-            })
-            .save(persister)?;
+        let is_broadcast_suitable = wallet
+            .can_broadcast(&proposal.extract_tx_to_check_broadcast_suitability())
+            .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))?;
+        let proposal =
+            proposal.apply_broadcast_suitability(None, is_broadcast_suitable).save(persister)?;
 
         println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:");
         println!("{}", serialize_hex(&proposal.extract_tx_to_schedule_broadcast()));
@@ -721,13 +720,11 @@ impl App {
         persister: &ReceiverPersister,
     ) -> Result<()> {
         let wallet = self.wallet();
-        let proposal = proposal
-            .check_inputs_not_owned(&mut |input| {
-                wallet
-                    .is_mine(input)
-                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
-            })
-            .save(persister)?;
+        let refs = proposal.get_input_script_refs()?;
+        let checked_refs = check_references(refs, &mut |input| {
+            wallet.is_mine(input).map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
+        })?;
+        let proposal = proposal.apply_input_owned_checks(checked_refs).save(persister)?;
         self.check_no_inputs_seen_before(proposal, persister).await
     }
 
@@ -736,11 +733,10 @@ impl App {
         proposal: Receiver<MaybeInputsSeen>,
         persister: &ReceiverPersister,
     ) -> Result<()> {
-        let proposal = proposal
-            .check_no_inputs_seen_before(&mut |input| {
-                Ok(self.db.insert_input_seen_before(*input)?)
-            })
-            .save(persister)?;
+        let refs = proposal.get_input_outpoint_refs();
+        let checked_refs =
+            check_references(refs, &mut |input| Ok(self.db.insert_input_seen_before(*input)?))?;
+        let proposal = proposal.apply_input_seen_checks(checked_refs).save(persister)?;
         self.identify_receiver_outputs(proposal, persister).await
     }
 
@@ -750,13 +746,13 @@ impl App {
         persister: &ReceiverPersister,
     ) -> Result<()> {
         let wallet = self.wallet();
-        let proposal = proposal
-            .identify_receiver_outputs(&mut |output_script| {
-                wallet
-                    .is_mine(output_script)
-                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
-            })
-            .save(persister)?;
+        let refs = proposal.get_output_script_refs();
+        let checked_refs = check_references(refs, &mut |output_script| {
+            wallet
+                .is_mine(output_script)
+                .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
+        })?;
+        let proposal = proposal.apply_output_owned_checks(checked_refs).save(persister)?;
         self.commit_outputs(proposal, persister).await
     }
 
@@ -804,13 +800,11 @@ impl App {
         persister: &ReceiverPersister,
     ) -> Result<()> {
         let wallet = self.wallet();
-        let proposal = proposal
-            .finalize_proposal(|psbt| {
-                wallet
-                    .process_psbt(psbt)
-                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
-            })
-            .save(persister)?;
+        let psbt = proposal.psbt_to_sign();
+        let signed_psbt = wallet
+            .process_psbt(&psbt)
+            .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))?;
+        let proposal = proposal.finalize_signed_proposal(&signed_psbt).save(persister)?;
         self.send_payjoin_proposal(proposal, persister).await
     }
 
@@ -847,27 +841,32 @@ impl App {
 
         tracing::debug!("Polling for payment confirmation");
 
+        let fallback_txid = proposal.extract_fallback_txid();
+        let payjoin_txid = proposal.extract_payjoin_proposal_txid();
+        let get_raw_tx = |txid| {
+            self.wallet()
+                .get_raw_transaction(&txid)
+                .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
+        };
+        match proposal.check_fallback_monitorable().save(persister)? {
+            OptionalTransitionOutcome::Progress(_) => {
+                println!("Unable to monitor for fallback tx containing non-segwit inputs, completing session");
+                return Ok(());
+            }
+            OptionalTransitionOutcome::Stasis(_) => {}
+        }
         let result = tokio::time::timeout(timeout_duration, async {
             loop {
                 interval.tick().await;
-                let check_result = proposal
-                    .check_payment(|txid| {
-                        self.wallet()
-                            .get_raw_transaction(&txid)
-                            .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
-                    })
-                    .save(persister);
-
-                match check_result {
-                    Ok(_) => {
-                        println!("Payjoin transaction detected in the mempool!");
-                        return Ok(());
-                    }
-                    Err(_) => {
-                        // keep polling
-
-                        continue;
-                    }
+                if let Some(tx) = get_raw_tx(payjoin_txid)? {
+                    proposal.payjoin_tx_exists(tx).save(persister)?;
+                    println!("Payjoin transaction detected in the mempool!");
+                    return Ok(());
+                };
+                if get_raw_tx(fallback_txid)?.is_some() {
+                    proposal.fallback_tx_exists().save(persister)?;
+                    println!("Fallback transaction detected in the mempool!");
+                    return Ok(());
                 }
             }
         })