mirror of
https://github.com/rzuasti/oott.git
synced 2026-07-08 19:21:54 +02:00
Add retention policy for device_events and notifications
Purges records older than a configurable window (default 365d) once per day. Also adds a composite index on device_events(mac_address, created_on DESC) for efficient scan history queries at scale, and fixes a non-deterministic pagination order by adding id as a tiebreaker to ORDER BY. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
888f18603f
commit
c7e1678407
@@ -0,0 +1 @@
|
|||||||
|
CREATE INDEX idx_device_events_mac_created ON device_events (mac_address, created_on DESC);
|
||||||
@@ -1,3 +1,3 @@
|
|||||||
#!/bin/sh
|
#!/bin/sh
|
||||||
sudo rm -f oott.db
|
sudo rm -f oott.db
|
||||||
sudo CARGO_HOME=$HOME/.cargo cargo test -- --show-output
|
sudo CARGO_HOME=$HOME/.cargo cargo test -- --show-output --test-threads=1
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
use crate::db;
|
use crate::db;
|
||||||
use crate::db::error::DbError;
|
use crate::db::error::DbError;
|
||||||
|
use chrono::{DateTime, Utc};
|
||||||
use log::{debug, error};
|
use log::{debug, error};
|
||||||
use rusqlite::{params, params_from_iter};
|
use rusqlite::{params, params_from_iter};
|
||||||
|
|
||||||
@@ -49,7 +50,7 @@ pub fn list(
|
|||||||
params.push(mac.into());
|
params.push(mac.into());
|
||||||
}
|
}
|
||||||
|
|
||||||
sql_statement.push_str(" ORDER BY created_on DESC");
|
sql_statement.push_str(" ORDER BY created_on DESC, id DESC");
|
||||||
|
|
||||||
if let (Some(page_offset), Some(page_limit)) = (page_offset, page_limit) {
|
if let (Some(page_offset), Some(page_limit)) = (page_offset, page_limit) {
|
||||||
debug!(
|
debug!(
|
||||||
@@ -79,6 +80,24 @@ pub fn list(
|
|||||||
Ok(events)
|
Ok(events)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn purge_older_than(cutoff: DateTime<Utc>) -> Result<usize, DbError> {
|
||||||
|
let conn = db::get_db_connection();
|
||||||
|
|
||||||
|
match conn.execute(
|
||||||
|
"DELETE FROM device_events WHERE created_on < ?1",
|
||||||
|
params![cutoff],
|
||||||
|
) {
|
||||||
|
Ok(count) => {
|
||||||
|
debug!("Purged {} device event(s) older than {}", count, cutoff);
|
||||||
|
Ok(count)
|
||||||
|
}
|
||||||
|
Err(error) => {
|
||||||
|
error!("Error purging old device events: {error}");
|
||||||
|
Err(DbError::from(error))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
fn read(id: i64) -> Option<DeviceEvent> {
|
fn read(id: i64) -> Option<DeviceEvent> {
|
||||||
let conn = db::get_db_connection();
|
let conn = db::get_db_connection();
|
||||||
@@ -198,18 +217,80 @@ mod tests {
|
|||||||
assert!(events.is_empty(), "Unknown MAC should return empty list");
|
assert!(events.is_empty(), "Unknown MAC should return empty list");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_purge_older_than() {
|
||||||
|
tests_common::setup().await;
|
||||||
|
|
||||||
|
let mac = "ee:ee:ee:ee:ee:ee".to_string();
|
||||||
|
let old_id = insert(DeviceEvent::new(
|
||||||
|
mac.clone(),
|
||||||
|
chrono::DateTime::parse_from_rfc3339("2020-01-01T00:00:00+00:00")
|
||||||
|
.unwrap()
|
||||||
|
.into(),
|
||||||
|
DeviceEventType::NewDevice,
|
||||||
|
"10.0.0.200".to_string(),
|
||||||
|
"Old Vendor".to_string(),
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let recent_id = insert(DeviceEvent::new(
|
||||||
|
mac.clone(),
|
||||||
|
Utc::now(),
|
||||||
|
DeviceEventType::DeviceSeen,
|
||||||
|
"10.0.0.200".to_string(),
|
||||||
|
"Old Vendor".to_string(),
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let cutoff = Utc::now() - chrono::TimeDelta::days(365);
|
||||||
|
let purged = purge_older_than(cutoff).unwrap();
|
||||||
|
|
||||||
|
assert!(purged >= 1, "At least 1 device event should have been purged");
|
||||||
|
assert!(read(old_id).is_none(), "Old device event should have been purged");
|
||||||
|
assert!(read(recent_id).is_some(), "Recent device event should not have been purged");
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_list_pagination() {
|
async fn test_list_pagination() {
|
||||||
tests_common::setup().await;
|
tests_common::setup().await;
|
||||||
|
|
||||||
let first_page = list(None, Some(0), Some(2)).unwrap();
|
let mac = "cc:cc:cc:cc:cc:cc".to_string();
|
||||||
|
insert(DeviceEvent::new(
|
||||||
|
mac.clone(),
|
||||||
|
chrono::DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
|
||||||
|
.unwrap()
|
||||||
|
.into(),
|
||||||
|
DeviceEventType::NewDevice,
|
||||||
|
"10.0.0.1".to_string(),
|
||||||
|
"Vendor C".to_string(),
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
insert(DeviceEvent::new(
|
||||||
|
mac.clone(),
|
||||||
|
chrono::DateTime::parse_from_rfc3339("2026-02-01T00:00:00Z")
|
||||||
|
.unwrap()
|
||||||
|
.into(),
|
||||||
|
DeviceEventType::DeviceSeen,
|
||||||
|
"10.0.0.1".to_string(),
|
||||||
|
"Vendor C".to_string(),
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
insert(DeviceEvent::new(
|
||||||
|
mac.clone(),
|
||||||
|
chrono::DateTime::parse_from_rfc3339("2026-03-01T00:00:00Z")
|
||||||
|
.unwrap()
|
||||||
|
.into(),
|
||||||
|
DeviceEventType::DeviceSeen,
|
||||||
|
"10.0.0.1".to_string(),
|
||||||
|
"Vendor C".to_string(),
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let first_page = list(Some(mac.clone()), Some(0), Some(2)).unwrap();
|
||||||
assert_eq!(first_page.len(), 2, "First page should have 2 events");
|
assert_eq!(first_page.len(), 2, "First page should have 2 events");
|
||||||
|
|
||||||
let second_page = list(None, Some(2), Some(2)).unwrap();
|
let second_page = list(Some(mac.clone()), Some(2), Some(2)).unwrap();
|
||||||
assert!(
|
assert_eq!(second_page.len(), 1, "Second page should have 1 event");
|
||||||
second_page.len() >= 1,
|
|
||||||
"Second page should have at least 1 event"
|
|
||||||
);
|
|
||||||
|
|
||||||
let first_ids: Vec<i64> = first_page.iter().map(|e| e.id).collect();
|
let first_ids: Vec<i64> = first_page.iter().map(|e| e.id).collect();
|
||||||
for event in &second_page {
|
for event in &second_page {
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
use crate::db;
|
use crate::db;
|
||||||
use crate::db::error::DbError;
|
use crate::db::error::DbError;
|
||||||
|
use chrono::{DateTime, Utc};
|
||||||
use log::{debug, error};
|
use log::{debug, error};
|
||||||
use rusqlite::{params, params_from_iter};
|
use rusqlite::{params, params_from_iter};
|
||||||
|
|
||||||
@@ -123,6 +124,24 @@ pub fn mark_all_as_old() -> Result<(), DbError> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn purge_older_than(cutoff: DateTime<Utc>) -> Result<usize, DbError> {
|
||||||
|
let conn = db::get_db_connection();
|
||||||
|
|
||||||
|
match conn.execute(
|
||||||
|
"DELETE FROM notifications WHERE created_on < ?1",
|
||||||
|
params![cutoff],
|
||||||
|
) {
|
||||||
|
Ok(count) => {
|
||||||
|
debug!("Purged {} notification(s) older than {}", count, cutoff);
|
||||||
|
Ok(count)
|
||||||
|
}
|
||||||
|
Err(error) => {
|
||||||
|
error!("Error purging old notifications: {error}");
|
||||||
|
Err(DbError::from(error))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub fn read(id: i64) -> Option<Notification> {
|
pub fn read(id: i64) -> Option<Notification> {
|
||||||
let conn = db::get_db_connection();
|
let conn = db::get_db_connection();
|
||||||
|
|
||||||
@@ -163,6 +182,40 @@ mod tests {
|
|||||||
use super::*;
|
use super::*;
|
||||||
use crate::{model::notifications::NotificationType, tests_common};
|
use crate::{model::notifications::NotificationType, tests_common};
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_purge_older_than() {
|
||||||
|
tests_common::setup().await;
|
||||||
|
|
||||||
|
let old_id = insert(Notification::new(
|
||||||
|
chrono::DateTime::parse_from_rfc3339("2020-01-01T00:00:00+00:00")
|
||||||
|
.unwrap()
|
||||||
|
.into(),
|
||||||
|
NotificationType::Other,
|
||||||
|
"Old notification".to_string(),
|
||||||
|
"Old body".to_string(),
|
||||||
|
false,
|
||||||
|
None,
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let recent_id = insert(Notification::new(
|
||||||
|
Utc::now(),
|
||||||
|
NotificationType::Other,
|
||||||
|
"Recent notification".to_string(),
|
||||||
|
"Recent body".to_string(),
|
||||||
|
false,
|
||||||
|
None,
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let cutoff = Utc::now() - chrono::TimeDelta::days(365);
|
||||||
|
let purged = purge_older_than(cutoff).unwrap();
|
||||||
|
|
||||||
|
assert!(purged >= 1, "At least 1 notification should have been purged");
|
||||||
|
assert!(read(old_id).is_none(), "Old notification should have been purged");
|
||||||
|
assert!(read(recent_id).is_some(), "Recent notification should not have been purged");
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_mark_as_old() {
|
async fn test_mark_as_old() {
|
||||||
tests_common::setup().await;
|
tests_common::setup().await;
|
||||||
|
|||||||
+3
-2
@@ -7,6 +7,7 @@ mod device_finders;
|
|||||||
mod events;
|
mod events;
|
||||||
mod mac_vendor_finder;
|
mod mac_vendor_finder;
|
||||||
mod model;
|
mod model;
|
||||||
|
mod retention;
|
||||||
mod scanner;
|
mod scanner;
|
||||||
mod settings;
|
mod settings;
|
||||||
mod utils;
|
mod utils;
|
||||||
@@ -56,6 +57,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|||||||
// Initialize database
|
// Initialize database
|
||||||
db::init_db().await?;
|
db::init_db().await?;
|
||||||
|
|
||||||
// Start the device scanner and web server (for API and UI) in parallel
|
// Start the device scanner, web server, and retention cleaner in parallel
|
||||||
tokio::join!(scanner::scan(), web_server::serve()).0
|
tokio::join!(scanner::scan(), web_server::serve(), retention::run()).0
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,33 @@
|
|||||||
|
use crate::db;
|
||||||
|
use crate::settings::get_settings;
|
||||||
|
use chrono::Utc;
|
||||||
|
use log::{error, info};
|
||||||
|
use tokio::time::{Duration, sleep};
|
||||||
|
|
||||||
|
pub async fn run() {
|
||||||
|
loop {
|
||||||
|
let window: std::time::Duration = get_settings().retention.window.into();
|
||||||
|
let cutoff = match chrono::Duration::from_std(window) {
|
||||||
|
Ok(d) => Utc::now() - d,
|
||||||
|
Err(e) => {
|
||||||
|
error!("Retention: invalid window duration: {}", e);
|
||||||
|
sleep(Duration::from_secs(24 * 60 * 60)).await;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
info!("Retention: purging records older than {}", cutoff);
|
||||||
|
|
||||||
|
match db::device_events::purge_older_than(cutoff) {
|
||||||
|
Ok(count) => info!("Retention: purged {} device event(s)", count),
|
||||||
|
Err(e) => error!("Retention: error purging device events: {}", e),
|
||||||
|
}
|
||||||
|
|
||||||
|
match db::notifications::purge_older_than(cutoff) {
|
||||||
|
Ok(count) => info!("Retention: purged {} notification(s)", count),
|
||||||
|
Err(e) => error!("Retention: error purging notifications: {}", e),
|
||||||
|
}
|
||||||
|
|
||||||
|
sleep(Duration::from_secs(24 * 60 * 60)).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -48,6 +48,19 @@ pub struct WebServer {
|
|||||||
pub api_key: String,
|
pub api_key: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Deserialize, Clone)]
|
||||||
|
pub struct Retention {
|
||||||
|
pub window: DurationString,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for Retention {
|
||||||
|
fn default() -> Self {
|
||||||
|
Retention {
|
||||||
|
window: DurationString::try_from("365d".to_string()).unwrap(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug, Deserialize, Clone)]
|
#[derive(Debug, Deserialize, Clone)]
|
||||||
pub struct Settings {
|
pub struct Settings {
|
||||||
pub database: Database,
|
pub database: Database,
|
||||||
@@ -56,6 +69,8 @@ pub struct Settings {
|
|||||||
pub timings: Timings,
|
pub timings: Timings,
|
||||||
pub notifications: Notifications,
|
pub notifications: Notifications,
|
||||||
pub web_server: WebServer,
|
pub web_server: WebServer,
|
||||||
|
#[serde(default)]
|
||||||
|
pub retention: Retention,
|
||||||
}
|
}
|
||||||
// End configuration structure
|
// End configuration structure
|
||||||
// -----------------------------------------------------------
|
// -----------------------------------------------------------
|
||||||
|
|||||||
@@ -24,3 +24,6 @@ user_key="" # User key goes here, this is the account wide code for pushover
|
|||||||
ip_address="0.0.0.0" # IP to bind the web server for the API and web UI to, use 0.0.0.0 to bind it to all interfaces
|
ip_address="0.0.0.0" # IP to bind the web server for the API and web UI to, use 0.0.0.0 to bind it to all interfaces
|
||||||
port=3000 # Port to listen on
|
port=3000 # Port to listen on
|
||||||
api_key="CHANGE_ME" # API Key to use the system's API, change this!
|
api_key="CHANGE_ME" # API Key to use the system's API, change this!
|
||||||
|
|
||||||
|
[retention]
|
||||||
|
window="365d" # How long to keep device events and notifications. Records older than this are deleted daily. Supports: d (days), w (weeks), h (hours), m (minutes)
|
||||||
|
|||||||
Reference in New Issue
Block a user