AkurAI Build
Menu

Public repository / d0b8d14afb524ab72435e5abd0bbd6049a205349

AkurAI-Monitor / src/schema.rs

Latest change d0b8d14afb524ab72435e5abd0bbd6049a205349 - Migrate storage layer from SQLite to MariaDB by Ólafur Búi Ólafsson

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

/// Create the full schema. Every statement is idempotent (`IF NOT EXISTS`),
/// so this is safe to run on every startup. MariaDB DDL is transactional per
/// statement; there is no legacy-SQLite migration path — the MariaDB schema
/// starts at this shape (2026-07-16 migration).
pub fn create_tables(conn: &mut impl Queryable) {
    const TABLES: &[&str] = &[
        "CREATE TABLE IF NOT EXISTS hosts (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(191) NOT NULL UNIQUE,
            address VARCHAR(255) NOT NULL,
            kind VARCHAR(32) NOT NULL DEFAULT 'linux',
            enabled TINYINT(1) NOT NULL DEFAULT 1,
            created_at BIGINT NOT NULL,
            updated_at BIGINT NOT NULL
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS applications (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            host_id BIGINT NOT NULL,
            name VARCHAR(191) NOT NULL,
            service_name VARCHAR(255) NULL,
            health_url TEXT NULL,
            enabled TINYINT(1) NOT NULL DEFAULT 1,
            app_version VARCHAR(64) NULL,
            framework VARCHAR(64) NULL,
            framework_version VARCHAR(64) NULL,
            versions_reported_at BIGINT NULL,
            created_at BIGINT NOT NULL,
            updated_at BIGINT NOT NULL,
            UNIQUE KEY uq_applications_host_name (host_id, name),
            CONSTRAINT fk_applications_host FOREIGN KEY (host_id)
                REFERENCES hosts(id) ON DELETE CASCADE
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS metrics (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            host_id BIGINT NULL,
            application_id BIGINT NULL,
            name VARCHAR(191) NOT NULL,
            value DOUBLE NOT NULL,
            ts BIGINT NOT NULL,
            INDEX idx_metrics_name_ts (name, ts),
            INDEX idx_metrics_host_ts (host_id, ts),
            CONSTRAINT fk_metrics_host FOREIGN KEY (host_id)
                REFERENCES hosts(id) ON DELETE SET NULL,
            CONSTRAINT fk_metrics_application FOREIGN KEY (application_id)
                REFERENCES applications(id) ON DELETE SET NULL
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS logs (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            host_id BIGINT NULL,
            application_id BIGINT NULL,
            source VARCHAR(191) NOT NULL,
            line MEDIUMTEXT NOT NULL,
            level VARCHAR(16) NULL,
            target VARCHAR(255) NULL,
            pid BIGINT NULL,
            origin VARCHAR(32) NULL,
            fields MEDIUMTEXT NULL,
            `cursor` VARCHAR(255) NULL,
            ts BIGINT NOT NULL,
            UNIQUE KEY uq_logs_cursor (`cursor`),
            INDEX idx_logs_source_ts (source, ts),
            INDEX idx_logs_host_ts (host_id, ts),
            INDEX idx_logs_level_ts (level, ts),
            INDEX idx_logs_app_ts (application_id, ts),
            CONSTRAINT fk_logs_host FOREIGN KEY (host_id)
                REFERENCES hosts(id) ON DELETE SET NULL,
            CONSTRAINT fk_logs_application FOREIGN KEY (application_id)
                REFERENCES applications(id) ON DELETE SET NULL
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS alert_rules (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            host_id BIGINT NULL,
            application_id BIGINT NULL,
            name VARCHAR(191) NOT NULL,
            metric_name VARCHAR(191) NOT NULL,
            operator VARCHAR(8) NOT NULL,
            threshold DOUBLE NOT NULL,
            duration_secs BIGINT NOT NULL,
            enabled TINYINT(1) NOT NULL DEFAULT 1,
            CONSTRAINT fk_alert_rules_host FOREIGN KEY (host_id)
                REFERENCES hosts(id) ON DELETE CASCADE,
            CONSTRAINT fk_alert_rules_application FOREIGN KEY (application_id)
                REFERENCES applications(id) ON DELETE CASCADE
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS alert_events (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            rule_id BIGINT NOT NULL,
            triggered_at BIGINT NOT NULL,
            resolved_at BIGINT NULL,
            value DOUBLE NOT NULL,
            CONSTRAINT fk_alert_events_rule FOREIGN KEY (rule_id)
                REFERENCES alert_rules(id) ON DELETE CASCADE
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS channels (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(191) NOT NULL UNIQUE,
            kind VARCHAR(32) NOT NULL,
            endpoint TEXT NOT NULL,
            token_env VARCHAR(191) NULL,
            enabled TINYINT(1) NOT NULL DEFAULT 1,
            created_at BIGINT NOT NULL,
            updated_at BIGINT NOT NULL
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS recipients (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            channel_id BIGINT NOT NULL,
            name VARCHAR(191) NOT NULL,
            target VARCHAR(255) NOT NULL,
            enabled TINYINT(1) NOT NULL DEFAULT 1,
            created_at BIGINT NOT NULL,
            updated_at BIGINT NOT NULL,
            UNIQUE KEY uq_recipients_channel_target (channel_id, target),
            CONSTRAINT fk_recipients_channel FOREIGN KEY (channel_id)
                REFERENCES channels(id) ON DELETE CASCADE
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS ingestion_state (
            source VARCHAR(191) PRIMARY KEY,
            kind VARCHAR(32) NOT NULL,
            last_cursor VARCHAR(255) NULL,
            last_ts BIGINT NULL,
            last_error TEXT NULL,
            last_error_ts BIGINT NULL,
            error_count BIGINT NOT NULL DEFAULT 0
        ) ENGINE=InnoDB",
        "CREATE TABLE IF NOT EXISTS current_metrics (
            host_id BIGINT NOT NULL,
            application_id BIGINT NOT NULL DEFAULT 0,
            name VARCHAR(191) NOT NULL,
            value DOUBLE NOT NULL,
            ts BIGINT NOT NULL,
            PRIMARY KEY (host_id, application_id, name),
            CONSTRAINT fk_current_metrics_host FOREIGN KEY (host_id)
                REFERENCES hosts(id) ON DELETE CASCADE
        ) ENGINE=InnoDB",
    ];
    for ddl in TABLES {
        conn.query_drop(*ddl).expect("failed to create table");
    }

    seed_local_host(conn);
    create_current_metrics_trigger(conn);
    create_application_host_triggers(conn);
}

/// Keep `current_metrics` as an upsert cache of the newest sample per
/// (host, application, metric). `VALUES(ts)` must be compared before the
/// `ts` column is overwritten — MariaDB applies SET clauses left to right.
fn create_current_metrics_trigger(conn: &mut impl Queryable) {
    conn.query_drop(
        "CREATE TRIGGER IF NOT EXISTS metrics_current_insert
         AFTER INSERT ON metrics FOR EACH ROW
         BEGIN
             IF NEW.host_id IS NOT NULL THEN
                 INSERT INTO current_metrics (host_id, application_id, name, value, ts)
                 VALUES (NEW.host_id, COALESCE(NEW.application_id, 0), NEW.name, NEW.value, NEW.ts)
                 ON DUPLICATE KEY UPDATE
                     value = IF(VALUES(ts) >= ts, VALUES(value), value),
                     ts = IF(VALUES(ts) >= ts, VALUES(ts), ts);
             END IF;
         END",
    )
    .expect("failed to create current metrics trigger");
}

fn create_application_host_triggers(conn: &mut impl Queryable) {
    for table in ["metrics", "logs", "alert_rules"] {
        for (suffix, event) in [("insert", "INSERT"), ("update", "UPDATE")] {
            conn.query_drop(format!(
                "CREATE TRIGGER IF NOT EXISTS {table}_application_host_{suffix}
                 BEFORE {event} ON {table} FOR EACH ROW
                 BEGIN
                     IF NEW.application_id IS NOT NULL AND (
                         NEW.host_id IS NULL
                         OR NOT EXISTS (
                             SELECT 1 FROM applications
                             WHERE id = NEW.application_id AND host_id = NEW.host_id
                         )
                     ) THEN
                         SIGNAL SQLSTATE '45000'
                             SET MESSAGE_TEXT = 'application does not belong to host';
                     END IF;
                 END"
            ))
            .expect("failed to create application ownership trigger");
        }
    }
    conn.query_drop(
        "CREATE TRIGGER IF NOT EXISTS applications_host_move_guard
         BEFORE UPDATE ON applications FOR EACH ROW
         BEGIN
             IF NEW.host_id != OLD.host_id AND (
                 EXISTS (SELECT 1 FROM metrics WHERE application_id = OLD.id)
                 OR EXISTS (SELECT 1 FROM logs WHERE application_id = OLD.id)
                 OR EXISTS (SELECT 1 FROM alert_rules WHERE application_id = OLD.id)
             ) THEN
                 SIGNAL SQLSTATE '45000'
                     SET MESSAGE_TEXT = 'application host cannot change while attributed data exists';
             END IF;
         END",
    )
    .expect("failed to create application move guard");
}

fn seed_local_host(conn: &mut impl Queryable) {
    let now = chrono::Utc::now().timestamp();
    conn.exec_drop(
        "INSERT IGNORE INTO hosts (name, address, kind, enabled, created_at, updated_at)
         VALUES ('local', '127.0.0.1', 'linux', 1, :now, :now)",
        params! { "now" => now },
    )
    .expect("failed to seed local host");
}

pub fn seed_alert_rules(conn: &mut impl Queryable) {
    let count: i64 = conn
        .query_first("SELECT COUNT(*) FROM alert_rules")
        .ok()
        .flatten()
        .unwrap_or(0);
    if count > 0 {
        return;
    }

    let host_id: i64 = conn
        .query_first("SELECT id FROM hosts WHERE name = 'local'")
        .expect("query local host")
        .expect("local host missing");
    let rules = [
        ("High CPU", "cpu.usage", "gt", 90.0, 300),
        ("High Memory", "mem.used_pct", "gt", 90.0, 300),
        ("Disk Almost Full", "disk.root.used_pct", "gt", 85.0, 60),
    ];

    for (name, metric, op, threshold, duration) in &rules {
        conn.exec_drop(
            "INSERT INTO alert_rules
             (host_id, name, metric_name, operator, threshold, duration_secs)
             VALUES (?, ?, ?, ?, ?, ?)",
            (host_id, name, metric, op, threshold, duration),
        )
        .expect("failed to seed alert rule");
    }

    tracing::info!("seeded {} default alert rules", rules.len());
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::testing::TestDb;
    use mysql::prelude::Queryable;

    #[test]
    fn creates_schema_idempotently_with_seeds_and_cache_trigger() {
        let Some(db) = TestDb::new() else { return };
        let mut conn = db.conn();
        create_tables(&mut conn);
        // Idempotent: a second run must not fail or duplicate seeds.
        create_tables(&mut conn);
        seed_alert_rules(&mut conn);

        let local_hosts: i64 = conn
            .query_first("SELECT COUNT(*) FROM hosts WHERE name = 'local'")
            .unwrap()
            .unwrap();
        assert_eq!(local_hosts, 1);

        let rule_count: i64 = conn
            .query_first("SELECT COUNT(*) FROM alert_rules")
            .unwrap()
            .unwrap();
        assert_eq!(rule_count, 3);

        // current_metrics cache follows newest inserts, ignores stale ones.
        let local: i64 = conn
            .query_first("SELECT id FROM hosts WHERE name = 'local'")
            .unwrap()
            .unwrap();
        conn.exec_drop(
            "INSERT INTO metrics (host_id, name, value, ts) VALUES (?, 'cpu.usage', 50, 10)",
            (local,),
        )
        .unwrap();
        conn.exec_drop(
            "INSERT INTO metrics (host_id, name, value, ts) VALUES (?, 'cpu.usage', 75, 20)",
            (local,),
        )
        .unwrap();
        conn.exec_drop(
            "INSERT INTO metrics (host_id, name, value, ts) VALUES (?, 'cpu.usage', 60, 15)",
            (local,),
        )
        .unwrap();
        let current: f64 = conn
            .query_first("SELECT value FROM current_metrics WHERE name = 'cpu.usage'")
            .unwrap()
            .unwrap();
        assert_eq!(current, 75.0);

        // Deleting a rule cascades to its events.
        conn.exec_drop(
            "INSERT INTO alert_events (rule_id, triggered_at, value)
             SELECT id, 1, 99 FROM alert_rules LIMIT 1",
            (),
        )
        .unwrap();
        conn.query_drop("DELETE FROM alert_rules").unwrap();
        let events: i64 = conn
            .query_first("SELECT COUNT(*) FROM alert_events")
            .unwrap()
            .unwrap();
        assert_eq!(events, 0);
    }

    #[test]
    fn rejects_application_attribution_to_the_wrong_host() {
        let Some(db) = TestDb::new() else { return };
        let mut conn = db.conn();
        create_tables(&mut conn);
        let now = chrono::Utc::now().timestamp();
        conn.exec_drop(
            "INSERT INTO hosts (name, address, kind, enabled, created_at, updated_at)
             VALUES ('remote', '100.88.0.8', 'linux', 1, ?, ?)",
            (now, now),
        )
        .unwrap();
        let remote_host = conn.last_insert_id() as i64;
        conn.exec_drop(
            "INSERT INTO applications (host_id, name, enabled, created_at, updated_at)
             VALUES (?, 'remote-app', 1, ?, ?)",
            (remote_host, now, now),
        )
        .unwrap();
        let application_id = conn.last_insert_id() as i64;
        let local_host: i64 = conn
            .query_first("SELECT id FROM hosts WHERE name = 'local'")
            .unwrap()
            .unwrap();

        let mismatch = conn.exec_drop(
            "INSERT INTO metrics (host_id, application_id, name, value, ts)
             VALUES (?, ?, 'health.ready', 1, ?)",
            (local_host, application_id, now),
        );
        assert!(mismatch.is_err());
        conn.exec_drop(
            "INSERT INTO metrics (host_id, application_id, name, value, ts)
             VALUES (?, ?, 'health.ready', 1, ?)",
            (remote_host, application_id, now),
        )
        .unwrap();

        let move_application = conn.exec_drop(
            "UPDATE applications SET host_id = ? WHERE id = ?",
            (local_host, application_id),
        );
        assert!(move_application.is_err());

        let ownership_triggers: i64 = conn
            .query_first(
                "SELECT COUNT(*) FROM information_schema.triggers
                 WHERE trigger_schema = DATABASE()
                   AND trigger_name LIKE '%_application_host_%'",
            )
            .unwrap()
            .unwrap();
        assert_eq!(ownership_triggers, 6);
    }
}