17672564771

Committed 12 Sep 2025 11:02AM UTC coverage: 84.452% (+84.5%) from 0.0%

Build # 17672564771

Build Type

Pull #22

github

Committed by

web-flow

Commit Message

Merge 52a9e4ef4 into 434582cfa

Pull Request Pull Request #22: Storage

Run Details

460 of 528 new or added lines in 9 files covered. (87.12%)

1 existing line in 1 file now uncovered.

1950 of 2309 relevant lines covered (84.45%)

6.17 hits per line

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

77.63

/databases/sqlite_db/src/sqlite.rs

// Copyright 2025 Kore Ledger, SL
// SPDX-License-Identifier: Apache-2.0

//! # SQLite database backend.
//!
//! This module contains the SQLite database backend implementation.
//!

//use crate::error::NodeError;

use store::{
    Error,
    database::{Collection, DbManager, State},
};

use rusqlite::{Connection, OpenFlags, Result as SQLiteResult, params};
use tracing::info;

use std::sync::{Arc, Mutex};
use std::{fs, path::Path};

/// SQLite database manager.
#[derive(Clone)]
pub struct SqliteManager {
    conn: Arc<Mutex<Connection>>,
}

impl SqliteManager {
    /// Create a new SQLite database manager.
    pub fn new(path: &str) -> Result<Self, Error> {
        info!("Creating SQLite database manager");
        if !Path::new(&path).exists() {
            info!("Path does not exist, creating it");
            fs::create_dir_all(path).map_err(|e| {
                Error::CreateStore(format!(
                    "fail SQLite create directory: {}",
                    e
                ))
            })?;
        }

        info!("Opening SQLite connection");
        let conn = open(format!("{}/database.db", path)).map_err(|e| {
            Error::CreateStore(format!("fail SQLite open connection: {}", e))
        })?;

        Ok(Self {
            conn: Arc::new(Mutex::new(conn)),
        })
    }
}

impl DbManager<SqliteCollection, SqliteCollection> for SqliteManager {
    fn create_state(
        &self,
        identifier: &str,
        prefix: &str,
    ) -> Result<SqliteCollection, Error> {
        // Create statement to create a table.
        let stmt = format!(
            "CREATE TABLE IF NOT EXISTS {} (prefix TEXT NOT NULL, value \
            BLOB NOT NULL, PRIMARY KEY (prefix))",
            identifier
        );

        {
            let conn = self.conn.lock().expect("open connection");

            conn.execute(stmt.as_str(), ()).map_err(|e| {
                Error::CreateStore(format!("fail SQLite create table: {}", e))
            })?;
        }

        Ok(SqliteCollection::new(self.conn.clone(), identifier, prefix))
    }

    fn create_collection(
        &self,
        identifier: &str,
        prefix: &str,
    ) -> Result<SqliteCollection, Error> {
        // Create statement to create a table.
        let stmt = format!(
            "CREATE TABLE IF NOT EXISTS {} (prefix TEXT NOT NULL, sn TEXT NOT NULL, value \
            BLOB NOT NULL, PRIMARY KEY (prefix, sn))",
            identifier
        );

        {
            let conn = self.conn.lock().expect("open connection");

            conn.execute(stmt.as_str(), ()).map_err(|e| {
                Error::CreateStore(format!("fail SQLite create table: {}", e))
            })?;
        }

        Ok(SqliteCollection::new(self.conn.clone(), identifier, prefix))
    }
}

/// SQLite collection
pub struct SqliteCollection {
    conn: Arc<Mutex<Connection>>,
    table: String,
    prefix: String,
}

impl SqliteCollection {
    /// Create a new SQLite collection.
    pub fn new(
        conn: Arc<Mutex<Connection>>,
        table: &str,
        prefix: &str,
    ) -> Self {
        Self {
            conn,
            table: table.to_owned(),
            prefix: prefix.to_owned(),
        }
    }

    /// Create a new iterartor filtering by prefix.
    fn make_iter<'a>(
        &'a self,
        reverse: bool,
    ) -> SQLiteResult<Box<dyn Iterator<Item = (String, Vec<u8>)> + 'a>> {
        let order = if reverse { "DESC" } else { "ASC" };
        let conn = self.conn.lock().expect("open connection");
        let query = format!(
            "SELECT sn, value FROM {} WHERE prefix = ?1 ORDER BY sn {}",
            self.table, order,
        );
        let mut stmt = conn.prepare(&query)?;
        let mut rows = stmt.query(params![self.prefix])?;
        let mut values = Vec::new();
        while let Some(row) = rows.next()? {
            let key: String = row.get(0)?;
            values.push((key, row.get(1)?));
        }
        Ok(Box::new(values.into_iter()))
    }
}

impl State for SqliteCollection {
    fn get(&self) -> Result<Vec<u8>, Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("sqlite open connection: {}", e))
        })?;
        let query =
            format!("SELECT value FROM {} WHERE prefix = ?1", &self.table);
        let row: Vec<u8> = conn
            .query_row(&query, params![self.prefix], |row| row.get(0))
            .map_err(|e| Error::EntryNotFound(e.to_string()))?;

        Ok(row)
    }

    fn put(&mut self, data: &[u8]) -> Result<(), Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("sqlite open connection: {}", e))
        })?;
        let stmt = format!(
            "INSERT OR REPLACE INTO {} (prefix, value) VALUES (?1, ?2)",
            &self.table
        );
        conn.execute(&stmt, params![self.prefix, data])
            .map_err(|e| Error::Store(format!("sqlite insert error: {}", e)))?;
        Ok(())
    }

    fn del(&mut self) -> Result<(), Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("SQLITE open connection: {}", e))
        })?;
        let stmt = format!("DELETE FROM {} WHERE prefix = ?1", &self.table);
        conn.execute(&stmt, params![self.prefix,])
            .map_err(|e| Error::EntryNotFound(e.to_string()))?;
        Ok(())
    }

    fn purge(&mut self) -> Result<(), Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("SQLITE open connection: {}", e))
        })?;
        let stmt = format!("DELETE FROM {} WHERE prefix = ?1", &self.table);
        conn.execute(&stmt, params![self.prefix])
            .map_err(|e| Error::Store(format!("SQLITE purge error: {}", e)))?;
        Ok(())
    }

    fn name(&self) -> &str {
        self.table.as_str()
    }
}

impl Collection for SqliteCollection {
    fn get(&self, key: &str) -> Result<Vec<u8>, Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("sqlite open connection: {}", e))
        })?;
        let query = format!(
            "SELECT value FROM {} WHERE prefix = ?1 AND sn = ?2",
            &self.table
        );
        let row: Vec<u8> = conn
            .query_row(&query, params![self.prefix, key], |row| row.get(0))
            .map_err(|e| Error::EntryNotFound(e.to_string()))?;

        Ok(row)
    }

    fn put(&mut self, key: &str, data: &[u8]) -> Result<(), Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("sqlite open connection: {}", e))
        })?;
        let stmt = format!(
            "INSERT OR REPLACE INTO {} (prefix, sn, value) VALUES (?1, ?2, ?3)",
            &self.table
        );
        conn.execute(&stmt, params![self.prefix, key, data])
            .map_err(|e| Error::Store(format!("sqlite insert error: {}", e)))?;
        Ok(())
    }

    fn del(&mut self, key: &str) -> Result<(), Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("SQLITE open connection: {}", e))
        })?;
        let stmt = format!(
            "DELETE FROM {} WHERE prefix = ?1 AND sn = ?2",
            &self.table
        );
        conn.execute(&stmt, params![self.prefix, key])
            .map_err(|e| Error::EntryNotFound(e.to_string()))?;
        Ok(())
    }

    fn purge(&mut self) -> Result<(), Error> {
        let conn = self.conn.lock().map_err(|e| {
            Error::Store(format!("SQLITE open connection: {}", e))
        })?;
        let stmt = format!("DELETE FROM {} WHERE prefix = ?1", &self.table);
        conn.execute(&stmt, params![self.prefix])
            .map_err(|e| Error::Store(format!("SQLITE purge error: {}", e)))?;
        Ok(())
    }

    fn iter<'a>(
        &'a self,
        reverse: bool,
    ) -> Box<dyn Iterator<Item = (String, Vec<u8>)> + 'a> {
        match self.make_iter(reverse) {
            Ok(iter) => {
                let iterator = SQLiteIterator { iter };
                Box::new(iterator)
            }
            Err(_) => Box::new(std::iter::empty()),
        }
    }

    fn name(&self) -> &str {
        self.table.as_str()
    }
}

pub struct SQLiteIterator<'a> {
    pub iter: Box<dyn Iterator<Item = (String, Vec<u8>)> + 'a>,
}

impl Iterator for SQLiteIterator<'_> {
    type Item = (String, Vec<u8>);

    fn next(&mut self) -> Option<Self::Item> {
        self.iter.next()
    }
}

/// Open a SQLite database connection.
pub fn open<P: AsRef<Path>>(path: P) -> Result<Connection, Error> {
    let path = path.as_ref();
    let flags =
        OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE;
    let conn = Connection::open_with_flags(path, flags).map_err(|e| {
        Error::Store(format!("SQLite failed to open connection: {}", e))
    })?;

    conn.execute_batch(
        "
        PRAGMA journal_mode=WAL;
        PRAGMA synchronous=NORMAL;
        ",
    )
    .map_err(|e| {
        Error::Store(format!("SQLite failed to execute batch: {}", e))
    })?;

    Ok(conn)
}

#[cfg(test)]
mod tests {
    pub fn create_temp_dir() -> String {
        let path = temp_dir();

        if fs::metadata(&path).is_err() {
            fs::create_dir_all(&path).unwrap();
        }
        path
    }

    fn temp_dir() -> String {
        let dir =
            tempfile::tempdir().expect("Can not create temporal directory.");
        dir.path().to_str().unwrap().to_owned()
    }

    impl Default for SqliteManager {
        fn default() -> Self {
            let path = format!("{}/database.db", create_temp_dir());
            let conn = open(&path)
                .map_err(|e| {
                    Error::CreateStore(format!(
                        "fail SQLite open connection: {}",
                        e
                    ))
                })
                .expect("Cannot open the database ");

            Self {
                conn: Arc::new(Mutex::new(conn)),
            }
        }
    }

    use super::*;
    use store::{
        database::{Collection, DbManager},
        test_store_trait,
    };

    test_store_trait! {
        unit_test_sqlite_manager:SqliteManager:SqliteCollection
    }
}

1	// Copyright 2025 Kore Ledger, SL
2	// SPDX-License-Identifier: Apache-2.0
3
4	//! # SQLite database backend.
5	//!
6	//! This module contains the SQLite database backend implementation.
7	//!
8
9	//use crate::error::NodeError;
10
11	use store::{
12	Error,
13	database::{Collection, DbManager, State},
14	};
15
16	use rusqlite::{Connection, OpenFlags, Result as SQLiteResult, params};
17	use tracing::info;
18
19	use std::sync::{Arc, Mutex};
20	use std::{fs, path::Path};
21
22	/// SQLite database manager.
23	#[derive(Clone)]
24	pub struct SqliteManager {
25	conn: Arc<Mutex<Connection>>,
26	}
27
28	impl SqliteManager {
29	/// Create a new SQLite database manager.
30	pub fn new(path: &str) -> Result<Self, Error> {	×
31	info!("Creating SQLite database manager");	×
32	if !Path::new(&path).exists() {	×
33	info!("Path does not exist, creating it");	×
34	fs::create_dir_all(path).map_err(\|e\| {	×
35	Error::CreateStore(format!(	×
36	"fail SQLite create directory: {}",	×
37	e	×
38	))	×
39	})?;	×
40	}	×
41
42	info!("Opening SQLite connection");	×
43	let conn = open(format!("{}/database.db", path)).map_err(\|e\| {	×
44	Error::CreateStore(format!("fail SQLite open connection: {}", e))	×
45	})?;	×
46
47	Ok(Self {	×
48	conn: Arc::new(Mutex::new(conn)),	×
49	})	×
50	}	×
51	}
52
53	impl DbManager<SqliteCollection, SqliteCollection> for SqliteManager {
54	fn create_state(	4✔
55	&self,	4✔
56	identifier: &str,	4✔
57	prefix: &str,	4✔
58	) -> Result<SqliteCollection, Error> {	4✔
59	// Create statement to create a table.
60	let stmt = format!(	4✔
61	"CREATE TABLE IF NOT EXISTS {} (prefix TEXT NOT NULL, value \	4✔
62	BLOB NOT NULL, PRIMARY KEY (prefix))",	4✔
63	identifier
64	);
65
66	{
67	let conn = self.conn.lock().expect("open connection");	4✔
68
69	conn.execute(stmt.as_str(), ()).map_err(\|e\| {	4✔
NEW 70	Error::CreateStore(format!("fail SQLite create table: {}", e))	×
NEW 71	})?;	×
72	}
73
74	Ok(SqliteCollection::new(self.conn.clone(), identifier, prefix))	4✔
75	}	4✔
76
77	fn create_collection(	8✔
78	&self,	8✔
79	identifier: &str,	8✔
80	prefix: &str,	8✔
81	) -> Result<SqliteCollection, Error> {	8✔
82	// Create statement to create a table.
83	let stmt = format!(	8✔
84	"CREATE TABLE IF NOT EXISTS {} (prefix TEXT NOT NULL, sn TEXT NOT NULL, value \	8✔
85	BLOB NOT NULL, PRIMARY KEY (prefix, sn))",	8✔
86	identifier
87	);
88
89	{
90	let conn = self.conn.lock().expect("open connection");	8✔
91
92	conn.execute(stmt.as_str(), ()).map_err(\|e\| {	8✔
93	Error::CreateStore(format!("fail SQLite create table: {}", e))	×
94	})?;	×
95	}
96
97	Ok(SqliteCollection::new(self.conn.clone(), identifier, prefix))	8✔
98	}	8✔
99	}
100
101	/// SQLite collection
102	pub struct SqliteCollection {
103	conn: Arc<Mutex<Connection>>,
104	table: String,
105	prefix: String,
106	}
107
108	impl SqliteCollection {
109	/// Create a new SQLite collection.
110	pub fn new(	12✔
111	conn: Arc<Mutex<Connection>>,	12✔
112	table: &str,	12✔
113	prefix: &str,	12✔
114	) -> Self {	12✔
115	Self {	12✔
116	conn,	12✔
117	table: table.to_owned(),	12✔
118	prefix: prefix.to_owned(),	12✔
119	}	12✔
120	}	12✔
121
122	/// Create a new iterartor filtering by prefix.
123	fn make_iter<'a>(	5✔
124	&'a self,	5✔
125	reverse: bool,	5✔
126	) -> SQLiteResult<Box<dyn Iterator<Item = (String, Vec<u8>)> + 'a>> {	5✔
127	let order = if reverse { "DESC" } else { "ASC" };	5✔
128	let conn = self.conn.lock().expect("open connection");	5✔
129	let query = format!(	5✔
130	"SELECT sn, value FROM {} WHERE prefix = ?1 ORDER BY sn {}",	5✔
131	self.table, order,
132	);
133	let mut stmt = conn.prepare(&query)?;	5✔
134	let mut rows = stmt.query(params![self.prefix])?;	5✔
135	let mut values = Vec::new();	5✔
136	while let Some(row) = rows.next()? {	20✔
137	let key: String = row.get(0)?;	15✔
138	values.push((key, row.get(1)?));	15✔
139	}
140	Ok(Box::new(values.into_iter()))	5✔
141	}	5✔
142	}
143
144	impl State for SqliteCollection {
145	fn get(&self) -> Result<Vec<u8>, Error> {	6✔
146	let conn = self.conn.lock().map_err(\|e\| {	6✔
NEW 147	Error::Store(format!("sqlite open connection: {}", e))	×
NEW 148	})?;	×
149	let query =	6✔
150	format!("SELECT value FROM {} WHERE prefix = ?1", &self.table);	6✔
151	let row: Vec<u8> = conn	6✔
152	.query_row(&query, params![self.prefix], \|row\| row.get(0))	6✔
153	.map_err(\|e\| Error::EntryNotFound(e.to_string()))?;	6✔
154
155	Ok(row)	3✔
156	}	6✔
157
158	fn put(&mut self, data: &[u8]) -> Result<(), Error> {	4✔
159	let conn = self.conn.lock().map_err(\|e\| {	4✔
NEW 160	Error::Store(format!("sqlite open connection: {}", e))	×
NEW 161	})?;	×
162	let stmt = format!(	4✔
163	"INSERT OR REPLACE INTO {} (prefix, value) VALUES (?1, ?2)",	4✔
164	&self.table	4✔
165	);
166	conn.execute(&stmt, params![self.prefix, data])	4✔
167	.map_err(\|e\| Error::Store(format!("sqlite insert error: {}", e)))?;	4✔
168	Ok(())	4✔
169	}	4✔
170
171	fn del(&mut self) -> Result<(), Error> {	1✔
172	let conn = self.conn.lock().map_err(\|e\| {	1✔
NEW 173	Error::Store(format!("SQLITE open connection: {}", e))	×
NEW 174	})?;	×
175	let stmt = format!("DELETE FROM {} WHERE prefix = ?1", &self.table);	1✔
176	conn.execute(&stmt, params![self.prefix,])	1✔
177	.map_err(\|e\| Error::EntryNotFound(e.to_string()))?;	1✔
178	Ok(())	1✔
179	}	1✔
180
181	fn purge(&mut self) -> Result<(), Error> {	2✔
182	let conn = self.conn.lock().map_err(\|e\| {	2✔
NEW 183	Error::Store(format!("SQLITE open connection: {}", e))	×
NEW 184	})?;	×
185	let stmt = format!("DELETE FROM {} WHERE prefix = ?1", &self.table);	2✔
186	conn.execute(&stmt, params![self.prefix])	2✔
187	.map_err(\|e\| Error::Store(format!("SQLITE purge error: {}", e)))?;	2✔
188	Ok(())	2✔
189	}	2✔
190
191	fn name(&self) -> &str {	1✔
192	self.table.as_str()	1✔
193	}	1✔
194	}
195
196	impl Collection for SqliteCollection {
197	fn get(&self, key: &str) -> Result<Vec<u8>, Error> {	8✔
198	let conn = self.conn.lock().map_err(\|e\| {	8✔
199	Error::Store(format!("sqlite open connection: {}", e))	×
200	})?;	×
201	let query = format!(	8✔
202	"SELECT value FROM {} WHERE prefix = ?1 AND sn = ?2",	8✔
203	&self.table	8✔
204	);
205	let row: Vec<u8> = conn	8✔
206	.query_row(&query, params![self.prefix, key], \|row\| row.get(0))	8✔
207	.map_err(\|e\| Error::EntryNotFound(e.to_string()))?;	8✔
208
209	Ok(row)	4✔
210	}	8✔
211
212	fn put(&mut self, key: &str, data: &[u8]) -> Result<(), Error> {	17✔
213	let conn = self.conn.lock().map_err(\|e\| {	17✔
214	Error::Store(format!("sqlite open connection: {}", e))	×
215	})?;	×
216	let stmt = format!(	17✔
217	"INSERT OR REPLACE INTO {} (prefix, sn, value) VALUES (?1, ?2, ?3)",	17✔
218	&self.table	17✔
219	);
220	conn.execute(&stmt, params![self.prefix, key, data])	17✔
221	.map_err(\|e\| Error::Store(format!("sqlite insert error: {}", e)))?;	17✔
222	Ok(())	17✔
223	}	17✔
224
225	fn del(&mut self, key: &str) -> Result<(), Error> {	1✔
226	let conn = self.conn.lock().map_err(\|e\| {	1✔
227	Error::Store(format!("SQLITE open connection: {}", e))	×
228	})?;	×
229	let stmt = format!(	1✔
230	"DELETE FROM {} WHERE prefix = ?1 AND sn = ?2",	1✔
231	&self.table	1✔
232	);
233	conn.execute(&stmt, params![self.prefix, key])	1✔
234	.map_err(\|e\| Error::EntryNotFound(e.to_string()))?;	1✔
235	Ok(())	1✔
236	}	1✔
237
238	fn purge(&mut self) -> Result<(), Error> {	1✔
239	let conn = self.conn.lock().map_err(\|e\| {	1✔
240	Error::Store(format!("SQLITE open connection: {}", e))	×
241	})?;	×
242	let stmt = format!("DELETE FROM {} WHERE prefix = ?1", &self.table);	1✔
243	conn.execute(&stmt, params![self.prefix])	1✔
244	.map_err(\|e\| Error::Store(format!("SQLITE purge error: {}", e)))?;	1✔
245	Ok(())	1✔
246	}	1✔
247
248	fn iter<'a>(	5✔
249	&'a self,	5✔
250	reverse: bool,	5✔
251	) -> Box<dyn Iterator<Item = (String, Vec<u8>)> + 'a> {	5✔
252	match self.make_iter(reverse) {	5✔
253	Ok(iter) => {	5✔
254	let iterator = SQLiteIterator { iter };	5✔
255	Box::new(iterator)	5✔
256	}
257	Err(_) => Box::new(std::iter::empty()),	×
258	}
259	}	5✔
260
261	fn name(&self) -> &str {	1✔
262	self.table.as_str()	1✔
263	}	1✔
264	}
265
266	pub struct SQLiteIterator<'a> {
267	pub iter: Box<dyn Iterator<Item = (String, Vec<u8>)> + 'a>,
268	}
269
270	impl Iterator for SQLiteIterator<'_> {
271	type Item = (String, Vec<u8>);
272
273	fn next(&mut self) -> Option<Self::Item> {	14✔
274	self.iter.next()	14✔
275	}	14✔
276	}
277
278	/// Open a SQLite database connection.
279	pub fn open<P: AsRef<Path>>(path: P) -> Result<Connection, Error> {	12✔
280	let path = path.as_ref();	12✔
281	let flags =	12✔
282	OpenFlags::SQLITE_OPEN_READ_WRITE \| OpenFlags::SQLITE_OPEN_CREATE;	12✔
283	let conn = Connection::open_with_flags(path, flags).map_err(\|e\| {	12✔
284	Error::Store(format!("SQLite failed to open connection: {}", e))	×
285	})?;	×
286
287	conn.execute_batch(	12✔
288	"	12✔
289	PRAGMA journal_mode=WAL;	12✔
290	PRAGMA synchronous=NORMAL;	12✔
291	",	12✔
292	)
293	.map_err(\|e\| {	12✔
294	Error::Store(format!("SQLite failed to execute batch: {}", e))	×
295	})?;	×
296
297	Ok(conn)	12✔
298	}	12✔
299
300	#[cfg(test)]
301	mod tests {
302	pub fn create_temp_dir() -> String {	12✔
303	let path = temp_dir();	12✔
304
305	if fs::metadata(&path).is_err() {	12✔
306	fs::create_dir_all(&path).unwrap();	12✔
307	}	12✔
308	path	12✔
309	}	12✔
310
311	fn temp_dir() -> String {	12✔
312	let dir =	12✔
313	tempfile::tempdir().expect("Can not create temporal directory.");	12✔
314	dir.path().to_str().unwrap().to_owned()	12✔
315	}	12✔
316
317	impl Default for SqliteManager {
318	fn default() -> Self {	12✔
319	let path = format!("{}/database.db", create_temp_dir());	12✔
320	let conn = open(&path)	12✔
321	.map_err(\|e\| {	12✔
322	Error::CreateStore(format!(	×
323	"fail SQLite open connection: {}",	×
324	e	×
325	))	×
326	})	×
327	.expect("Cannot open the database ");	12✔
328
329	Self {	12✔
330	conn: Arc::new(Mutex::new(conn)),	12✔
331	}	12✔
332	}	12✔
333	}
334
335	use super::*;
336	use store::{
337	database::{Collection, DbManager},
338	test_store_trait,
339	};
340
341	test_store_trait! {
342	unit_test_sqlite_manager:SqliteManager:SqliteCollection
343	}
344	}

kore-ledger / rush-rs / 17672564771

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