diff --git a/.gitignore b/.gitignore index e5ef396..11de1d9 100644 --- a/.gitignore +++ b/.gitignore @@ -2,5 +2,7 @@ /backend/target /result oott.db +oott.db-wal +oott.db-shm oott.toml /.claude diff --git a/backend/src/db.rs b/backend/src/db.rs index 8293686..25d7dc6 100644 --- a/backend/src/db.rs +++ b/backend/src/db.rs @@ -26,19 +26,43 @@ lazy_static! { 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() -> PooledConnection { - let result = POOL.get(); +pub fn get_db_connection() -> Result, DbError> { + POOL.get().map_err(|error| { + error!("Error obtaining database connection from the pool: {error}"); + DbError::from(error) + }) +} - match result { - Ok(value) => value, - Err(error) => { - error!("Error obtaining database connection from the pool: {error}"); - panic!("Error obtaining database connection from the pool: {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 @@ -69,7 +93,7 @@ pub async fn init_db() -> Result<(), DbError> { debug!("Getting database connection"); - let mut conn = get_db_connection(); + let mut conn = get_db_connection()?; debug!("Executing database migrations if needed."); diff --git a/backend/src/db/device_events.rs b/backend/src/db/device_events.rs index 16048c9..1f6bce9 100644 --- a/backend/src/db/device_events.rs +++ b/backend/src/db/device_events.rs @@ -8,7 +8,7 @@ use crate::model::device_events::{DeviceEvent, DeviceEventScanner}; use crate::utils::network::normalize_mac; pub fn insert(event: DeviceEvent) -> Result { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = normalize_mac(&event.mac_address); match conn.execute( @@ -42,7 +42,7 @@ pub fn recent_duplicate_exists( scanner: &DeviceEventScanner, since: DateTime, ) -> Result { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = normalize_mac(mac_address); let count: i64 = conn.query_row( @@ -66,7 +66,7 @@ pub fn list( page_limit: Option, ) -> Result, DbError> { debug!("Listing device events"); - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mut sql_statement = "SELECT id, mac_address, created_on, event_type, ipv4_address, vendor, scanner FROM device_events WHERE 1=1" @@ -111,7 +111,7 @@ pub fn list( } pub fn purge_older_than(cutoff: DateTime) -> Result { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; match conn.execute( "DELETE FROM device_events WHERE created_on < ?1", @@ -130,7 +130,13 @@ pub fn purge_older_than(cutoff: DateTime) -> Result { #[cfg(test)] fn read(id: i64) -> Option { - let conn = db::get_db_connection(); + let conn = match db::get_db_connection() { + Ok(conn) => conn, + Err(error) => { + error!("Error obtaining database connection: {error}"); + return None; + } + }; let result: Result = conn.query_one( "SELECT id, mac_address, created_on, event_type, ipv4_address, vendor, scanner FROM device_events WHERE id=?1", diff --git a/backend/src/db/devices.rs b/backend/src/db/devices.rs index bf83975..d5116d6 100644 --- a/backend/src/db/devices.rs +++ b/backend/src/db/devices.rs @@ -91,7 +91,7 @@ pub fn list_devices( page_limit: Option, ) -> Result, DbError> { debug!("Listing devices"); - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; // Prepare SQL and parameters let (filters, mut params) = build_device_filters( @@ -151,7 +151,7 @@ pub fn count_devices( vendor: Option, ) -> Result { debug!("Counting devices"); - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let (filters, params) = build_device_filters( is_registered, @@ -174,7 +174,13 @@ pub fn count_devices( // Read device from its MAC address pub fn read(mac_address: String) -> Option { - let conn = db::get_db_connection(); + let conn = match db::get_db_connection() { + Ok(conn) => conn, + Err(error) => { + error!("Error obtaining database connection: {error}"); + return None; + } + }; let mac_address = normalize_mac(&mac_address); let result: Result = conn.query_one( @@ -209,7 +215,7 @@ pub fn read(mac_address: String) -> Option { } pub fn insert(device: Device) -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = normalize_mac(&device.mac_address); match conn.execute( @@ -228,7 +234,7 @@ pub fn insert(device: Device) -> Result<(), DbError> { pub fn get_summary() -> Result { debug!("Getting device summary"); - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let one_day_ago = (Utc::now() - Duration::days(1)).to_rfc3339(); let one_week_ago = (Utc::now() - Duration::weeks(1)).to_rfc3339(); @@ -277,7 +283,7 @@ pub fn seen( device_type: String, name: Option, ) -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = normalize_mac(&mac_address); let mut sql = "UPDATE devices SET ipv4_address=?, last_seen=?".to_string(); @@ -331,7 +337,7 @@ pub fn update( vendor: String, name: Option, ) -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = normalize_mac(&mac_address); match conn.execute( @@ -355,7 +361,7 @@ pub fn register( device_type: String, name: Option, ) -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = normalize_mac(&mac_address); // Only write the name column when supplied, so a user registering without typing a name @@ -382,7 +388,7 @@ pub fn register( } pub fn unregister(mac_address: String) -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = normalize_mac(&mac_address); match conn.execute( diff --git a/backend/src/db/error.rs b/backend/src/db/error.rs index 66d50cd..9adce22 100644 --- a/backend/src/db/error.rs +++ b/backend/src/db/error.rs @@ -4,6 +4,7 @@ use std::{error, fmt}; pub enum DbError { ParseRusqlite(rusqlite::Error), ParseRusqliteMigration(rusqlite_migration::Error), + Pool(r2d2::Error), } impl fmt::Display for DbError { @@ -11,6 +12,7 @@ impl fmt::Display for DbError { match *self { DbError::ParseRusqlite(..) => write!(f, "Database access error"), DbError::ParseRusqliteMigration(..) => write!(f, "Database migration error"), + DbError::Pool(..) => write!(f, "Database connection pool error"), } } } @@ -20,6 +22,7 @@ impl error::Error for DbError { match *self { DbError::ParseRusqlite(ref e) => Some(e), DbError::ParseRusqliteMigration(ref e) => Some(e), + DbError::Pool(ref e) => Some(e), } } } @@ -29,6 +32,11 @@ impl From for DbError { DbError::ParseRusqlite(err) } } +impl From for DbError { + fn from(err: r2d2::Error) -> DbError { + DbError::Pool(err) + } +} impl From for DbError { fn from(err: rusqlite_migration::Error) -> DbError { DbError::ParseRusqliteMigration(err) diff --git a/backend/src/db/notifications.rs b/backend/src/db/notifications.rs index 1e3cdae..65cfbd6 100644 --- a/backend/src/db/notifications.rs +++ b/backend/src/db/notifications.rs @@ -13,7 +13,7 @@ pub fn list( page_limit: Option, ) -> Result, DbError> { debug!("Listing notifications"); - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mut sql_statement = "SELECT id, created_on, notification_type, title, body, is_new, mac_address FROM notifications WHERE 1=1" @@ -57,7 +57,7 @@ pub fn list( // report the total number of pages alongside a `list` page. pub fn count(is_new: Option) -> Result { debug!("Counting notifications"); - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mut sql_statement = "SELECT COUNT(*) FROM notifications WHERE 1=1".to_string(); let mut params: Vec = Vec::new(); @@ -78,7 +78,7 @@ pub fn count(is_new: Option) -> Result { } pub fn insert(notification: Notification) -> Result { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; let mac_address = notification.mac_address.as_deref().map(normalize_mac); match conn.execute( @@ -96,7 +96,7 @@ pub fn insert(notification: Notification) -> Result { } pub fn mark_as_old(id: i64) -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; match conn.execute("UPDATE notifications SET is_new=0 WHERE id=?1", params![id]) { Ok(_) => { @@ -111,7 +111,7 @@ pub fn mark_as_old(id: i64) -> Result<(), DbError> { } pub fn mark_as_new(id: i64) -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; match conn.execute("UPDATE notifications SET is_new=1 WHERE id=?1", params![id]) { Ok(_) => { @@ -126,7 +126,7 @@ pub fn mark_as_new(id: i64) -> Result<(), DbError> { } pub fn mark_all_as_old() -> Result<(), DbError> { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; match conn.execute("UPDATE notifications SET is_new=0 WHERE is_new=1", []) { Ok(_) => { @@ -141,7 +141,7 @@ pub fn mark_all_as_old() -> Result<(), DbError> { } pub fn purge_older_than(cutoff: DateTime) -> Result { - let conn = db::get_db_connection(); + let conn = db::get_db_connection()?; match conn.execute( "DELETE FROM notifications WHERE created_on < ?1", @@ -159,7 +159,13 @@ pub fn purge_older_than(cutoff: DateTime) -> Result { } pub fn read(id: i64) -> Option { - let conn = db::get_db_connection(); + let conn = match db::get_db_connection() { + Ok(conn) => conn, + Err(error) => { + error!("Error obtaining database connection: {error}"); + return None; + } + }; let result: Result = conn.query_one( "SELECT id, created_on, notification_type, title, body, is_new, mac_address FROM notifications WHERE id=?1", diff --git a/backend/src/events.rs b/backend/src/events.rs index 1ddd4bc..26ec315 100644 --- a/backend/src/events.rs +++ b/backend/src/events.rs @@ -14,6 +14,82 @@ use crate::settings::get_settings; use chrono::{Local, Utc}; use duration_string::DurationString; use log::{debug, error, info, warn}; +use once_cell::sync::OnceCell; +use tokio::sync::mpsc; + +// A notification handed to the delivery loop. Delivery (a blocking Pushover HTTP call) runs on a +// dedicated task so a slow or unreachable Pushover can never stall the scan loops. +struct DeliveryRequest { + title: String, + body: String, +} + +// Bounded so a stuck delivery loop cannot grow memory without limit; on overflow we drop and warn +// (delivery is best-effort, matching the "never stop the loop" policy in the scanner pipeline). +const DELIVERY_QUEUE_CAPACITY: usize = 100; + +static DELIVERY_TX: OnceCell> = OnceCell::new(); + +/// Owns the receiving end of the notification-delivery channel and delivers notifications off the +/// scan loop. Run this as its own task (see `main`); it returns only if the channel is closed. +pub async fn run_delivery() { + let (tx, mut rx) = mpsc::channel::(DELIVERY_QUEUE_CAPACITY); + if DELIVERY_TX.set(tx).is_err() { + error!("Notification delivery loop started more than once; ignoring"); + return; + } + + while let Some(request) = rx.recv().await { + deliver(request).await; + } +} + +// Deliver a single notification according to the configured method. The Pushover call is blocking, +// so it runs on the blocking thread pool rather than the delivery task's async thread. +async fn deliver(request: DeliveryRequest) { + match get_settings().notifications.method.as_str() { + "pushover" => match &get_settings().notifications.pushover { + Some(config) => { + let config = config.clone(); + let result = tokio::task::spawn_blocking(move || { + pushover::send_message(&config, request.title, request.body) + }) + .await; + match result { + Ok(Ok(())) => {} + Ok(Err(err)) => error!("Failed to deliver notification via Pushover: {err}"), + Err(err) => error!("Notification delivery task panicked: {err}"), + } + } + None => { + error!( + "Notification method is 'pushover' but no [notifications.pushover] section is \ + configured; cannot deliver notification." + ); + } + }, + other => { + warn!("Notification method set to '{other}'. Set logs to 'info' to see notifications."); + info!("Notification: {}", request.body); + } + } +} + +// Hand a notification to the delivery loop. Never blocks: if the loop is not running (e.g. in +// tests) or its queue is full, the notification is logged and dropped rather than stalling the +// caller (a scan loop). +fn enqueue_delivery(title: String, body: String) { + match DELIVERY_TX.get() { + Some(tx) => { + if let Err(err) = tx.try_send(DeliveryRequest { title, body }) { + warn!("Notification delivery queue unavailable; dropping notification: {err}"); + } + } + None => { + info!("Notification (delivery loop not running): {body}"); + } + } +} // Device name for display in messages; falls back to "(unknown)" for devices with no // mDNS-discovered hostname (e.g. those found only via ARP). @@ -177,30 +253,18 @@ fn render_device_changed( (title, body) } -// Private helper function to deliver messages +// Private helper function to record a notification and hand it off for delivery. Delivery happens +// on a separate task (see `run_delivery`), so this returns as soon as the notification is persisted +// and never blocks the caller on the (potentially slow) Pushover HTTP call. fn send_notification(notification: Notification) -> Result<(), Box> { debug!("About to record notification in database"); - db::notifications::insert(notification.clone())?; - debug!("About to send notification ({notification})."); + let title = notification.title.clone(); + let body = notification.body.clone(); + db::notifications::insert(notification)?; - match get_settings().notifications.method.as_str() { - "pushover" => match &get_settings().notifications.pushover { - Some(pushover) => { - pushover::send_message(pushover, notification.title, notification.body)?; - } - None => { - error!( - "Notification method is 'pushover' but no [notifications.pushover] section is \ - configured; cannot deliver notification." - ); - } - }, - other => { - warn!("Notification method set to '{other}'. Set logs to 'info' to see notifications."); - info!("Notification: {}", notification.body); - } - }; + debug!("Queued notification for delivery: {title}"); + enqueue_delivery(title, body); Ok(()) } @@ -506,4 +570,25 @@ mod tests { "A sighting from a different scanner should be recorded" ); } + + #[tokio::test] + async fn triggering_a_new_device_records_a_notification() { + crate::tests_common::setup().await; + + let mac = "fa:ce:fa:ce:00:02".to_string(); + let mut device = sample_device(Some("printer")); + device.mac_address = mac.clone(); + + // The delivery loop is not running in tests, so delivery is a no-op; the notification must + // still be persisted regardless of whether it is ever delivered. + trigger_new_device(device, DeviceEventScanner::Arp).unwrap(); + + let notifications = db::notifications::list(None, None, None).unwrap(); + assert!( + notifications + .iter() + .any(|n| n.mac_address.as_deref() == Some(mac.as_str())), + "A new-device sighting should persist a notification" + ); + } } diff --git a/backend/src/main.rs b/backend/src/main.rs index 3d385e8..91a7ebb 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -63,7 +63,8 @@ async fn main() -> Result<(), Box> { scanners::dhcp::status::STATUS.init(); scanners::snmp::status::STATUS.init(); - // Start the device scanners, web server, and retention cleaner in parallel + // Start the device scanners, web server, retention cleaner, and notification delivery loop in + // parallel. Notification delivery runs on its own task so a slow Pushover never stalls a scan. tokio::join!( scanners::arp::scanner::scan(), scanners::mdns::scanner::listen(), @@ -71,7 +72,8 @@ async fn main() -> Result<(), Box> { scanners::dhcp::scanner::listen(), scanners::snmp::scanner::scan(), web_server::serve(), - retention::run() + retention::run(), + events::run_delivery() ) .0 } diff --git a/backend/src/tests_common.rs b/backend/src/tests_common.rs index cd6c4d6..392784e 100644 --- a/backend/src/tests_common.rs +++ b/backend/src/tests_common.rs @@ -36,7 +36,7 @@ pub async fn setup() { // Initialize database db::init_db().await.unwrap(); info!("Initializing database for testing."); - let conn = db::get_db_connection(); + let conn = db::get_db_connection().expect("failed to get a database connection for tests"); debug!("Running database setup scripts for testing."); for entry in TEST_MIGRATIONS_DIR.entries() { diff --git a/backend/src/web_server/device_events.rs b/backend/src/web_server/device_events.rs index d8671b3..8b5a4a0 100644 --- a/backend/src/web_server/device_events.rs +++ b/backend/src/web_server/device_events.rs @@ -33,11 +33,14 @@ pub async fn list( let page_offset: Option = utils::parse_parameter(¶ms, "page_offset"); let page_limit: Option = utils::parse_parameter(¶ms, "page_limit"); - match db::device_events::list(Some(mac_address), created_from, page_offset, page_limit) { - Ok(value) => Ok(Json(value)), - Err(err) => { - error!("Error listing device events: {}", err); - Err(StatusCode::INTERNAL_SERVER_ERROR) + db::run_blocking(move || { + match db::device_events::list(Some(mac_address), created_from, page_offset, page_limit) { + Ok(value) => Ok(Json(value)), + Err(err) => { + error!("Error listing device events: {}", err); + Err(StatusCode::INTERNAL_SERVER_ERROR) + } } - } + }) + .await } diff --git a/backend/src/web_server/devices.rs b/backend/src/web_server/devices.rs index e96c3c7..6495b82 100644 --- a/backend/src/web_server/devices.rs +++ b/backend/src/web_server/devices.rs @@ -52,41 +52,44 @@ pub async fn list( let page_offset: Option = utils::parse_parameter(¶ms, "page_offset"); let page_limit: Option = utils::parse_parameter(¶ms, "page_limit"); - let items = match db::devices::list_devices( - is_registered, - last_seen_from, - last_seen_to, - owner.clone(), - device_type.clone(), - vendor.clone(), - sort_by, - sort_order, - page_offset, - page_limit, - ) { - Ok(value) => value, - Err(err) => { - error!("Error listing devices: {}", err); - return Err(StatusCode::INTERNAL_SERVER_ERROR); - } - }; + db::run_blocking(move || { + let items = match db::devices::list_devices( + is_registered, + last_seen_from, + last_seen_to, + owner.clone(), + device_type.clone(), + vendor.clone(), + sort_by, + sort_order, + page_offset, + page_limit, + ) { + Ok(value) => value, + Err(err) => { + error!("Error listing devices: {}", err); + return Err(StatusCode::INTERNAL_SERVER_ERROR); + } + }; - let total_count = match db::devices::count_devices( - is_registered, - last_seen_from, - last_seen_to, - owner, - device_type, - vendor, - ) { - Ok(value) => value, - Err(err) => { - error!("Error counting devices: {}", err); - return Err(StatusCode::INTERNAL_SERVER_ERROR); - } - }; + let total_count = match db::devices::count_devices( + is_registered, + last_seen_from, + last_seen_to, + owner, + device_type, + vendor, + ) { + Ok(value) => value, + Err(err) => { + error!("Error counting devices: {}", err); + return Err(StatusCode::INTERNAL_SERVER_ERROR); + } + }; - Ok(Json(DeviceListResponse { items, total_count })) + Ok(Json(DeviceListResponse { items, total_count })) + }) + .await } #[utoipa::path( @@ -103,10 +106,11 @@ pub async fn list( security(("bearer_auth" = [])) )] pub async fn read(Path(mac_address): Path) -> Result, StatusCode> { - match db::devices::read(mac_address) { + db::run_blocking(move || match db::devices::read(mac_address) { Some(value) => Ok(Json(value)), None => Err(StatusCode::NOT_FOUND), - } + }) + .await } #[utoipa::path( @@ -128,38 +132,41 @@ pub async fn register(Json(payload): Json) -> impl IntoRe payload.mac_address, payload.owner, payload.device_type ); - let device = match db::devices::read(payload.mac_address.clone()) { - Some(value) => value, - None => { + db::run_blocking(move || { + let device = match db::devices::read(payload.mac_address.clone()) { + Some(value) => value, + None => { + return ( + axum::http::StatusCode::NOT_FOUND, + "Device not found or could not be read", + ); + } + }; + + if device.is_registered { return ( - axum::http::StatusCode::NOT_FOUND, - "Device not found or could not be read", + axum::http::StatusCode::CONFLICT, + "Device already registered", ); } - }; - if device.is_registered { - return ( - axum::http::StatusCode::CONFLICT, - "Device already registered", - ); - } - - match db::devices::register( - payload.mac_address, - payload.owner, - payload.device_type, - payload.name, - ) { - Ok(_) => (axum::http::StatusCode::CREATED, "Device registered"), - Err(err) => { - error!("Error registering device in the database: {}", err); - ( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "Error registering device in the server, check your logs", - ) + match db::devices::register( + payload.mac_address, + payload.owner, + payload.device_type, + payload.name, + ) { + Ok(_) => (axum::http::StatusCode::CREATED, "Device registered"), + Err(err) => { + error!("Error registering device in the database: {}", err); + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + "Error registering device in the server, check your logs", + ) + } } - } + }) + .await } #[utoipa::path( @@ -187,39 +194,42 @@ pub async fn update( mac_address, payload.owner, payload.device_type, payload.vendor, payload.name ); - let device = match db::devices::read(mac_address.clone()) { - Some(value) => value, - None => { + db::run_blocking(move || { + let device = match db::devices::read(mac_address.clone()) { + Some(value) => value, + None => { + return ( + axum::http::StatusCode::NOT_FOUND, + "Device not found or could not be read", + ); + } + }; + + if !device.is_registered { return ( - axum::http::StatusCode::NOT_FOUND, - "Device not found or could not be read", + axum::http::StatusCode::CONFLICT, + "Device is not registered, register it before modifying", ); } - }; - if !device.is_registered { - return ( - axum::http::StatusCode::CONFLICT, - "Device is not registered, register it before modifying", - ); - } - - match db::devices::update( - mac_address, - payload.owner, - payload.device_type, - payload.vendor, - payload.name, - ) { - Ok(_) => (axum::http::StatusCode::OK, "Device updated"), - Err(err) => { - error!("Error updating device in the database: {}", err); - ( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "Error updating device in the server, check your logs", - ) + match db::devices::update( + mac_address, + payload.owner, + payload.device_type, + payload.vendor, + payload.name, + ) { + Ok(_) => (axum::http::StatusCode::OK, "Device updated"), + Err(err) => { + error!("Error updating device in the database: {}", err); + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + "Error updating device in the server, check your logs", + ) + } } - } + }) + .await } #[utoipa::path( @@ -238,33 +248,36 @@ pub async fn update( security(("bearer_auth" = [])) )] pub async fn unregister(Path(mac_address): Path) -> impl IntoResponse { - let device = match db::devices::read(mac_address.clone()) { - Some(value) => value, - None => { + db::run_blocking(move || { + let device = match db::devices::read(mac_address.clone()) { + Some(value) => value, + None => { + return ( + axum::http::StatusCode::NOT_FOUND, + "Device not found or could not be read", + ); + } + }; + + if !device.is_registered { return ( - axum::http::StatusCode::NOT_FOUND, - "Device not found or could not be read", + axum::http::StatusCode::CONFLICT, + "Device not registered, you cannot un-register it again", ); } - }; - if !device.is_registered { - return ( - axum::http::StatusCode::CONFLICT, - "Device not registered, you cannot un-register it again", - ); - } - - match db::devices::unregister(mac_address) { - Ok(_) => (axum::http::StatusCode::OK, "Device un-registered"), - Err(err) => { - error!("Error updating device in the database: {}", err); - ( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "Error updating device in the server, check your logs", - ) + match db::devices::unregister(mac_address) { + Ok(_) => (axum::http::StatusCode::OK, "Device un-registered"), + Err(err) => { + error!("Error updating device in the database: {}", err); + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + "Error updating device in the server, check your logs", + ) + } } - } + }) + .await } #[utoipa::path( @@ -278,13 +291,14 @@ pub async fn unregister(Path(mac_address): Path) -> impl IntoResponse { security(("bearer_auth" = [])) )] pub async fn summary() -> Result, StatusCode> { - match db::devices::get_summary() { + db::run_blocking(move || match db::devices::get_summary() { Ok(value) => Ok(Json(value)), Err(err) => { error!("Error getting device summary: {}", err); Err(StatusCode::INTERNAL_SERVER_ERROR) } - } + }) + .await } // Payload structs diff --git a/backend/src/web_server/notifications.rs b/backend/src/web_server/notifications.rs index 6a63738..488b802 100644 --- a/backend/src/web_server/notifications.rs +++ b/backend/src/web_server/notifications.rs @@ -29,18 +29,21 @@ use crate::{ security(("bearer_auth" = [])) )] pub async fn read(Path(id): Path) -> Result, StatusCode> { - match db::notifications::mark_as_old(id) { - Ok(_) => {} - Err(err) => { - error!("Error marking notification (id={id}) as old: {}", err); - return Err(StatusCode::INTERNAL_SERVER_ERROR); - } - }; + db::run_blocking(move || { + match db::notifications::mark_as_old(id) { + Ok(_) => {} + Err(err) => { + error!("Error marking notification (id={id}) as old: {}", err); + return Err(StatusCode::INTERNAL_SERVER_ERROR); + } + }; - match db::notifications::read(id) { - Some(value) => Ok(Json(value)), - None => Err(StatusCode::NOT_FOUND), - } + match db::notifications::read(id) { + Some(value) => Ok(Json(value)), + None => Err(StatusCode::NOT_FOUND), + } + }) + .await } #[utoipa::path( @@ -57,10 +60,11 @@ pub async fn read(Path(id): Path) -> Result, StatusCode> security(("bearer_auth" = [])) )] pub async fn read_without_flagging(Path(id): Path) -> Result, StatusCode> { - match db::notifications::read(id) { + db::run_blocking(move || match db::notifications::read(id) { Some(value) => Ok(Json(value)), None => Err(StatusCode::NOT_FOUND), - } + }) + .await } #[utoipa::path( @@ -77,7 +81,7 @@ pub async fn read_without_flagging(Path(id): Path) -> Result) -> impl IntoResponse { - match db::notifications::mark_as_new(id) { + db::run_blocking(move || match db::notifications::mark_as_new(id) { Ok(_) => (StatusCode::OK, "Notification marked as new"), Err(err) => { error!("Error marking notification (id={id}) as new: {}", err); @@ -86,7 +90,8 @@ pub async fn mark_as_new(Path(id): Path) -> impl IntoResponse { "Error updating notification in the server, check your logs", ) } - } + }) + .await } #[utoipa::path( @@ -100,7 +105,7 @@ pub async fn mark_as_new(Path(id): Path) -> impl IntoResponse { security(("bearer_auth" = [])) )] pub async fn mark_all_as_old() -> impl IntoResponse { - match db::notifications::mark_all_as_old() { + db::run_blocking(move || match db::notifications::mark_all_as_old() { Ok(_) => (StatusCode::OK, "All notifications marked as old"), Err(err) => { error!("Error marking all notifications as old: {}", err); @@ -109,7 +114,8 @@ pub async fn mark_all_as_old() -> impl IntoResponse { "Error updating notifications in the server, check your logs", ) } - } + }) + .await } #[utoipa::path( @@ -134,21 +140,24 @@ pub async fn list( let page_offset: Option = utils::parse_parameter(¶ms, "page_offset"); let page_limit: Option = utils::parse_parameter(¶ms, "page_limit"); - let items = match db::notifications::list(is_new, page_offset, page_limit) { - Ok(value) => value, - Err(err) => { - error!("Error listing notifications: {}", err); - return Err(StatusCode::INTERNAL_SERVER_ERROR); - } - }; + db::run_blocking(move || { + let items = match db::notifications::list(is_new, page_offset, page_limit) { + Ok(value) => value, + Err(err) => { + error!("Error listing notifications: {}", err); + return Err(StatusCode::INTERNAL_SERVER_ERROR); + } + }; - let total_count = match db::notifications::count(is_new) { - Ok(value) => value, - Err(err) => { - error!("Error counting notifications: {}", err); - return Err(StatusCode::INTERNAL_SERVER_ERROR); - } - }; + let total_count = match db::notifications::count(is_new) { + Ok(value) => value, + Err(err) => { + error!("Error counting notifications: {}", err); + return Err(StatusCode::INTERNAL_SERVER_ERROR); + } + }; - Ok(Json(NotificationListResponse { items, total_count })) + Ok(Json(NotificationListResponse { items, total_count })) + }) + .await }