diff --git a/LORAWEB_NEU/backend/daemon/src/alarm.rs b/LORAWEB_NEU/backend/daemon/src/alarm.rs new file mode 100644 index 0000000..69482f3 --- /dev/null +++ b/LORAWEB_NEU/backend/daemon/src/alarm.rs @@ -0,0 +1,141 @@ +// backend/daemon/src/alarm.rs +use sqlx::SqlitePool; +use chrono::Utc; +use std::time::Duration; +use tokio::time; + +pub async fn run_alarm_monitor( + pool: SqlitePool, + warning_hours: f64, + alarm_hours: f64, + interval_secs: u64, +) { + let mut interval = time::interval(Duration::from_secs(interval_secs)); + tracing::info!( + "Alarm monitor started (warning: {}h, alarm: {}h, interval: {}s)", + warning_hours, alarm_hours, interval_secs + ); + + loop { + interval.tick().await; + if let Err(e) = check_alarms(&pool, warning_hours, alarm_hours).await { + tracing::warn!("Alarm check failed: {}", e); + } + } +} + +pub async fn check_alarms( + pool: &SqlitePool, + global_warning_hours: f64, + global_alarm_hours: f64, +) -> Result<(), sqlx::Error> { + let topics: Vec<(String, Option, f64, String)> = sqlx::query_as( + "SELECT id, last_seen, alarm_threshold_hours, alarm_status FROM topics WHERE approved = 1" + ) + .fetch_all(pool) + .await?; + + let now = Utc::now(); + + for (id, last_seen, threshold_hours, current_status) in topics { + let new_status = match last_seen { + None => "unknown".to_string(), + Some(ref ts) => { + let alarm_h = if threshold_hours > 0.0 { threshold_hours } else { global_alarm_hours }; + let warning_h = if threshold_hours > 0.0 { + threshold_hours / 3.0 + } else { + global_warning_hours + }; + + match chrono::DateTime::parse_from_rfc3339(ts) { + Ok(last) => { + let hours = (now - last.with_timezone(&Utc)) + .num_minutes() as f64 / 60.0; + if hours < warning_h { + "active".to_string() + } else if hours < alarm_h { + "warning".to_string() + } else { + "alarm".to_string() + } + } + Err(_) => "unknown".to_string(), + } + } + }; + + if new_status != current_status { + tracing::info!( + "Topic {} status: {} -> {}", + id, current_status, new_status + ); + sqlx::query("UPDATE topics SET alarm_status = ? WHERE id = ?") + .bind(&new_status) + .bind(&id) + .execute(pool) + .await?; + } + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db; + + async fn setup_db() -> SqlitePool { + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); + db::init_db(&pool).await.unwrap(); + pool + } + + #[tokio::test] + async fn test_alarm_no_last_seen_is_unknown() { + let pool = setup_db().await; + sqlx::query( + "INSERT INTO topics (id, topic, approved, alarm_status, alarm_threshold_hours, created_at) \ + VALUES ('t1', 'dev1', 1, 'active', 0.0, '2026-01-01T00:00:00Z')" + ).execute(&pool).await.unwrap(); + + check_alarms(&pool, 2.0, 6.0).await.unwrap(); + + let status: (String,) = sqlx::query_as("SELECT alarm_status FROM topics WHERE id = 't1'") + .fetch_one(&pool).await.unwrap(); + assert_eq!(status.0, "unknown"); + } + + #[tokio::test] + async fn test_alarm_recent_is_active() { + let pool = setup_db().await; + let recent = Utc::now().to_rfc3339(); + sqlx::query( + "INSERT INTO topics (id, topic, approved, last_seen, alarm_status, alarm_threshold_hours, created_at) \ + VALUES ('t1', 'dev1', 1, ?, 'unknown', 0.0, '2026-01-01T00:00:00Z')" + ).bind(&recent).execute(&pool).await.unwrap(); + + check_alarms(&pool, 2.0, 6.0).await.unwrap(); + + let status: (String,) = sqlx::query_as("SELECT alarm_status FROM topics WHERE id = 't1'") + .fetch_one(&pool).await.unwrap(); + assert_eq!(status.0, "active"); + } + + #[tokio::test] + async fn test_alarm_old_is_alarm() { + let pool = setup_db().await; + let old = (Utc::now() - chrono::Duration::hours(10)).to_rfc3339(); + sqlx::query( + "INSERT INTO topics (id, topic, approved, last_seen, alarm_status, alarm_threshold_hours, created_at) \ + VALUES ('t1', 'dev1', 1, ?, 'active', 0.0, '2026-01-01T00:00:00Z')" + ).bind(&old).execute(&pool).await.unwrap(); + + check_alarms(&pool, 2.0, 6.0).await.unwrap(); + + let status: (String,) = sqlx::query_as("SELECT alarm_status FROM topics WHERE id = 't1'") + .fetch_one(&pool).await.unwrap(); + assert_eq!(status.0, "alarm"); + } +} diff --git a/LORAWEB_NEU/backend/daemon/src/main.rs b/LORAWEB_NEU/backend/daemon/src/main.rs index 924fc72..6b3475c 100644 --- a/LORAWEB_NEU/backend/daemon/src/main.rs +++ b/LORAWEB_NEU/backend/daemon/src/main.rs @@ -1,8 +1,14 @@ +mod alarm; mod config; mod db; +mod http; +mod influx; use crate::config::AppConfig; +use crate::http::IngestState; +use crate::influx::InfluxWriter; use sqlx::sqlite::SqlitePoolOptions; +use std::sync::Arc; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -22,14 +28,36 @@ async fn main() -> anyhow::Result<()> { .await?; db::init_db(&pool).await?; - tracing::info!( - "LoRaWAN Daemon ready — Ingest: http://{}:{}/", - cfg.http.bind, cfg.http.port + let influx = InfluxWriter::new( + &cfg.influx.url, &cfg.influx.token, + &cfg.influx.org, &cfg.influx.bucket, ); + + // Spawn alarm monitor + let alarm_pool = pool.clone(); + let alarm_cfg = cfg.alarm.clone(); + tokio::spawn(async move { + alarm::run_alarm_monitor( + alarm_pool, + alarm_cfg.warning_threshold_hours, + alarm_cfg.alarm_threshold_hours, + alarm_cfg.check_interval_secs, + ).await; + }); + + // Start HTTP ingest + let state = Arc::new(IngestState { pool, influx }); + let app = http::router(state); + + let addr = format!("{}:{}", cfg.http.bind, cfg.http.port); + tracing::info!("LoRaWAN Daemon ready — Ingest: http://{}/", addr); tracing::info!( "Configure ChirpStack HTTP Integration: POST http://:{}/", cfg.http.port ); + let listener = tokio::net::TcpListener::bind(&addr).await?; + axum::serve(listener, app).await?; + Ok(()) }