17080593272

Committed 19 Aug 2025 08:08PM UTC coverage: 86.412% (-0.1%) from 86.514%

Build # 17080593272

Build Type

Pull #873

github

Committed by

web-flow

Commit Message

Merge 39f114476 into 10f1a31d5

Pull Request Pull Request #873: Migrate payjoin-cli from sled to rusqlite

Run Details

154 of 179 new or added lines in 4 files covered. (86.03%)

1 existing line in 1 file now uncovered.

7911 of 9155 relevant lines covered (86.41%)

509.35 hits per line

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

97.22

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

use std::sync::Arc;

use payjoin::persist::SessionPersister;
use payjoin::receive::v2::SessionEvent as ReceiverSessionEvent;
use payjoin::send::v2::SessionEvent as SenderSessionEvent;
use rusqlite::params;

use super::*;

#[derive(Debug, Clone)]
pub enum SessionId {
    Send(i64),
    Receive(i64),
}

impl SessionId {
    pub fn as_integer(&self) -> i64 {
        match self {
            SessionId::Send(id) => *id,
            SessionId::Receive(id) => *id,
        }
    }
}

#[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 conn = db.get_connection()?;

        // Create a new session in send_sessions and get its ID
        let session_id: i64 = conn.query_row(
            "INSERT INTO send_sessions (session_id) VALUES (NULL) RETURNING session_id",
            [],
            |row| row.get(0),
        )?;

        Ok(Self { db, session_id: SessionId::Send(session_id) })
    }

    pub fn from_id(db: Arc<Database>, id: SessionId) -> crate::db::Result<Self> {
        match id {
            SessionId::Send(_) => Ok(Self { db, session_id: id }),
            SessionId::Receive(_) =>
                panic!("Attempted to create SenderPersister with Receive session 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 conn = self.db.get_connection()?;
        let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;

        conn.execute(
            "INSERT INTO send_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",
            params![self.session_id.as_integer(), event_data, now()],
        )?;

        Ok(())
    }

    fn load(
        &self,
    ) -> std::result::Result<Box<dyn Iterator<Item = SenderSessionEvent>>, Self::InternalStorageError>
    {
        let conn = self.db.get_connection()?;
        let mut stmt = conn.prepare(
            "SELECT event_data FROM send_session_events WHERE session_id = ?1 ORDER BY created_at ASC",
        )?;

        let event_rows = stmt.query_map(params![self.session_id.as_integer()], |row| {
            let event_data: String = row.get(0)?;
            Ok(event_data)
        })?;

        let events: Vec<SenderSessionEvent> = event_rows
            .map(|row| {
                let event_data = row.expect("Failed to read event data from database");
                serde_json::from_str::<SenderSessionEvent>(&event_data)
                    .expect("Database corruption: failed to deserialize session event")
            })
            .collect();

        Ok(Box::new(events.into_iter()))
    }

    fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {
        let conn = self.db.get_connection()?;

        conn.execute(
            "UPDATE send_sessions SET completed_at = ?1 WHERE session_id = ?2",
            params![now(), self.session_id.as_integer()],
        )?;

        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 conn = db.get_connection()?;

        // Create a new session in receive_sessions and get its ID
        let session_id: i64 = conn.query_row(
            "INSERT INTO receive_sessions (session_id) VALUES (NULL) RETURNING session_id",
            [],
            |row| row.get(0),
        )?;

        Ok(Self { db, session_id: SessionId::Receive(session_id) })
    }

    pub fn from_id(db: Arc<Database>, id: SessionId) -> crate::db::Result<Self> {
        match id {
            SessionId::Receive(_) => Ok(Self { db, session_id: id }),
            SessionId::Send(_) =>
                panic!("Attempted to create ReceiverPersister with Send session 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 conn = self.db.get_connection()?;
        let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;

        conn.execute(
            "INSERT INTO receive_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",
            params![self.session_id.as_integer(), event_data, now()],
        )?;

        Ok(())
    }

    fn load(
        &self,
    ) -> std::result::Result<
        Box<dyn Iterator<Item = ReceiverSessionEvent>>,
        Self::InternalStorageError,
    > {
        let conn = self.db.get_connection()?;
        let mut stmt = conn.prepare(
            "SELECT event_data FROM receive_session_events WHERE session_id = ?1 ORDER BY created_at ASC",
        )?;

        let event_rows = stmt.query_map(params![self.session_id.as_integer()], |row| {
            let event_data: String = row.get(0)?;
            Ok(event_data)
        })?;

        let events: Vec<ReceiverSessionEvent> = event_rows
            .map(|row| {
                let event_data = row.expect("Failed to read event data from database");
                serde_json::from_str::<ReceiverSessionEvent>(&event_data)
                    .expect("Database corruption: failed to deserialize session event")
            })
            .collect();

        Ok(Box::new(events.into_iter()))
    }

    fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {
        let conn = self.db.get_connection()?;

        conn.execute(
            "UPDATE receive_sessions SET completed_at = ?1 WHERE session_id = ?2",
            params![now(), self.session_id.as_integer()],
        )?;

        Ok(())
    }
}

impl Database {
    pub(crate) fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {
        let conn = self.get_connection()?;
        let mut stmt =
            conn.prepare("SELECT session_id FROM receive_sessions WHERE completed_at IS NULL")?;

        let session_rows = stmt.query_map([], |row| {
            let session_id: i64 = row.get(0)?;
            Ok(SessionId::Receive(session_id))
        })?;

        let mut session_ids = Vec::new();
        for session_row in session_rows {
            let session_id = session_row?;
            session_ids.push(session_id);
        }

        Ok(session_ids)
    }

    pub(crate) fn get_send_session_ids(&self) -> Result<Vec<SessionId>> {
        let conn = self.get_connection()?;
        let mut stmt =
            conn.prepare("SELECT session_id FROM send_sessions WHERE completed_at IS NULL")?;

        let session_rows = stmt.query_map([], |row| {
            let session_id: i64 = row.get(0)?;
            Ok(SessionId::Send(session_id))
        })?;

        let mut session_ids = Vec::new();
        for session_row in session_rows {
            let session_id = session_row?;
            session_ids.push(session_id);
        }

        Ok(session_ids)
    }
}

1	use std::sync::Arc;
2
3	use payjoin::persist::SessionPersister;
4	use payjoin::receive::v2::SessionEvent as ReceiverSessionEvent;
5	use payjoin::send::v2::SessionEvent as SenderSessionEvent;
6	use rusqlite::params;
7
8	use super::*;
9
10	#[derive(Debug, Clone)]
11	pub enum SessionId {
12	Send(i64),
13	Receive(i64),
14	}
15
16	impl SessionId {
17	pub fn as_integer(&self) -> i64 {	18✔
18	match self {	18✔
19	SessionId::Send(id) => *id,	5✔
20	SessionId::Receive(id) => *id,	13✔
21	}
22	}	18✔
23	}
24
25	#[derive(Clone)]
26	pub(crate) struct SenderPersister {
27	db: Arc<Database>,
28	session_id: SessionId,
29	}
30
31	impl SenderPersister {
32	pub fn new(db: Arc<Database>) -> crate::db::Result<Self> {	1✔
33	let conn = db.get_connection()?;	1✔
34
35	// Create a new session in send_sessions and get its ID
36	let session_id: i64 = conn.query_row(	1✔
37	"INSERT INTO send_sessions (session_id) VALUES (NULL) RETURNING session_id",	1✔
38	[],	1✔
39	\|row\| row.get(0),	1✔
NEW 40	)?;	×
41
42	Ok(Self { db, session_id: SessionId::Send(session_id) })	1✔
43	}	1✔
44
45	pub fn from_id(db: Arc<Database>, id: SessionId) -> crate::db::Result<Self> {	1✔
46	match id {	1✔
47	SessionId::Send(_) => Ok(Self { db, session_id: id }),	1✔
48	SessionId::Receive(_) =>
NEW 49	panic!("Attempted to create SenderPersister with Receive session ID"),	×
50	}
51	}	1✔
52	}
53
54	impl SessionPersister for SenderPersister {
55	type SessionEvent = SenderSessionEvent;
56	type InternalStorageError = crate::db::error::Error;
57
58	fn save_event(	3✔
59	&self,	3✔
60	event: SenderSessionEvent,	3✔
61	) -> std::result::Result<(), Self::InternalStorageError> {	3✔
62	let conn = self.db.get_connection()?;	3✔
63	let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;	3✔
64
65	conn.execute(	3✔
66	"INSERT INTO send_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",	3✔
67	params![self.session_id.as_integer(), event_data, now()],	3✔
68	)?;	3✔
69
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	{
77	let conn = self.db.get_connection()?;	1✔
78	let mut stmt = conn.prepare(	1✔
79	"SELECT event_data FROM send_session_events WHERE session_id = ?1 ORDER BY created_at ASC",	1✔
80	)?;	1✔
81
82	let event_rows = stmt.query_map(params![self.session_id.as_integer()], \|row\| {	2✔
83	let event_data: String = row.get(0)?;	2✔
84	Ok(event_data)	2✔
85	})?;	2✔
86
87	let events: Vec<SenderSessionEvent> = event_rows	1✔
88	.map(\|row\| {	2✔
89	let event_data = row.expect("Failed to read event data from database");	2✔
90	serde_json::from_str::<SenderSessionEvent>(&event_data)	2✔
91	.expect("Database corruption: failed to deserialize session event")	2✔
92	})	2✔
93	.collect();	1✔
94
95	Ok(Box::new(events.into_iter()))	1✔
96	}	1✔
97
98	fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {	1✔
99	let conn = self.db.get_connection()?;	1✔
100
101	conn.execute(	1✔
102	"UPDATE send_sessions SET completed_at = ?1 WHERE session_id = ?2",	1✔
103	params![now(), self.session_id.as_integer()],	1✔
104	)?;	1✔
105
106	Ok(())	1✔
107	}	1✔
108	}
109
110	#[derive(Clone)]
111	pub(crate) struct ReceiverPersister {
112	db: Arc<Database>,
113	session_id: SessionId,
114	}
115
116	impl ReceiverPersister {
117	pub fn new(db: Arc<Database>) -> crate::db::Result<Self> {	1✔
118	let conn = db.get_connection()?;	1✔
119
120	// Create a new session in receive_sessions and get its ID
121	let session_id: i64 = conn.query_row(	1✔
122	"INSERT INTO receive_sessions (session_id) VALUES (NULL) RETURNING session_id",	1✔
123	[],	1✔
124	\|row\| row.get(0),	1✔
NEW 125	)?;	×
126
127	Ok(Self { db, session_id: SessionId::Receive(session_id) })	1✔
128	}	1✔
129
130	pub fn from_id(db: Arc<Database>, id: SessionId) -> crate::db::Result<Self> {	1✔
131	match id {	1✔
132	SessionId::Receive(_) => Ok(Self { db, session_id: id }),	1✔
133	SessionId::Send(_) =>
NEW 134	panic!("Attempted to create ReceiverPersister with Send session ID"),	×
135	}
136	}	1✔
137	}
138
139	impl SessionPersister for ReceiverPersister {
140	type SessionEvent = ReceiverSessionEvent;
141	type InternalStorageError = crate::db::error::Error;
142
143	fn save_event(	10✔
144	&self,	10✔
145	event: ReceiverSessionEvent,	10✔
146	) -> std::result::Result<(), Self::InternalStorageError> {	10✔
147	let conn = self.db.get_connection()?;	10✔
148	let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;	10✔
149
150	conn.execute(	10✔
151	"INSERT INTO receive_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",	10✔
152	params![self.session_id.as_integer(), event_data, now()],	10✔
153	)?;	10✔
154
155	Ok(())	10✔
156	}	10✔
157
158	fn load(	2✔
159	&self,	2✔
160	) -> std::result::Result<	2✔
161	Box<dyn Iterator<Item = ReceiverSessionEvent>>,	2✔
162	Self::InternalStorageError,	2✔
163	> {	2✔
164	let conn = self.db.get_connection()?;	2✔
165	let mut stmt = conn.prepare(	2✔
166	"SELECT event_data FROM receive_session_events WHERE session_id = ?1 ORDER BY created_at ASC",	2✔
167	)?;	2✔
168
169	let event_rows = stmt.query_map(params![self.session_id.as_integer()], \|row\| {	2✔
170	let event_data: String = row.get(0)?;	2✔
171	Ok(event_data)	2✔
172	})?;	2✔
173
174	let events: Vec<ReceiverSessionEvent> = event_rows	2✔
175	.map(\|row\| {	2✔
176	let event_data = row.expect("Failed to read event data from database");	2✔
177	serde_json::from_str::<ReceiverSessionEvent>(&event_data)	2✔
178	.expect("Database corruption: failed to deserialize session event")	2✔
179	})	2✔
180	.collect();	2✔
181
182	Ok(Box::new(events.into_iter()))	2✔
183	}	2✔
184
185	fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {	1✔
186	let conn = self.db.get_connection()?;	1✔
187
188	conn.execute(	1✔
189	"UPDATE receive_sessions SET completed_at = ?1 WHERE session_id = ?2",	1✔
190	params![now(), self.session_id.as_integer()],	1✔
191	)?;	1✔
192
193	Ok(())	1✔
194	}	1✔
195	}
196
197	impl Database {
198	pub(crate) fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {	3✔
199	let conn = self.get_connection()?;	3✔
200	let mut stmt =	3✔
201	conn.prepare("SELECT session_id FROM receive_sessions WHERE completed_at IS NULL")?;	3✔
202
203	let session_rows = stmt.query_map([], \|row\| {	3✔
204	let session_id: i64 = row.get(0)?;	1✔
205	Ok(SessionId::Receive(session_id))	1✔
206	})?;	1✔
207
208	let mut session_ids = Vec::new();	3✔
209	for session_row in session_rows {	4✔
210	let session_id = session_row?;	1✔
211	session_ids.push(session_id);	1✔
212	}
213
214	Ok(session_ids)	3✔
215	}	3✔
216
217	pub(crate) fn get_send_session_ids(&self) -> Result<Vec<SessionId>> {	5✔
218	let conn = self.get_connection()?;	5✔
219	let mut stmt =	5✔
220	conn.prepare("SELECT session_id FROM send_sessions WHERE completed_at IS NULL")?;	5✔
221
222	let session_rows = stmt.query_map([], \|row\| {	5✔
223	let session_id: i64 = row.get(0)?;	1✔
224	Ok(SessionId::Send(session_id))	1✔
225	})?;	1✔
226
227	let mut session_ids = Vec::new();	5✔
228	for session_row in session_rows {	6✔
229	let session_id = session_row?;	1✔
230	session_ids.push(session_id);	1✔
231	}
232
233	Ok(session_ids)	5✔
234	}	5✔
235	}

payjoin / rust-payjoin / 17080593272

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