15829549096

Committed 23 Jun 2025 04:16PM UTC coverage: 86.01% (-0.08%) from 86.087%

Build # 15829549096

Build Type

Pull #789

github

Committed by

web-flow

Commit Message

Merge 3b4c3b426 into 9f403f159

Pull Request Pull Request #789: Replace `Persister` with `SessionPersister` for v2 Sender

Run Details

384 of 458 new or added lines in 6 files covered. (83.84%)

1 existing line in 1 file now uncovered.

7814 of 9085 relevant lines covered (86.01%)

511.9 hits per line

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

98.41

/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::SessionPersister;
use payjoin::receive::v2::SessionEvent as ReceiverSessionEvent;
use payjoin::send::v2::SessionEvent as SenderSessionEvent;
use serde::{Deserialize, Serialize};

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() }
}

#[derive(Clone)]
pub(crate) struct SenderPersister {
    db: Arc<Database>,
    session_id: SessionId,
}
impl SenderPersister {
    pub fn new(db: Arc<Database>) -> crate::db::Result<Self> {
        let id = SessionId::new(db.0.generate_id().unwrap());
        let send_tree = db.0.open_tree("send_sessions")?;
        let empty_session: SessionWrapper<SenderSessionEvent> =
            SessionWrapper { completed_at: None, events: vec![] };
        let value = serde_json::to_vec(&empty_session).map_err(Error::Serialize)?;
        send_tree.insert(id.as_ref(), value.as_slice())?;
        send_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 SenderPersister {
    type SessionEvent = SenderSessionEvent;
    type InternalStorageError = crate::db::error::Error;
    fn save_event(
        &self,
        event: &SenderSessionEvent,
    ) -> std::result::Result<(), Self::InternalStorageError> {
        let send_tree = self.db.0.open_tree("send_sessions")?;
        let key = self.session_id.as_ref();
        let session = send_tree.get(key)?.expect("key should exist");
        let mut session_wrapper: SessionWrapper<SenderSessionEvent> =
            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)?;
        send_tree.insert(key, value.as_slice())?;

        send_tree.flush()?;
        Ok(())
    }

    fn load(
        &self,
    ) -> std::result::Result<Box<dyn Iterator<Item = SenderSessionEvent>>, Self::InternalStorageError>
    {
        let send_tree = self.db.0.open_tree("send_sessions")?;
        let session_wrapper = send_tree.get(self.session_id.as_ref())?;
        let value = session_wrapper.expect("key should exist");
        let wrapper: SessionWrapper<SenderSessionEvent> =
            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 send_tree = self.db.0.open_tree("send_sessions")?;
        let key = self.session_id.as_ref();
        if let Some(existing) = send_tree.get(key)? {
            let mut wrapper: SessionWrapper<SenderSessionEvent> =
                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)?;
            send_tree.insert(key, value.as_slice())?;
        }
        send_tree.flush()?;
        Ok(())
    }
}

#[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_session_ids(&self) -> Result<Vec<SessionId>> {
        let send_tree = self.0.open_tree("send_sessions")?;
        let mut session_ids = Vec::new();
        for item in send_tree.iter() {
            let (key, _) = item?;
            session_ids.push(SessionId::new(u64::from_be_bytes(key.as_ref().try_into().unwrap())));
        }
        Ok(session_ids)
    }
}

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

payjoin / rust-payjoin / 15829549096

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