16533244097

Committed 25 Jul 2025 11:11PM UTC coverage: 85.705% (-0.08%) from 85.783%

Build # 16533244097

Build Type

Pull #873

github

Committed by

web-flow

Commit Message

Merge c6a55118a into 5ecfed745

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

Coverage Stats

152 of 175 new or added lines in 4 files covered. (86.86%)

1 existing line in 1 file now uncovered.

7920 of 9241 relevant lines covered (85.71%)

507.16 hits per line

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

98.6

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

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

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

use super::*;

#[inline]
fn now() -> i64 {
    SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as i64
}

#[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 table
        conn.execute("INSERT INTO send_sessions (completed_at) VALUES (NULL)", [])?;

        // Get the generated session ID
        let session_id = conn.last_insert_rowid();

        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)?;
        let timestamp = now();

        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, timestamp],
        )?;

        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 mut events = Vec::new();
        for event_row in event_rows {
            let event_data = event_row?;
            let event: SenderSessionEvent =
                serde_json::from_str(&event_data).map_err(Error::Deserialize)?;
            events.push(event);
        }

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

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

        conn.execute(
            "UPDATE send_sessions SET completed_at = ?1 WHERE session_id = ?2",
            params![timestamp, 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()?;

        conn.execute("INSERT INTO receive_sessions (completed_at) VALUES (NULL)", [])?;

        let session_id = conn.last_insert_rowid();

        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)?;
        let timestamp = now();

        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, timestamp],
        )?;

        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 mut events = Vec::new();
        for event_row in event_rows {
            let event_data = event_row?;
            let event: ReceiverSessionEvent =
                serde_json::from_str(&event_data).map_err(Error::Deserialize)?;
            events.push(event);
        }

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

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

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

payjoin / rust-payjoin / 16533244097

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