Persist receiver when initialized

This commit introduces a persistence flow for constructing
`v2::Receiver`. We enforce that receivers must be persisted before
initiating their typestate. To accommodate this, callers
will first create a `NewReceiver`, which must be persisted.
Upon persistence, they receive a storage token that can be
passed to `v2::Receiver::load(...)` to construct a `v2::Receiver`.

For cases where persistence is unnecessary (e.g. in testing),
implementers can use the `NoopPersister` available in
`persist/mod.rs`.

Issue: payjoin#336

Author: arminsabouri

payjoin-cli/src/app/v2.rs

@@ -4,7 +4,7 @@ use anyhow::{anyhow, Context, Result};
 use payjoin::bitcoin::consensus::encode::serialize_hex;
 use payjoin::bitcoin::psbt::Psbt;
 use payjoin::bitcoin::{Amount, FeeRate};
-use payjoin::receive::v2::{Receiver, UncheckedProposal};
+use payjoin::receive::v2::{NewReceiver, Receiver, UncheckedProposal};
 use payjoin::receive::{Error, ImplementationError, ReplyableError};
 use payjoin::send::v2::{Sender, SenderBuilder};
 use payjoin::Uri;
@@ -14,6 +14,7 @@ use super::config::Config;
 use super::wallet::BitcoindWallet;
 use super::App as AppTrait;
 use crate::app::{handle_interrupt, http_agent};
+use crate::db::v2::ReceiverPersister;
 use crate::db::Database;
 
 #[derive(Clone)]
@@ -65,13 +66,18 @@ impl AppTrait for App {
     async fn receive_payjoin(&self, amount: Amount) -> Result<()> {
         let address = self.wallet().get_new_address()?;
         let ohttp_keys = unwrap_ohttp_keys_or_else_fetch(&self.config).await?;
-        let session = Receiver::new(
+        let mut persister = ReceiverPersister::new(self.db.clone());
+        let new_receiver = NewReceiver::new(
             address,
             self.config.v2()?.pj_directory.clone(),
             ohttp_keys.clone(),
             None,
         )?;
-        self.db.insert_recv_session(session.clone())?;
+        let storage_token = new_receiver
+            .persist(&mut persister)
+            .map_err(|e| anyhow!("Failed to persist receiver: {}", e))?;
+        let session = Receiver::load(storage_token, &persister)
+            .map_err(|e| anyhow!("Failed to load receiver: {}", e))?;
         self.spawn_payjoin_receiver(session, Some(amount)).await
     }
 

payjoin-cli/src/db/error.rs

@@ -13,6 +13,8 @@ pub(crate) enum Error {
     Serialize(serde_json::Error),
     #[cfg(feature = "v2")]
     Deserialize(serde_json::Error),
+    #[cfg(feature = "v2")]
+    NotFound(String),
 }
 
 impl fmt::Display for Error {
@@ -23,6 +25,8 @@ impl fmt::Display for Error {
             Error::Serialize(e) => write!(f, "Serialization failed: {}", e),
             #[cfg(feature = "v2")]
             Error::Deserialize(e) => write!(f, "Deserialization failed: {}", e),
+            #[cfg(feature = "v2")]
+            Error::NotFound(key) => write!(f, "Key not found: {}", key),
         }
     }
 }

payjoin-cli/src/db/mod.rs

@@ -27,4 +27,4 @@ impl Database {
 }
 
 #[cfg(feature = "v2")]
-mod v2;
+pub(crate) mod v2;

payjoin-cli/src/db/v2.rs

@@ -1,21 +1,38 @@
+use std::sync::Arc;
+
 use bitcoincore_rpc::jsonrpc::serde_json;
-use payjoin::receive::v2::Receiver;
+use payjoin::persist::{Persister, Value};
+use payjoin::receive::v2::{Receiver, ReceiverToken};
 use payjoin::send::v2::Sender;
-use sled::{IVec, Tree};
+use sled::Tree;
 use url::Url;
 
 use super::*;
 
-impl Database {
-    pub(crate) fn insert_recv_session(&self, session: Receiver) -> Result<()> {
-        let recv_tree = self.0.open_tree("recv_sessions")?;
-        let key = &session.id();
-        let value = serde_json::to_string(&session).map_err(Error::Serialize)?;
-        recv_tree.insert(key.as_slice(), IVec::from(value.as_str()))?;
+pub(crate) struct ReceiverPersister(Arc<Database>);
+impl ReceiverPersister {
+    pub fn new(db: Arc<Database>) -> Self { Self(db) }
+}
+
+impl Persister<Receiver> for ReceiverPersister {
+    type Token = ReceiverToken;
+    type Error = crate::db::error::Error;
+    fn save(&mut self, value: Receiver) -> std::result::Result<ReceiverToken, Self::Error> {
+        let recv_tree = self.0 .0.open_tree("recv_sessions")?;
+        let key = value.key();
+        let value = serde_json::to_vec(&value).map_err(Error::Serialize)?;
+        recv_tree.insert(key.clone(), value.as_slice())?;
         recv_tree.flush()?;
-        Ok(())
+        Ok(key)
+    }
+    fn load(&self, key: ReceiverToken) -> std::result::Result<Receiver, Self::Error> {
+        let recv_tree = self.0 .0.open_tree("recv_sessions")?;
+        let value = recv_tree.get(key.as_ref())?.ok_or(Error::NotFound(key.to_string()))?;
+        serde_json::from_slice(&value).map_err(Error::Deserialize)
     }
+}
 
+impl Database {
     pub(crate) fn get_recv_sessions(&self) -> Result<Vec<Receiver>> {
         let recv_tree = self.0.open_tree("recv_sessions")?;
         let mut sessions = Vec::new();

payjoin/src/receive/v2/mod.rs

@@ -1,4 +1,5 @@
 //! Receive BIP 77 Payjoin v2
+use std::fmt::{self, Display};
 use std::str::FromStr;
 use std::time::{Duration, SystemTime};
 
@@ -19,6 +20,7 @@ use super::{
 use crate::hpke::{decrypt_message_a, encrypt_message_b, HpkeKeyPair, HpkePublicKey};
 use crate::ohttp::{ohttp_decapsulate, ohttp_encapsulate, OhttpEncapsulationError, OhttpKeys};
 use crate::output_substitution::OutputSubstitution;
+use crate::persist::{self, Persister};
 use crate::receive::{parse_payload, InputPair};
 use crate::uri::ShortId;
 use crate::{IntoUrl, IntoUrlError, Request};
@@ -71,12 +73,12 @@ fn subdir_path_from_pubkey(pubkey: &HpkePublicKey) -> ShortId {
 
 /// A payjoin V2 receiver, allowing for polled requests to the
 /// payjoin directory and response processing.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct Receiver {
+#[derive(Debug)]
+pub struct NewReceiver {
     context: SessionContext,
 }
 
-impl Receiver {
+impl NewReceiver {
     /// Creates a new `Receiver` with the provided parameters.
     ///
     /// # Parameters
@@ -96,7 +98,7 @@ impl Receiver {
         ohttp_keys: OhttpKeys,
         expire_after: Option<Duration>,
     ) -> Result<Self, IntoUrlError> {
-        Ok(Self {
+        let receiver = Self {
             context: SessionContext {
                 address,
                 directory: directory.into_url()?,
@@ -107,9 +109,53 @@ impl Receiver {
                 s: HpkeKeyPair::gen_keypair(),
                 e: None,
             },
-        })
+        };
+        Ok(receiver)
+    }
+
+    pub fn persist<P: Persister<Receiver>>(
+        &self,
+        persister: &mut P,
+    ) -> Result<P::Token, ImplementationError> {
+        let receiver = Receiver { context: self.context.clone() };
+        Ok(persister.save(receiver)?)
     }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Receiver {
+    context: SessionContext,
+}
 
+/// Opaque key type for the receiver
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ReceiverToken(ShortId);
+
+impl Display for ReceiverToken {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "{}", self.0) }
+}
+
+impl From<Receiver> for ReceiverToken {
+    fn from(receiver: Receiver) -> Self { ReceiverToken(id(&receiver.context.s)) }
+}
+
+impl AsRef<[u8]> for ReceiverToken {
+    fn as_ref(&self) -> &[u8] { self.0.as_bytes() }
+}
+
+impl persist::Value for Receiver {
+    type Key = ReceiverToken;
+
+    fn key(&self) -> Self::Key { ReceiverToken(id(&self.context.s)) }
+}
+
+impl Receiver {
+    pub fn load<P: Persister<Receiver>>(
+        token: P::Token,
+        persister: &P,
+    ) -> Result<Self, ImplementationError> {
+        persister.load(token).map_err(ImplementationError::from)
+    }
     /// Extract an OHTTP Encapsulated HTTP GET request for the Original PSBT
     pub fn extract_req(
         &mut self,

payjoin/tests/integration.rs

@@ -171,7 +171,8 @@ mod integration {
 
         use bitcoin::Address;
         use http::StatusCode;
-        use payjoin::receive::v2::{PayjoinProposal, Receiver, UncheckedProposal};
+        use payjoin::persist::NoopPersister;
+        use payjoin::receive::v2::{NewReceiver, PayjoinProposal, Receiver, UncheckedProposal};
         use payjoin::send::v2::SenderBuilder;
         use payjoin::{OhttpKeys, PjUri, UriExt};
         use payjoin_test_utils::{BoxSendSyncError, TestServices};
@@ -205,8 +206,9 @@ mod integration {
                 let ohttp_relay = services.ohttp_relay_url();
                 let mock_address = Address::from_str("tb1q6d3a2w975yny0asuvd9a67ner4nks58ff0q8g4")?
                     .assume_checked();
-                let mut bad_initializer =
-                    Receiver::new(mock_address, directory, bad_ohttp_keys, None)?;
+                let new_receiver = NewReceiver::new(mock_address, directory, bad_ohttp_keys, None)?;
+                let storage_token = new_receiver.persist(&mut NoopPersister)?;
+                let mut bad_initializer = Receiver::load(storage_token, &NoopPersister)?;
                 let (req, _ctx) = bad_initializer.extract_req(&ohttp_relay)?;
                 agent
                     .post(req.url)
@@ -242,12 +244,16 @@ mod integration {
                 // Inside the Receiver:
                 let address = receiver.get_new_address(None, None)?.assume_checked();
                 // test session with expiry in the past
-                let mut expired_receiver = Receiver::new(
+                let new_receiver = NewReceiver::new(
                     address.clone(),
                     directory.clone(),
                     ohttp_keys.clone(),
                     Some(Duration::from_secs(0)),
                 )?;
+                let storage_token =
+                    new_receiver.persist(&mut NoopPersister).map_err(|e| e.to_string())?;
+                let mut expired_receiver =
+                    Receiver::load(storage_token, &NoopPersister).map_err(|e| e.to_string())?;
                 match expired_receiver.extract_req(&ohttp_relay) {
                     // Internal error types are private, so check against a string
                     Err(err) => assert!(err.to_string().contains("expired")),
@@ -294,8 +300,12 @@ mod integration {
                 let address = receiver.get_new_address(None, None)?.assume_checked();
 
                 // test session with expiry in the future
+                let new_receiver =
+                    NewReceiver::new(address.clone(), directory.clone(), ohttp_keys.clone(), None)?;
+                let storage_token =
+                    new_receiver.persist(&mut NoopPersister).map_err(|e| e.to_string())?;
                 let mut session =
-                    Receiver::new(address.clone(), directory.clone(), ohttp_keys.clone(), None)?;
+                    Receiver::load(storage_token, &NoopPersister).map_err(|e| e.to_string())?;
                 println!("session: {:#?}", &session);
                 // Poll receive request
                 let ohttp_relay = services.ohttp_relay_url();
@@ -465,9 +475,12 @@ mod integration {
                 let directory = services.directory_url();
                 let ohttp_keys = services.fetch_ohttp_keys().await?;
                 let address = receiver.get_new_address(None, None)?.assume_checked();
-
+                let new_receiver =
+                    NewReceiver::new(address, directory.clone(), ohttp_keys.clone(), None)?;
+                let storage_token =
+                    new_receiver.persist(&mut NoopPersister).map_err(|e| e.to_string())?;
                 let mut session =
-                    Receiver::new(address, directory.clone(), ohttp_keys.clone(), None)?;
+                    Receiver::load(storage_token, &NoopPersister).map_err(|e| e.to_string())?;
 
                 // **********************
                 // Inside the V1 Sender:
@@ -690,7 +703,8 @@ mod integration {
     #[cfg(feature = "_multiparty")]
     mod multiparty {
         use bitcoin::ScriptBuf;
-        use payjoin::receive::v2::Receiver;
+        use payjoin::persist::NoopPersister;
+        use payjoin::receive::v2::{NewReceiver, Receiver};
         use payjoin::send::multiparty::{
             GetContext as MultiPartyGetContext, SenderBuilder as MultiPartySenderBuilder,
         };
@@ -738,12 +752,16 @@ mod integration {
                 // Senders will generate a sweep psbt and send PSBT to receiver subdir
                 for sender in senders.iter() {
                     let address = receiver.get_new_address(None, None)?.assume_checked();
-                    let receiver_session = Receiver::new(
+                    let new_receiver = NewReceiver::new(
                         address.clone(),
                         directory.clone(),
                         ohttp_keys.clone(),
                         None,
                     )?;
+                    let storage_token =
+                        new_receiver.persist(&mut NoopPersister).map_err(|e| e.to_string())?;
+                    let receiver_session =
+                        Receiver::load(storage_token, &NoopPersister).map_err(|e| e.to_string())?;
                     let pj_uri = receiver_session.pj_uri();
                     let psbt = build_sweep_psbt(sender, &pj_uri)?;
                     let sender_ctx = MultiPartySenderBuilder::new(psbt.clone(), pj_uri.clone())