Persist payjoin receive sessions temporarily

Payjoin sessions poll while the wallet is active until they expire.

Author: DanGould

mutiny-core/src/lib.rs

@@ -49,6 +49,7 @@ pub use crate::gossip::{GOSSIP_SYNC_TIME_KEY, NETWORK_GRAPH_KEY, PROB_SCORER_KEY
 pub use crate::keymanager::generate_seed;
 pub use crate::ldkstorage::{CHANNEL_CLOSURE_PREFIX, CHANNEL_MANAGER_KEY, MONITORS_PREFIX_KEY};
 use crate::nostr::primal::{PrimalApi, PrimalClient};
+use crate::payjoin::PayjoinStorage;
 use crate::storage::{
     get_payment_hash_from_key, list_payment_info, persist_payment_info, update_nostr_contact_list,
     IndexItem, MutinyStorage, DEVICE_ID_KEY, EXPECTED_NETWORK_KEY, NEED_FULL_SYNC_KEY,
@@ -1188,6 +1189,7 @@ impl<S: MutinyStorage> MutinyWallet<S> {
         // when we restart, gen a new session id
         self.node_manager = Arc::new(nm_builder.build().await?);
         NodeManager::start_sync(self.node_manager.clone());
+        NodeManager::resume_payjoins(self.node_manager.clone());
 
         Ok(())
     }
@@ -1577,8 +1579,12 @@ impl<S: MutinyStorage> MutinyWallet<S> {
 
         let (pj, ohttp) = match self.node_manager.start_payjoin_session().await {
             Ok((enrolled, ohttp_keys)) => {
-                let pj_uri = enrolled.fallback_target();
-                self.node_manager.spawn_payjoin_receiver(enrolled);
+                let session = self
+                    .node_manager
+                    .storage
+                    .store_new_recv_session(enrolled.clone())?;
+                let pj_uri = session.enrolled.fallback_target();
+                self.node_manager.spawn_payjoin_receiver(session);
                 let ohttp = base64::encode_config(
                     ohttp_keys
                         .encode()

mutiny-core/src/nodemanager.rs

@@ -2,7 +2,7 @@ use crate::auth::MutinyAuthClient;
 use crate::labels::LabelStorage;
 use crate::ldkstorage::CHANNEL_CLOSURE_PREFIX;
 use crate::logging::LOGGING_KEY;
-use crate::payjoin::Error as PayjoinError;
+use crate::payjoin::{Error as PayjoinError, PayjoinStorage, RecvSession};
 use crate::utils::{sleep, spawn};
 use crate::MutinyInvoice;
 use crate::MutinyWalletConfig;
@@ -578,6 +578,14 @@ impl<S: MutinyStorage> NodeManager<S> {
         Ok(())
     }
 
+    /// Starts a background task to poll payjoin sessions to attempt receiving.
+    pub(crate) fn resume_payjoins(nm: Arc<NodeManager<S>>) {
+        let all = nm.storage.list_recv_sessions().unwrap_or_default();
+        for payjoin in all {
+            nm.clone().spawn_payjoin_receiver(payjoin);
+        }
+    }
+
     /// Creates a background process that will sync the wallet with the blockchain.
     /// This will also update the fee estimates every 10 minutes.
     pub fn start_sync(nm: Arc<NodeManager<S>>) {
@@ -769,12 +777,13 @@ impl<S: MutinyStorage> NodeManager<S> {
         Ok(txid)
     }
 
-    pub fn spawn_payjoin_receiver(&self, enrolled: Enrolled) {
+    pub fn spawn_payjoin_receiver(&self, session: RecvSession) {
         let logger = self.logger.clone();
         let stop = self.stop.clone();
+        let storage = Arc::new(self.storage.clone());
         let wallet = self.wallet.clone();
         utils::spawn(async move {
-            match Self::receive_payjoin(wallet, stop, enrolled).await {
+            match Self::receive_payjoin(wallet, stop, storage, session).await {
                 Ok(txid) => log_info!(logger, "Received payjoin txid: {txid}"),
                 Err(e) => log_error!(logger, "Error receiving payjoin: {e}"),
             };
@@ -785,13 +794,14 @@ impl<S: MutinyStorage> NodeManager<S> {
     async fn receive_payjoin(
         wallet: Arc<OnChainWallet<S>>,
         stop: Arc<AtomicBool>,
-        mut enrolled: payjoin::receive::v2::Enrolled,
+        storage: Arc<S>,
+        mut session: crate::payjoin::RecvSession,
     ) -> Result<Txid, MutinyError> {
         let http_client = reqwest::Client::builder()
             .build()
             .map_err(PayjoinError::Reqwest)?;
         let proposal: payjoin::receive::v2::UncheckedProposal =
-            Self::poll_for_fallback_psbt(stop, &http_client, &mut enrolled).await?;
+            Self::poll_for_fallback_psbt(stop, storage, &http_client, &mut session).await?;
         let original_tx = proposal.extract_tx_to_schedule_broadcast();
         let mut payjoin_proposal = match wallet.process_payjoin_proposal(proposal) {
             Ok(p) => p,
@@ -821,22 +831,30 @@ impl<S: MutinyStorage> NodeManager<S> {
 
     async fn poll_for_fallback_psbt(
         stop: Arc<AtomicBool>,
+        storage: Arc<S>,
         client: &reqwest::Client,
-        enroller: &mut payjoin::receive::v2::Enrolled,
+        session: &mut crate::payjoin::RecvSession,
     ) -> Result<payjoin::receive::v2::UncheckedProposal, PayjoinError> {
         loop {
             if stop.load(Ordering::Relaxed) {
                 return Err(crate::payjoin::Error::Shutdown);
             }
-            let (req, context) = enroller.extract_req()?;
+
+            if session.expiry < utils::now() {
+                let _ = storage.delete_recv_session(&session.enrolled.pubkey());
+                return Err(crate::payjoin::Error::SessionExpired);
+            }
+            let (req, context) = session.enrolled.extract_req()?;
             let ohttp_response = client
                 .post(req.url)
                 .header("Content-Type", "message/ohttp-req")
                 .body(req.body)
                 .send()
                 .await?;
             let ohttp_response = ohttp_response.bytes().await?;
-            let proposal = enroller.process_res(ohttp_response.as_ref(), context)?;
+            let proposal = session
+                .enrolled
+                .process_res(ohttp_response.as_ref(), context)?;
             match proposal {
                 Some(proposal) => return Ok(proposal),
                 None => utils::sleep(5000).await,

mutiny-core/src/payjoin.rs

@@ -1,5 +1,13 @@
+use std::collections::HashMap;
+
+use crate::error::MutinyError;
+use crate::storage::MutinyStorage;
+use core::time::Duration;
+use hex_conservative::DisplayHex;
 use once_cell::sync::Lazy;
+use payjoin::receive::v2::Enrolled;
 use payjoin::OhttpKeys;
+use serde::{Deserialize, Serialize};
 use url::Url;
 
 pub(crate) static OHTTP_RELAYS: [Lazy<Url>; 2] = [
@@ -17,6 +25,50 @@ pub fn random_ohttp_relay() -> &'static Url {
 pub(crate) static PAYJOIN_DIR: Lazy<Url> =
     Lazy::new(|| Url::parse("https://payjo.in").expect("Invalid URL"));
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct RecvSession {
+    pub enrolled: Enrolled,
+    pub expiry: Duration,
+}
+
+impl RecvSession {
+    pub fn pubkey(&self) -> [u8; 33] {
+        self.enrolled.pubkey()
+    }
+}
+pub trait PayjoinStorage {
+    fn list_recv_sessions(&self) -> Result<Vec<RecvSession>, MutinyError>;
+    fn store_new_recv_session(&self, session: Enrolled) -> Result<RecvSession, MutinyError>;
+    fn delete_recv_session(&self, id: &[u8; 33]) -> Result<(), MutinyError>;
+}
+
+const PAYJOIN_KEY_PREFIX: &str = "recvpj/";
+
+fn get_payjoin_key(id: &[u8; 33]) -> String {
+    format!("{PAYJOIN_KEY_PREFIX}{}", id.as_hex())
+}
+
+impl<S: MutinyStorage> PayjoinStorage for S {
+    fn list_recv_sessions(&self) -> Result<Vec<RecvSession>, MutinyError> {
+        let map: HashMap<String, RecvSession> = self.scan(PAYJOIN_KEY_PREFIX, None)?;
+        Ok(map.values().map(|v| v.to_owned()).collect())
+    }
+
+    fn store_new_recv_session(&self, enrolled: Enrolled) -> Result<RecvSession, MutinyError> {
+        let in_24_hours = crate::utils::now() + Duration::from_secs(60 * 60 * 24);
+        let session = RecvSession {
+            enrolled,
+            expiry: in_24_hours,
+        };
+        self.set_data(get_payjoin_key(&session.pubkey()), session.clone(), None)
+            .map(|_| session)
+    }
+
+    fn delete_recv_session(&self, id: &[u8; 33]) -> Result<(), MutinyError> {
+        self.delete(&[get_payjoin_key(id)])
+    }
+}
+
 pub async fn fetch_ohttp_keys(directory: Url) -> Result<OhttpKeys, Error> {
     let http_client = reqwest::Client::builder().build()?;
 
@@ -36,6 +88,7 @@ pub enum Error {
     Txid(bitcoin::hashes::hex::Error),
     OhttpDecodeFailed,
     Shutdown,
+    SessionExpired,
 }
 
 impl std::error::Error for Error {}
@@ -48,6 +101,7 @@ impl std::fmt::Display for Error {
             Error::Txid(e) => write!(f, "Payjoin txid error: {}", e),
             Error::OhttpDecodeFailed => write!(f, "Failed to decode ohttp keys"),
             Error::Shutdown => write!(f, "Payjoin stopped by application shutdown"),
+            Error::SessionExpired => write!(f, "Payjoin session expired. Create a new payment request and have the sender try again."),
         }
     }
 }