16473108548

Committed 23 Jul 2025 02:11PM UTC coverage: 85.622% (-0.08%) from 85.706%

Build # 16473108548

Build Type

Pull #873

github

Committed by

web-flow

Commit Message

Merge cbb0e6d26 into e4badaa21

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

Coverage Stats

155 of 179 new or added lines in 4 files covered. (86.59%)

1 existing line in 1 file now uncovered.

7801 of 9111 relevant lines covered (85.62%)

514.1 hits per line

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

97.45

/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::*;

macro_rules! now {
    () => {
        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,
        }
    }

    pub fn session_type(&self) -> &'static str {
        match self {
            SessionId::Send(_) => "send",
            SessionId::Receive(_) => "receive",
        }
    }
}

#[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 (?1)",
            params![Option::<i64>::None],
        )?;

        // 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 session_events (session_id, session_type, event_data, created_at) VALUES (?1, ?2, ?3, ?4)",
            params![self.session_id.as_integer(), self.session_id.session_type(), 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 session_events WHERE session_id = ?1 AND session_type = ?2 ORDER BY created_at ASC",
        )?;

        let event_rows = stmt.query_map(
            params![self.session_id.as_integer(), self.session_id.session_type()],
            |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 (?1)",
            params![Option::<i64>::None],
        )?;

        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 session_events (session_id, session_type, event_data, created_at) VALUES (?1, ?2, ?3, ?4)",
            params![self.session_id.as_integer(), self.session_id.session_type(), 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 session_events WHERE session_id = ?1 AND session_type = ?2 ORDER BY created_at ASC",
        )?;

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

payjoin / rust-payjoin / 16473108548

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