AkurAI-Monitor / src/version_poller.rs

Latest change 9a35bd9c668fae6e3383420387d1a0b1b4b3f524 - Read-only bearer token and host filter on /api/status by Ólafur Búi Ólafsson

// Background thread that polls application health endpoints for version info.
// Every 5 minutes, iterates enabled applications with a non-null health_url,
// fetches each, and updates the version columns.

use std::collections::HashSet;
use std::sync::{LazyLock, Mutex};
use std::time::Duration;

use crate::db;
use mysql::{params, prelude::Queryable};

/// Track which apps have failed so we only log once per failure streak.
static FAILED: LazyLock<Mutex<HashSet<i64>>> = LazyLock::new(|| Mutex::new(HashSet::new()));

pub fn spawn() {
    std::thread::spawn(move || {
        loop {
            std::thread::sleep(Duration::from_secs(300));

            let apps: Vec<(i64, String)> = db::with_db(|conn| {
                conn.query(
                    "SELECT id, health_url FROM applications
                     WHERE enabled = 1 AND health_url IS NOT NULL",
                )
                .unwrap_or_default()
            });

            if apps.is_empty() {
                continue;
            }

            let client = reqwest::blocking::Client::builder()
                .timeout(Duration::from_secs(10))
                .build()
                .unwrap();

            for (app_id, health_url) in &apps {
                match client.get(health_url).send() {
                    Ok(resp) => {
                        let status = resp.status();
                        let body: Result<serde_json::Value, _> = resp.json();
                        let now = chrono::Utc::now().timestamp();

                        match body {
                            Ok(ref json) if status.is_success() => {
                                let app_version = extract_field(json, "version");
                                let framework = extract_field(json, "framework");
                                let framework_version_val =
                                    extract_field(json, "framework_version");

                                db::with_db(|conn| {
                                    conn.exec_drop(
                                        "UPDATE applications SET
                                     app_version = :app_version,
                                     framework = :framework,
                                     framework_version = :framework_version,
                                     versions_reported_at = :now,
                                     updated_at = :now
                                     WHERE id = :id",
                                        params! {
                                            "app_version" => &app_version,
                                            "framework" => &framework,
                                            "framework_version" => &framework_version_val,
                                            "now" => now,
                                            "id" => app_id,
                                        },
                                    )
                                    .ok();
                                });
                                // Clear failure tracking on success
                                FAILED.lock().unwrap().remove(app_id);
                            }
                            _ => {
                                log_failure_once(
                                    *app_id,
                                    &format!("health poll {health_url}: HTTP {status}"),
                                );
                            }
                        }
                    }
                    Err(e) => {
                        log_failure_once(*app_id, &format!("health poll {health_url}: {e}"));
                    }
                }
            }
        }
    });
}

fn log_failure_once(app_id: i64, msg: &str) {
    let mut failed = FAILED.lock().unwrap();
    if failed.insert(app_id) {
        tracing::warn!(app_id, "{msg}");
    }
}

/// Reads a string field from the top level of a health body, falling back to `data.<key>`.
fn extract_field(json: &serde_json::Value, key: &str) -> Option<String> {
    json.get(key)
        .or_else(|| json.get("data").and_then(|d| d.get(key)))
        .and_then(|v| v.as_str())
        .map(String::from)
}

#[cfg(test)]
mod tests {
    use super::extract_field;
    use serde_json::json;

    #[test]
    fn reads_top_level() {
        assert_eq!(
            extract_field(&json!({"version":"1.0"}), "version").as_deref(),
            Some("1.0")
        );
    }

    #[test]
    fn reads_nested_data() {
        let body =
            json!({"ok":true,"data":{"version":"1.4.9","framework":"f","framework_version":"2"}});
        assert_eq!(extract_field(&body, "version").as_deref(), Some("1.4.9"));
        assert_eq!(
            extract_field(&body, "framework_version").as_deref(),
            Some("2")
        );
    }

    #[test]
    fn top_level_wins_and_missing_is_none() {
        let body = json!({"version":"top","data":{"version":"nested"}});
        assert_eq!(extract_field(&body, "version").as_deref(), Some("top"));
        assert_eq!(extract_field(&json!({"ok":true}), "version"), None);
    }
}