pub mod device_events; pub mod devices; pub mod error; pub mod notifications; use include_dir::{Dir, include_dir}; use lazy_static::lazy_static; use log::{debug, error}; use r2d2::PooledConnection; use r2d2_sqlite::SqliteConnectionManager; use rusqlite_migration::Migrations; use tokio::sync::Mutex; use crate::{db::error::DbError, settings::get_settings}; static MIGRATIONS_DIR: Dir = include_dir!("$CARGO_MANIFEST_DIR/database_migrations"); static INITIALISED: Mutex = Mutex::const_new(false); lazy_static! { // Define migrations. These are applied atomically. static ref MIGRATIONS: Migrations<'static> = Migrations::from_directory(&MIGRATIONS_DIR).unwrap(); // TODO : Move pool size to configuration file static ref POOL: r2d2::Pool = r2d2::Pool::builder(). max_size(10). build( r2d2_sqlite::SqliteConnectionManager::file(get_settings().database.path.as_str()) .with_init(|conn| { // Tune every pooled connection for the concurrent access pattern of this // app (five scanners + web server + retention sharing the pool): // - WAL lets readers and a writer proceed concurrently instead of blocking. // - synchronous=NORMAL is the safe, recommended pairing with WAL and avoids // an fsync on every commit (the default FULL fsyncs on each write). // - busy_timeout makes a connection wait for a lock rather than failing // immediately with "database is locked". // - foreign_keys are off by default in SQLite and must be set per connection. conn.execute_batch( "PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL; PRAGMA busy_timeout = 5000; PRAGMA foreign_keys = ON;", ) }) ).unwrap(); } pub fn get_db_connection() -> Result, DbError> { POOL.get().map_err(|error| { error!("Error obtaining database connection from the pool: {error}"); DbError::from(error) }) } // Runs a blocking database operation on tokio's blocking thread pool so it never stalls an async // worker thread. The DB layer uses synchronous `rusqlite`, so axum handlers must wrap their DB // work in this rather than calling `db::*` functions inline. pub async fn run_blocking(f: F) -> T where F: FnOnce() -> T + Send + 'static, T: Send + 'static, { tokio::task::spawn_blocking(f) .await .expect("blocking database task panicked") } // Appends the shared `LIMIT ? OFFSET ?` paging clause (and its bound parameters) to a list query // when both an offset and a limit are supplied. Used by the list endpoints (devices, notifications, // device_events) so they page identically. pub fn apply_paging( sql: &mut String, params: &mut Vec, page_offset: Option, page_limit: Option, ) { if let (Some(page_offset), Some(page_limit)) = (page_offset, page_limit) { debug!( "Adding paging to list with offset={} and limit={}", page_offset, page_limit ); sql.push_str(" LIMIT ? OFFSET ?"); params.push(page_limit.into()); params.push(page_offset.into()); } } pub async fn init_db() -> Result<(), DbError> { let mut initialised = INITIALISED.lock().await; if *initialised { return Ok(()); } debug!("Getting database connection"); let mut conn = get_db_connection()?; debug!("Executing database migrations if needed."); let result = match MIGRATIONS.to_latest(&mut conn) { Ok(_) => { debug!("Database up to date."); Ok(()) } Err(error) => { error!("Error updating database: {error}"); Err(DbError::from(error)) } }; *initialised = true; result }