15568116171

Committed 10 Jun 2025 07:07PM UTC coverage: 85.714% (+0.2%) from 85.474%

Build # 15568116171

Build Type

Pull #750

github

Committed by

web-flow

Commit Message

Merge 06a46f9e7 into 89331880f

Pull Request Pull Request #750: Replace `Persister` with `SessionPersister` for `v2::Receiver`

Run Details

840 of 962 new or added lines in 10 files covered. (87.32%)

5 existing lines in 3 files now uncovered.

7596 of 8862 relevant lines covered (85.71%)

525.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.33

/payjoin-cli/src/db/v2.rs

use std::sync::Arc;
use std::time::SystemTime;

use bitcoincore_rpc::jsonrpc::serde_json;
use payjoin::bitcoin::hex::DisplayHex;
use payjoin::persist::{Persister, SessionPersister, Value};
use payjoin::receive::v2::ReceiverSessionEvent;
use payjoin::send::v2::{Sender, SenderToken, WithReplyKey};
use serde::{Deserialize, Serialize};
use sled::Tree;
use url::Url;

use super::*;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct SessionWrapper<V> {
    pub(crate) completed_at: Option<SystemTime>,
    pub(crate) events: Vec<V>,
}

#[derive(Debug, Clone)]
pub struct SessionId([u8; 8]);

impl SessionId {
    pub fn new(id: u64) -> Self { Self(id.to_be_bytes()) }
}

impl AsRef<[u8]> for SessionId {
    fn as_ref(&self) -> &[u8] { self.0.as_ref() }
}

pub(crate) struct SenderPersister(Arc<Database>);
impl SenderPersister {
    pub fn new(db: Arc<Database>) -> Self { Self(db) }
}

impl Persister<Sender<WithReplyKey>> for SenderPersister {
    type Token = SenderToken;
    type Error = crate::db::error::Error;
    fn save(
        &mut self,
        value: Sender<WithReplyKey>,
    ) -> std::result::Result<SenderToken, Self::Error> {
        let send_tree = self.0 .0.open_tree("send_sessions")?;
        let key = value.key();
        let value = serde_json::to_vec(&value).map_err(Error::Serialize)?;
        send_tree.insert(key.clone(), value.as_slice())?;
        send_tree.flush()?;
        Ok(key)
    }

    fn load(&self, key: SenderToken) -> std::result::Result<Sender<WithReplyKey>, Self::Error> {
        let send_tree = self.0 .0.open_tree("send_sessions")?;
        let value = send_tree.get(key.as_ref())?.ok_or(Error::NotFound(key.to_string()))?;
        serde_json::from_slice(&value).map_err(Error::Deserialize)
    }
}

#[derive(Clone)]
pub(crate) struct ReceiverPersister {
    db: Arc<Database>,
    session_id: SessionId,
}
impl ReceiverPersister {
    pub fn new(db: Arc<Database>) -> crate::db::Result<Self> {
        let id = SessionId::new(db.0.generate_id()?);
        let recv_tree = db.0.open_tree("recv_sessions")?;
        let empty_session: SessionWrapper<ReceiverSessionEvent> =
            SessionWrapper { completed_at: None, events: vec![] };
        let value = serde_json::to_vec(&empty_session).map_err(Error::Serialize)?;
        recv_tree.insert(id.as_ref(), value.as_slice())?;
        recv_tree.flush()?;

        Ok(Self { db: db.clone(), session_id: id })
    }

    pub fn from_id(db: Arc<Database>, id: SessionId) -> crate::db::Result<Self> {
        Ok(Self { db: db.clone(), session_id: id })
    }
}

impl SessionPersister for ReceiverPersister {
    type SessionEvent = ReceiverSessionEvent;
    type InternalStorageError = crate::db::error::Error;

    fn save_event(
        &self,
        event: &ReceiverSessionEvent,
    ) -> std::result::Result<(), Self::InternalStorageError> {
        let recv_tree = self.db.0.open_tree("recv_sessions")?;
        let key = self.session_id.as_ref();
        let session =
            recv_tree.get(key)?.ok_or(Error::NotFound(key.to_vec().to_lower_hex_string()))?;
        let mut session_wrapper: SessionWrapper<ReceiverSessionEvent> =
            serde_json::from_slice(&session).map_err(Error::Deserialize)?;
        session_wrapper.events.push(event.clone());
        let value = serde_json::to_vec(&session_wrapper).map_err(Error::Serialize)?;
        recv_tree.insert(key, value.as_slice())?;
        recv_tree.flush()?;
        Ok(())
    }

    fn load(
        &self,
    ) -> std::result::Result<
        Box<dyn Iterator<Item = ReceiverSessionEvent>>,
        Self::InternalStorageError,
    > {
        let recv_tree = self.db.0.open_tree("recv_sessions")?;
        let session_wrapper = recv_tree.get(self.session_id.as_ref())?;
        let value = session_wrapper.expect("key should exist");
        let wrapper: SessionWrapper<ReceiverSessionEvent> =
            serde_json::from_slice(&value).map_err(Error::Deserialize)?;
        Ok(Box::new(wrapper.events.into_iter()))
    }

    fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {
        let recv_tree = self.db.0.open_tree("recv_sessions")?;
        let key = self.session_id.as_ref();
        if let Some(existing) = recv_tree.get(key)? {
            let mut wrapper: SessionWrapper<ReceiverSessionEvent> =
                serde_json::from_slice(&existing).map_err(Error::Deserialize)?;
            wrapper.completed_at = Some(SystemTime::now());
            let value = serde_json::to_vec(&wrapper).map_err(Error::Serialize)?;
            recv_tree.insert(key, value.as_slice())?;
        }
        recv_tree.flush()?;
        Ok(())
    }
}

impl Database {
    pub(crate) fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {
        let recv_tree = self.0.open_tree("recv_sessions")?;
        let mut session_ids = Vec::new();
        for item in recv_tree.iter() {
            let (key, _) = item?;
            session_ids.push(SessionId::new(u64::from_be_bytes(
                key.as_ref().try_into().map_err(Error::TryFromSlice)?,
            )));
        }
        Ok(session_ids)
    }

    pub(crate) fn get_send_sessions(&self) -> Result<Vec<Sender<WithReplyKey>>> {
        let send_tree: Tree = self.0.open_tree("send_sessions")?;
        let mut sessions = Vec::new();
        for item in send_tree.iter() {
            let (_, value) = item?;
            let session: Sender<WithReplyKey> =
                serde_json::from_slice(&value).map_err(Error::Deserialize)?;
            sessions.push(session);
        }
        Ok(sessions)
    }

    pub(crate) fn get_send_session(&self, pj_url: &Url) -> Result<Option<Sender<WithReplyKey>>> {
        let send_tree = self.0.open_tree("send_sessions")?;
        if let Some(val) = send_tree.get(pj_url.as_str())? {
            let session: Sender<WithReplyKey> =
                serde_json::from_slice(&val).map_err(Error::Deserialize)?;
            Ok(Some(session))
        } else {
            Ok(None)
        }
    }

    pub(crate) fn clear_send_session(&self, pj_url: &Url) -> Result<()> {
        let send_tree: Tree = self.0.open_tree("send_sessions")?;
        send_tree.remove(pj_url.as_str())?;
        send_tree.flush()?;
        Ok(())
    }
}

1	use std::sync::Arc;
2	use std::time::SystemTime;
3
4	use bitcoincore_rpc::jsonrpc::serde_json;
5	use payjoin::bitcoin::hex::DisplayHex;
6	use payjoin::persist::{Persister, SessionPersister, Value};
7	use payjoin::receive::v2::ReceiverSessionEvent;
8	use payjoin::send::v2::{Sender, SenderToken, WithReplyKey};
9	use serde::{Deserialize, Serialize};
10	use sled::Tree;
11	use url::Url;
12
13	use super::*;
14
15	#[derive(Debug, Clone, Serialize, Deserialize)]
16	pub(crate) struct SessionWrapper<V> {
17	pub(crate) completed_at: Option<SystemTime>,
18	pub(crate) events: Vec<V>,
19	}
20
21	#[derive(Debug, Clone)]
22	pub struct SessionId([u8; 8]);
23
24	impl SessionId {
25	pub fn new(id: u64) -> Self { Self(id.to_be_bytes()) }	2✔
26	}
27
28	impl AsRef<[u8]> for SessionId {
29	fn as_ref(&self) -> &[u8] { self.0.as_ref() }	13✔
30	}
31
32	pub(crate) struct SenderPersister(Arc<Database>);
33	impl SenderPersister {
34	pub fn new(db: Arc<Database>) -> Self { Self(db) }	1✔
35	}
36
37	impl Persister<Sender<WithReplyKey>> for SenderPersister {
38	type Token = SenderToken;
39	type Error = crate::db::error::Error;
40	fn save(	1✔
41	&mut self,	1✔
42	value: Sender<WithReplyKey>,	1✔
43	) -> std::result::Result<SenderToken, Self::Error> {	1✔
44	let send_tree = self.0 .0.open_tree("send_sessions")?;	1✔
45	let key = value.key();	1✔
46	let value = serde_json::to_vec(&value).map_err(Error::Serialize)?;	1✔
47	send_tree.insert(key.clone(), value.as_slice())?;	1✔
48	send_tree.flush()?;	1✔
49	Ok(key)	1✔
50	}	1✔
51
52	fn load(&self, key: SenderToken) -> std::result::Result<Sender<WithReplyKey>, Self::Error> {	1✔
53	let send_tree = self.0 .0.open_tree("send_sessions")?;	1✔
54	let value = send_tree.get(key.as_ref())?.ok_or(Error::NotFound(key.to_string()))?;	1✔
55	serde_json::from_slice(&value).map_err(Error::Deserialize)	1✔
56	}	1✔
57	}
58
59	#[derive(Clone)]
60	pub(crate) struct ReceiverPersister {
61	db: Arc<Database>,
62	session_id: SessionId,
63	}
64	impl ReceiverPersister {
65	pub fn new(db: Arc<Database>) -> crate::db::Result<Self> {	1✔
66	let id = SessionId::new(db.0.generate_id()?);	1✔
67	let recv_tree = db.0.open_tree("recv_sessions")?;	1✔
68	let empty_session: SessionWrapper<ReceiverSessionEvent> =	1✔
69	SessionWrapper { completed_at: None, events: vec![] };	1✔
70	let value = serde_json::to_vec(&empty_session).map_err(Error::Serialize)?;	1✔
71	recv_tree.insert(id.as_ref(), value.as_slice())?;	1✔
72	recv_tree.flush()?;	1✔
73
74	Ok(Self { db: db.clone(), session_id: id })	1✔
75	}	1✔
76
77	pub fn from_id(db: Arc<Database>, id: SessionId) -> crate::db::Result<Self> {	1✔
78	Ok(Self { db: db.clone(), session_id: id })	1✔
79	}	1✔
80	}
81
82	impl SessionPersister for ReceiverPersister {
83	type SessionEvent = ReceiverSessionEvent;
84	type InternalStorageError = crate::db::error::Error;
85
86	fn save_event(	9✔
87	&self,	9✔
88	event: &ReceiverSessionEvent,	9✔
89	) -> std::result::Result<(), Self::InternalStorageError> {	9✔
90	let recv_tree = self.db.0.open_tree("recv_sessions")?;	9✔
91	let key = self.session_id.as_ref();	9✔
92	let session =	9✔
93	recv_tree.get(key)?.ok_or(Error::NotFound(key.to_vec().to_lower_hex_string()))?;	9✔
94	let mut session_wrapper: SessionWrapper<ReceiverSessionEvent> =	9✔
95	serde_json::from_slice(&session).map_err(Error::Deserialize)?;	9✔
96	session_wrapper.events.push(event.clone());	9✔
97	let value = serde_json::to_vec(&session_wrapper).map_err(Error::Serialize)?;	9✔
98	recv_tree.insert(key, value.as_slice())?;	9✔
99	recv_tree.flush()?;	9✔
100	Ok(())	9✔
101	}	9✔
102
103	fn load(	2✔
104	&self,	2✔
105	) -> std::result::Result<	2✔
106	Box<dyn Iterator<Item = ReceiverSessionEvent>>,	2✔
107	Self::InternalStorageError,	2✔
108	> {	2✔
109	let recv_tree = self.db.0.open_tree("recv_sessions")?;	2✔
110	let session_wrapper = recv_tree.get(self.session_id.as_ref())?;	2✔
111	let value = session_wrapper.expect("key should exist");	2✔
112	let wrapper: SessionWrapper<ReceiverSessionEvent> =	2✔
113	serde_json::from_slice(&value).map_err(Error::Deserialize)?;	2✔
114	Ok(Box::new(wrapper.events.into_iter()))	2✔
115	}	2✔
116
117	fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {	1✔
118	let recv_tree = self.db.0.open_tree("recv_sessions")?;	1✔
119	let key = self.session_id.as_ref();	1✔
120	if let Some(existing) = recv_tree.get(key)? {	1✔
121	let mut wrapper: SessionWrapper<ReceiverSessionEvent> =	1✔
122	serde_json::from_slice(&existing).map_err(Error::Deserialize)?;	1✔
123	wrapper.completed_at = Some(SystemTime::now());	1✔
124	let value = serde_json::to_vec(&wrapper).map_err(Error::Serialize)?;	1✔
125	recv_tree.insert(key, value.as_slice())?;	1✔
NEW 126	}	×
127	recv_tree.flush()?;	1✔
128	Ok(())	1✔
129	}	1✔
130	}
131
132	impl Database {
133	pub(crate) fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {	1✔
134	let recv_tree = self.0.open_tree("recv_sessions")?;	1✔
135	let mut session_ids = Vec::new();	1✔
136	for item in recv_tree.iter() {	1✔
137	let (key, _) = item?;	1✔
138	session_ids.push(SessionId::new(u64::from_be_bytes(	1✔
139	key.as_ref().try_into().map_err(Error::TryFromSlice)?,	1✔
140	)));
141	}
142	Ok(session_ids)	1✔
143	}	1✔
144
145	pub(crate) fn get_send_sessions(&self) -> Result<Vec<Sender<WithReplyKey>>> {	1✔
146	let send_tree: Tree = self.0.open_tree("send_sessions")?;	1✔
147	let mut sessions = Vec::new();	1✔
148	for item in send_tree.iter() {	1✔
149	let (_, value) = item?;	×
150	let session: Sender<WithReplyKey> =	×
151	serde_json::from_slice(&value).map_err(Error::Deserialize)?;	×
152	sessions.push(session);	×
153	}
154	Ok(sessions)	1✔
155	}	1✔
156
157	pub(crate) fn get_send_session(&self, pj_url: &Url) -> Result<Option<Sender<WithReplyKey>>> {	2✔
158	let send_tree = self.0.open_tree("send_sessions")?;	2✔
159	if let Some(val) = send_tree.get(pj_url.as_str())? {	2✔
160	let session: Sender<WithReplyKey> =	1✔
161	serde_json::from_slice(&val).map_err(Error::Deserialize)?;	1✔
162	Ok(Some(session))	1✔
163	} else {
164	Ok(None)	1✔
165	}
166	}	2✔
167
168	pub(crate) fn clear_send_session(&self, pj_url: &Url) -> Result<()> {	1✔
169	let send_tree: Tree = self.0.open_tree("send_sessions")?;	1✔
170	send_tree.remove(pj_url.as_str())?;	1✔
171	send_tree.flush()?;	1✔
172	Ok(())	1✔
173	}	1✔
174	}

payjoin / rust-payjoin / 15568116171

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous