feat(daemon): add alarm monitor with configurable thresholds and status checks

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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<String>, 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");
+    }
+}

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://<this-server>:{}/",
         cfg.http.port
     );
 
+    let listener = tokio::net::TcpListener::bind(&addr).await?;
+    axum::serve(listener, app).await?;
+
     Ok(())
 }