feat: reorganize to one flake one rust project

remote/cache-service/src/api.rs

@@ -23,6 +23,8 @@ pub struct AppState {
     pub webhooks: Vec<WebhookTarget>,
     pub retry_attempts: u32,
     pub retry_base_delay_secs: u64,
+    pub webhook_last_request: std::sync::atomic::AtomicU64,
+    pub webhook_tokens: std::sync::atomic::AtomicU32,
 }
 
 fn unix_now() -> u64 {
@@ -32,6 +34,9 @@ fn unix_now() -> u64 {
         .as_secs()
 }
 
+const WEBHOOK_RATE_LIMIT: u32 = 10;
+const WEBHOOK_RATE_WINDOW_SECS: u64 = 60;
+
 pub async fn post_webhook(
     State(state): State<Arc<AppState>>,
     headers: HeaderMap,
@@ -41,6 +46,22 @@ pub async fn post_webhook(
         return StatusCode::UNAUTHORIZED;
     }
 
+    // Simple rate limiting: reset tokens every window, reject if exhausted
+    let now = unix_now();
+    let last = state.webhook_last_request.load(std::sync::atomic::Ordering::Relaxed);
+    if now.saturating_sub(last) >= WEBHOOK_RATE_WINDOW_SECS {
+        state.webhook_tokens.store(WEBHOOK_RATE_LIMIT, std::sync::atomic::Ordering::Relaxed);
+        state.webhook_last_request.store(now, std::sync::atomic::Ordering::Relaxed);
+    }
+    let remaining = state.webhook_tokens.fetch_update(
+        std::sync::atomic::Ordering::Relaxed,
+        std::sync::atomic::Ordering::Relaxed,
+        |n| if n > 0 { Some(n - 1) } else { None },
+    );
+    if remaining.is_err() {
+        return StatusCode::TOO_MANY_REQUESTS;
+    }
+
     let Some(status) = DoorStatus::from_str(&body.status) else {
         return StatusCode::BAD_REQUEST;
     };
@@ -148,17 +169,16 @@ pub async fn get_image(State(state): State<Arc<AppState>>) -> impl IntoResponse
     let db = state.db.clone();
     let status = tokio::task::spawn_blocking(move || {
         let conn = db.blocking_lock();
-        db::get_current_status_str(&conn)
+        db::get_current_status(&conn)
     })
     .await
     .expect("db task panicked")
-    .ok()
-    .flatten();
+    .unwrap_or(DoorStatus::Offline);
 
-    let image = match status.as_deref() {
-        Some("open") => OPEN_PNG,
-        Some("closed") => CLOSED_PNG,
-        _ => OFFLINE_PNG,
+    let image = match status {
+        DoorStatus::Open => OPEN_PNG,
+        DoorStatus::Closed => CLOSED_PNG,
+        DoorStatus::Offline => OFFLINE_PNG,
     };
     ([(header::CONTENT_TYPE, "image/png"), (header::CACHE_CONTROL, "public, max-age=5")], image)
 }
@@ -167,17 +187,16 @@ pub async fn get_badge(State(state): State<Arc<AppState>>) -> impl IntoResponse
     let db = state.db.clone();
     let status = tokio::task::spawn_blocking(move || {
         let conn = db.blocking_lock();
-        db::get_current_status_str(&conn)
+        db::get_current_status(&conn)
     })
     .await
     .expect("db task panicked")
-    .ok()
-    .flatten();
+    .unwrap_or(DoorStatus::Offline);
 
-    let (label, color) = match status.as_deref() {
-        Some("open") => ("open", "#57f287"),
-        Some("closed") => ("closed", "#ed4245"),
-        _ => ("offline", "#99aab5"),
+    let (label, color) = match status {
+        DoorStatus::Open => ("open", "#57f287"),
+        DoorStatus::Closed => ("closed", "#ed4245"),
+        DoorStatus::Offline => ("offline", "#99aab5"),
     };
 
     let label_width = 70u32;

remote/cache-service/src/db.rs

@@ -12,7 +12,8 @@ pub fn init(path: &str) -> Result<Connection> {
             id INTEGER PRIMARY KEY CHECK (id = 1),
             status TEXT,
             timestamp INTEGER,
-            last_seen INTEGER
+            last_seen INTEGER,
+            last_checked INTEGER
         );
 
         CREATE TABLE IF NOT EXISTS state_log (
@@ -28,17 +29,29 @@ pub fn init(path: &str) -> Result<Connection> {
             fetched_at INTEGER NOT NULL
         );
 
-        INSERT OR IGNORE INTO current_state (id, status, timestamp, last_seen) VALUES (1, NULL, NULL, NULL);
+        INSERT OR IGNORE INTO current_state (id, status, timestamp, last_seen, last_checked) VALUES (1, 'offline', NULL, NULL, NULL);
         INSERT OR IGNORE INTO pi_info (id, data, fetched_at) VALUES (1, '{}', 0);
         ",
     )
     .context("failed to initialize database schema")?;
+
+    // Migration: add last_checked column if missing (existing databases)
+    let has_last_checked: bool = conn
+        .prepare("SELECT last_checked FROM current_state LIMIT 1")
+        .is_ok();
+    if !has_last_checked {
+        conn.execute_batch("ALTER TABLE current_state ADD COLUMN last_checked INTEGER")?;
+    }
+
+    // Migration: convert NULL status to 'offline'
+    conn.execute("UPDATE current_state SET status = 'offline' WHERE status IS NULL", [])?;
+
     Ok(conn)
 }
 
 pub fn get_status(conn: &Connection) -> Result<StatusResponse> {
-    let (status, timestamp, last_seen) = conn.query_row(
-        "SELECT status, timestamp, last_seen FROM current_state WHERE id = 1",
+    let (status_str, timestamp, last_checked) = conn.query_row(
+        "SELECT status, timestamp, last_checked FROM current_state WHERE id = 1",
         [],
         |row| {
             Ok((
@@ -48,10 +61,14 @@ pub fn get_status(conn: &Connection) -> Result<StatusResponse> {
             ))
         },
     )?;
+    let status = status_str
+        .as_deref()
+        .and_then(DoorStatus::from_str)
+        .unwrap_or(DoorStatus::Offline);
     Ok(StatusResponse {
-        status: status.unwrap_or_else(|| "offline".to_string()),
-        timestamp,
-        last_seen,
+        status,
+        since: timestamp,
+        last_checked,
     })
 }
 
@@ -59,7 +76,7 @@ pub fn update_state(conn: &Connection, status: DoorStatus, timestamp: u64, now:
     let status_str = status.as_str();
     conn.execute(
         "UPDATE current_state SET status = ?1, timestamp = ?2, last_seen = ?3 WHERE id = 1",
-        rusqlite::params![status_str, timestamp, now],
+        rusqlite::params![status_str, now, now],
     )?;
     conn.execute(
         "INSERT INTO state_log (status, timestamp, recorded_at) VALUES (?1, ?2, ?3)",
@@ -76,10 +93,18 @@ pub fn update_last_seen(conn: &Connection, now: u64) -> Result<()> {
     Ok(())
 }
 
+pub fn update_last_checked(conn: &Connection, now: u64) -> Result<()> {
+    conn.execute(
+        "UPDATE current_state SET last_checked = ?1 WHERE id = 1",
+        rusqlite::params![now],
+    )?;
+    Ok(())
+}
+
 pub fn mark_offline(conn: &Connection, now: u64) -> Result<()> {
     conn.execute(
-        "UPDATE current_state SET status = NULL WHERE id = 1",
-        [],
+        "UPDATE current_state SET status = 'offline', timestamp = ?1 WHERE id = 1",
+        rusqlite::params![now],
     )?;
     conn.execute(
         "INSERT INTO state_log (status, timestamp, recorded_at) VALUES ('offline', ?1, ?1)",
@@ -88,13 +113,16 @@ pub fn mark_offline(conn: &Connection, now: u64) -> Result<()> {
     Ok(())
 }
 
-pub fn get_current_status_str(conn: &Connection) -> Result<Option<String>> {
-    let status = conn.query_row(
+pub fn get_current_status(conn: &Connection) -> Result<DoorStatus> {
+    let status_str: Option<String> = conn.query_row(
         "SELECT status FROM current_state WHERE id = 1",
         [],
-        |row| row.get::<_, Option<String>>(0),
+        |row| row.get(0),
     )?;
-    Ok(status)
+    Ok(status_str
+        .as_deref()
+        .and_then(DoorStatus::from_str)
+        .unwrap_or(DoorStatus::Offline))
 }
 
 pub fn get_history(conn: &Connection, limit: u32) -> Result<Vec<noisebell_common::HistoryEntry>> {
@@ -131,3 +159,105 @@ pub fn update_pi_info(conn: &Connection, data: &serde_json::Value, now: u64) ->
     )?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_db() -> Connection {
+        init(":memory:").expect("failed to init test db")
+    }
+
+    #[test]
+    fn initial_status_is_offline() {
+        let conn = test_db();
+        let status = get_status(&conn).unwrap();
+        assert_eq!(status.status, DoorStatus::Offline);
+        assert!(status.since.is_none());
+        assert!(status.last_checked.is_none());
+    }
+
+    #[test]
+    fn update_state_changes_status() {
+        let conn = test_db();
+        update_state(&conn, DoorStatus::Open, 1000, 1001).unwrap();
+
+        let status = get_status(&conn).unwrap();
+        assert_eq!(status.status, DoorStatus::Open);
+        assert_eq!(status.since, Some(1001));
+    }
+
+    #[test]
+    fn mark_offline_sets_offline_status() {
+        let conn = test_db();
+        update_state(&conn, DoorStatus::Open, 1000, 1001).unwrap();
+        mark_offline(&conn, 2000).unwrap();
+
+        let status = get_status(&conn).unwrap();
+        assert_eq!(status.status, DoorStatus::Offline);
+        assert_eq!(status.since, Some(2000));
+    }
+
+    #[test]
+    fn get_current_status_matches_get_status() {
+        let conn = test_db();
+        assert_eq!(get_current_status(&conn).unwrap(), DoorStatus::Offline);
+
+        update_state(&conn, DoorStatus::Closed, 1000, 1001).unwrap();
+        assert_eq!(get_current_status(&conn).unwrap(), DoorStatus::Closed);
+    }
+
+    #[test]
+    fn update_last_checked_is_readable() {
+        let conn = test_db();
+        update_last_checked(&conn, 5000).unwrap();
+
+        let status = get_status(&conn).unwrap();
+        assert_eq!(status.last_checked, Some(5000));
+    }
+
+    #[test]
+    fn history_records_state_changes() {
+        let conn = test_db();
+        update_state(&conn, DoorStatus::Open, 1000, 1001).unwrap();
+        update_state(&conn, DoorStatus::Closed, 2000, 2001).unwrap();
+        mark_offline(&conn, 3000).unwrap();
+
+        let history = get_history(&conn, 10).unwrap();
+        assert_eq!(history.len(), 3);
+        assert_eq!(history[0].status, "offline");
+        assert_eq!(history[1].status, "closed");
+        assert_eq!(history[2].status, "open");
+    }
+
+    #[test]
+    fn status_response_serializes_correctly() {
+        let resp = StatusResponse {
+            status: DoorStatus::Open,
+            since: Some(1234),
+            last_checked: Some(5678),
+        };
+        let json = serde_json::to_value(&resp).unwrap();
+        assert_eq!(json["status"], "open");
+        assert_eq!(json["since"], 1234);
+        assert_eq!(json["last_checked"], 5678);
+    }
+
+    #[test]
+    fn null_status_migration_converts_to_offline() {
+        // Simulate an old database with NULL status
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute_batch("
+            CREATE TABLE current_state (id INTEGER PRIMARY KEY CHECK (id = 1), status TEXT, timestamp INTEGER, last_seen INTEGER);
+            INSERT INTO current_state (id, status, timestamp, last_seen) VALUES (1, NULL, NULL, NULL);
+            CREATE TABLE state_log (id INTEGER PRIMARY KEY AUTOINCREMENT, status TEXT NOT NULL, timestamp INTEGER NOT NULL, recorded_at INTEGER NOT NULL);
+            CREATE TABLE pi_info (id INTEGER PRIMARY KEY CHECK (id = 1), data TEXT NOT NULL, fetched_at INTEGER NOT NULL);
+            INSERT INTO pi_info (id, data, fetched_at) VALUES (1, '{}', 0);
+        ").unwrap();
+
+        // Re-init should migrate
+        let conn = init(":memory:").unwrap();
+        let status = get_current_status(&conn).unwrap();
+        assert_eq!(status, DoorStatus::Offline);
+    }
+}

remote/cache-service/src/main.rs

@@ -5,6 +5,7 @@ use anyhow::{Context, Result};
 use axum::routing::{get, post};
 use axum::Router;
 use tokio::sync::Mutex;
+use std::sync::atomic::AtomicU64;
 use tower_http::trace::TraceLayer;
 use tracing::{info, Level};
 
@@ -122,6 +123,8 @@ async fn main() -> Result<()> {
         webhooks,
         retry_attempts,
         retry_base_delay_secs,
+        webhook_last_request: AtomicU64::new(0),
+        webhook_tokens: std::sync::atomic::AtomicU32::new(10),
     });
 
     let app = Router::new()

remote/cache-service/src/poller.rs

@@ -37,6 +37,18 @@ pub fn spawn_status_poller(
         let mut was_offline = false;
 
         loop {
+            {
+                let now = unix_now();
+                let db = db.clone();
+                let _ = tokio::task::spawn_blocking(move || {
+                    let conn = db.blocking_lock();
+                    if let Err(e) = db::update_last_checked(&conn, now) {
+                        error!(error = %e, "failed to update last_checked");
+                    }
+                })
+                .await;
+            }
+
             let result = client
                 .get(format!("{}/", config.pi_address))
                 .bearer_auth(&config.pi_api_key)
@@ -65,10 +77,9 @@ pub fn spawn_status_poller(
 
                             if let Some(ref status_str) = status_str {
                                 if let Some(status) = DoorStatus::from_str(status_str) {
-                                    let current = db::get_current_status_str(&conn);
+                                    let current = db::get_current_status(&conn);
                                     let changed = match &current {
-                                        Ok(Some(s)) => s != status.as_str(),
-                                        Ok(None) => true,
+                                        Ok(current) => *current != status,
                                         Err(_) => true,
                                     };
                                     if changed {

remote/cache-service/src/types.rs

@@ -5,6 +5,7 @@ use serde::{Deserialize, Serialize};
 pub enum DoorStatus {
     Open,
     Closed,
+    Offline,
 }
 
 impl DoorStatus {
@@ -12,6 +13,7 @@ impl DoorStatus {
         match self {
             DoorStatus::Open => "open",
             DoorStatus::Closed => "closed",
+            DoorStatus::Offline => "offline",
         }
     }
 
@@ -19,6 +21,7 @@ impl DoorStatus {
         match s {
             "open" => Some(DoorStatus::Open),
             "closed" => Some(DoorStatus::Closed),
+            "offline" => Some(DoorStatus::Offline),
             _ => None,
         }
     }
@@ -26,9 +29,9 @@ impl DoorStatus {
 
 #[derive(Debug, Clone, Serialize)]
 pub struct StatusResponse {
-    pub status: String, // "open", "closed", or "offline"
-    pub timestamp: Option<u64>,
-    pub last_seen: Option<u64>,
+    pub status: DoorStatus,
+    pub since: Option<u64>,        // when the current status was set
+    pub last_checked: Option<u64>, // when the cache last attempted to poll
 }
 
 #[derive(Debug, Clone)]
@@ -36,3 +39,34 @@ pub struct WebhookTarget {
     pub url: String,
     pub secret: Option<String>,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn door_status_round_trip() {
+        for status in [DoorStatus::Open, DoorStatus::Closed, DoorStatus::Offline] {
+            let s = status.as_str();
+            assert_eq!(DoorStatus::from_str(s), Some(status));
+        }
+    }
+
+    #[test]
+    fn door_status_from_str_rejects_unknown() {
+        assert_eq!(DoorStatus::from_str("unknown"), None);
+        assert_eq!(DoorStatus::from_str(""), None);
+    }
+
+    #[test]
+    fn door_status_serde_lowercase() {
+        let json = serde_json::to_string(&DoorStatus::Open).unwrap();
+        assert_eq!(json, "\"open\"");
+
+        let deserialized: DoorStatus = serde_json::from_str("\"closed\"").unwrap();
+        assert_eq!(deserialized, DoorStatus::Closed);
+
+        let deserialized: DoorStatus = serde_json::from_str("\"offline\"").unwrap();
+        assert_eq!(deserialized, DoorStatus::Offline);
+    }
+}