AkurAI-Monitor / src/db.rs

Latest change 0f9f642e4702d9b174c075dc1d32cd5bb60a63c4 - style: cargo fmt (AKURAI-MONITOR-8) by Ólafur Búi Ólafsson

use crate::config;
use crate::models::{DegradedState, HealthSummary, SourceFreshness, VersionFreshness};
use crate::schema;
use mysql::prelude::Queryable;
use mysql::{Opts, OptsBuilder, Pool, PoolConstraints, PoolOpts, PooledConn};
use std::sync::LazyLock;

static POOL: LazyLock<Pool> = LazyLock::new(|| {
    let cfg = config::get();
    let opts = Opts::from_url(&cfg.db_url).expect("invalid MONITOR_DB_URL");
    let constraints = PoolConstraints::new(1, 8).expect("valid pool constraints");
    let opts =
        OptsBuilder::from_opts(opts).pool_opts(PoolOpts::default().with_constraints(constraints));
    let pool = Pool::new(opts).expect("failed to create MariaDB pool");
    let mut conn = pool.get_conn().expect("failed to connect to MariaDB");
    schema::create_tables(&mut conn);
    schema::seed_alert_rules(&mut conn);
    pool
});

/// Run `f` with a pooled MariaDB connection. Connections are returned to the
/// pool on drop; the pool is bounded (1..=8) so DB memory stays flat.
pub fn with_db<F, R>(f: F) -> R
where
    F: FnOnce(&mut PooledConn) -> R,
{
    let mut conn = POOL.get_conn().expect("failed to get MariaDB connection");
    f(&mut conn)
}

/// Append exact-match tenant predicates and their positional values.
///
/// With no tenant context, no predicates are added so legacy callers keep
/// seeing all rows. Callers must use the same table alias in both arguments.
pub fn append_tenant_filters(
    sql: &mut String,
    params: &mut Vec<mysql::Value>,
    table_alias: &str,
    org_id: Option<&str>,
    workspace_id: Option<&str>,
) {
    if let Some(org_id) = org_id {
        sql.push_str(&format!(" AND {table_alias}.org_id = ?"));
        params.push(org_id.into());
    }
    if let Some(workspace_id) = workspace_id {
        sql.push_str(&format!(" AND {table_alias}.workspace_id = ?"));
        params.push(workspace_id.into());
    }
}

/// Compute tenant-isolated availability, backup age, version freshness, and
/// unresolved incident counts for the given window.
///
/// Every aggregate is scoped to exactly one (org_id, workspace_id) pair.
/// Missing data produces `None` / `"unknown"` — never a misleading zero.
pub fn query_tenant_health(org_id: &str, workspace_id: &str, window_secs: i64) -> HealthSummary {
    with_db(|conn| {
        let cutoff = chrono::Utc::now().timestamp() - window_secs;
        let now = chrono::Utc::now().timestamp();
        let mut degraded = Vec::new();

        // ── availability: up / total "up" metric samples ──
        let (total, up): (i64, i64) = conn
            .exec_first(
                "SELECT COUNT(*) AS total, \
                 COALESCE(SUM(CASE WHEN value >= 1.0 THEN 1 ELSE 0 END), 0) AS up \
                 FROM metrics \
                 WHERE name = 'up' \
                   AND org_id = ? AND workspace_id = ? \
                   AND ts >= ?",
                (org_id, workspace_id, cutoff),
            )
            .ok()
            .flatten()
            .unwrap_or((0, 0));

        let availability_pct = if total > 0 {
            Some((up as f64 / total as f64) * 100.0)
        } else {
            None
        };

        // Collect evidence ids for "up" metric degradation.
        let up_evidence: Vec<i64> = if total > 0 && up < total {
            conn.exec_map(
                "SELECT id FROM metrics \
                 WHERE name = 'up' \
                   AND org_id = ? AND workspace_id = ? \
                   AND ts >= ? AND value < 1.0 \
                 ORDER BY ts DESC LIMIT 5",
                (org_id, workspace_id, cutoff),
                |id: i64| id,
            )
            .unwrap_or_default()
        } else {
            Vec::new()
        };

        if let Some(pct) = availability_pct {
            if pct < 100.0 {
                degraded.push(DegradedState {
                    component: "availability".into(),
                    description: format!("{:.1}% up ({} / {})", pct, up, total),
                    evidence_ids: up_evidence,
                });
            }
        } else {
            degraded.push(DegradedState {
                component: "availability".into(),
                description: "no up metric found for this tenant".into(),
                evidence_ids: Vec::new(),
            });
        }

        // ── source freshness ──
        // Telemetry is collected every minute; five missed intervals makes the
        // summary unknown rather than accidentally presenting old data as healthy.
        const SOURCE_STALE_SECS: i64 = 5 * 60;
        let latest_sample: Option<(i64, i64)> = conn
            .exec_first(
                "SELECT id, ts FROM metrics \
                 WHERE org_id = ? AND workspace_id = ? \
                 ORDER BY ts DESC LIMIT 1",
                (org_id, workspace_id),
            )
            .ok()
            .flatten();
        let source_freshness = match latest_sample {
            Some((id, ts)) => {
                let age_seconds = (now - ts).max(0);
                if age_seconds > SOURCE_STALE_SECS {
                    degraded.push(DegradedState {
                        component: "source_freshness".into(),
                        description: format!(
                            "tenant telemetry is {}s old (stale > {}s)",
                            age_seconds, SOURCE_STALE_SECS
                        ),
                        evidence_ids: vec![id],
                    });
                    SourceFreshness {
                        status: "stale".into(),
                        last_observed_at: Some(ts),
                        age_seconds: Some(age_seconds),
                        evidence_ids: vec![id],
                    }
                } else {
                    SourceFreshness {
                        status: "fresh".into(),
                        last_observed_at: Some(ts),
                        age_seconds: Some(age_seconds),
                        evidence_ids: vec![id],
                    }
                }
            }
            None => {
                degraded.push(DegradedState {
                    component: "source_freshness".into(),
                    description: "no tenant telemetry has been received".into(),
                    evidence_ids: Vec::new(),
                });
                SourceFreshness {
                    status: "unknown".into(),
                    last_observed_at: None,
                    age_seconds: None,
                    evidence_ids: Vec::new(),
                }
            }
        };

        // ── last backup age ──
        let backup_sample: Option<(i64, f64, i64)> = conn
            .exec_first(
                "SELECT id, value, ts FROM metrics \
                 WHERE name = 'backup.age_seconds' \
                   AND org_id = ? AND workspace_id = ? \
                 ORDER BY ts DESC LIMIT 1",
                (org_id, workspace_id),
            )
            .ok()
            .flatten();
        let backup_age = backup_sample.map(|(_, value, _)| value as i64);
        let backup_fresh = match backup_sample {
            None => {
                degraded.push(DegradedState {
                    component: "backup".into(),
                    description: "no backup.age_seconds metric found".into(),
                    evidence_ids: Vec::new(),
                });
                false
            }
            Some((id, _, ts)) if now - ts > SOURCE_STALE_SECS => {
                degraded.push(DegradedState {
                    component: "backup".into(),
                    description: format!(
                        "backup telemetry is {}s old (stale > {}s)",
                        now - ts,
                        SOURCE_STALE_SECS
                    ),
                    evidence_ids: vec![id],
                });
                false
            }
            Some((id, age, _)) => {
                const BACKUP_STALE_SECS: i64 = 86400;
                if age as i64 > BACKUP_STALE_SECS {
                    degraded.push(DegradedState {
                        component: "backup".into(),
                        description: format!(
                            "last backup {}s ago (stale > {}s)",
                            age as i64, BACKUP_STALE_SECS
                        ),
                        evidence_ids: vec![id],
                    });
                }
                true
            }
        };

        // ── version freshness ──
        let framework_version = crate::framework_version::FRAMEWORK_VERSION;
        let apps: Vec<(i64, Option<String>)> = conn
            .exec_map(
                "SELECT DISTINCT a.id, a.framework_version FROM applications a \
                 JOIN metrics m ON m.application_id = a.id \
                 WHERE a.framework = 'AkurAI-Framework' AND a.enabled = 1 \
                   AND m.org_id = ? AND m.workspace_id = ? AND m.ts >= ?",
                (org_id, workspace_id, cutoff),
                |(id, fw): (i64, Option<String>)| (id, fw),
            )
            .unwrap_or_default();
        let app_count = apps.len() as i64;
        let outdated: Vec<i64> = apps
            .iter()
            .filter(|(_, fw)| fw.as_deref() != Some(framework_version))
            .map(|(id, _)| *id)
            .collect();
        let outdated_count = outdated.len() as i64;

        let version_freshness = if app_count == 0 {
            VersionFreshness {
                status: "unknown".into(),
                app_count: 0,
                outdated_count: 0,
                evidence_ids: Vec::new(),
            }
        } else if outdated_count > 0 {
            VersionFreshness {
                status: "outdated".into(),
                app_count,
                outdated_count,
                evidence_ids: outdated.clone(),
            }
        } else {
            VersionFreshness {
                status: "current".into(),
                app_count,
                outdated_count: 0,
                evidence_ids: Vec::new(),
            }
        };

        if version_freshness.status == "outdated" {
            degraded.push(DegradedState {
                component: "versions".into(),
                description: format!(
                    "{}/{} apps outdated (framework {})",
                    outdated_count, app_count, framework_version
                ),
                evidence_ids: outdated,
            });
        }

        // ── unresolved critical incidents ──
        let unresolved: i64 = conn
            .exec_first(
                "SELECT COUNT(*) FROM alert_events \
                 WHERE org_id = ? AND workspace_id = ? \
                   AND resolved_at IS NULL",
                (org_id, workspace_id),
            )
            .ok()
            .flatten()
            .unwrap_or(0);

        let incident_evidence: Vec<i64> = if unresolved > 0 {
            conn.exec_map(
                "SELECT id FROM alert_events \
                 WHERE org_id = ? AND workspace_id = ? \
                   AND resolved_at IS NULL \
                 ORDER BY triggered_at DESC LIMIT 20",
                (org_id, workspace_id),
                |id: i64| id,
            )
            .unwrap_or_default()
        } else {
            Vec::new()
        };

        if unresolved > 0 {
            degraded.push(DegradedState {
                component: "incidents".into(),
                description: format!("{} unresolved critical incidents", unresolved),
                evidence_ids: incident_evidence,
            });
        }

        // ── total sample count (any metric, scoped) ──
        let sample_count: i64 = conn
            .exec_first(
                "SELECT COUNT(*) FROM metrics \
                 WHERE org_id = ? AND workspace_id = ? \
                   AND ts >= ?",
                (org_id, workspace_id, cutoff),
            )
            .ok()
            .flatten()
            .unwrap_or(0);

        // Missing or stale availability, backup, or source telemetry is unknown,
        // never healthy; current violations remain an actionable degradation.
        let status =
            if source_freshness.status != "fresh" || availability_pct.is_none() || !backup_fresh {
                "unknown"
            } else if degraded.is_empty() {
                "healthy"
            } else {
                "degraded"
            };

        HealthSummary {
            org_id: org_id.to_string(),
            workspace_id: workspace_id.to_string(),
            status: status.to_string(),
            availability_pct,
            last_backup_age_seconds: backup_age,
            source_freshness,
            version_freshness,
            unresolved_critical_incidents: unresolved,
            degraded_states: degraded,
            sample_count,
            window_seconds: window_secs,
            computed_at: now,
        }
    })
}

/// Test support: scratch databases on the dev MariaDB instance.
///
/// Tests are skipped (with a note on stderr) unless `MONITOR_TEST_DB_URL`
/// points at a user allowed to CREATE/DROP `akurai_test_%` databases —
/// on Titan/EC2 that is the dev container on port 3307.
#[cfg(test)]
pub mod testing {
    use mysql::prelude::Queryable;
    use mysql::{Opts, OptsBuilder, Pool, PoolConstraints, PoolOpts, PooledConn};
    use std::sync::atomic::{AtomicU32, Ordering};

    static COUNTER: AtomicU32 = AtomicU32::new(0);

    pub struct TestDb {
        pool: Pool,
        admin: Pool,
        pub name: String,
    }

    impl TestDb {
        /// Returns `None` (after logging) when no test database is configured.
        pub fn new() -> Option<TestDb> {
            let Ok(url) = std::env::var("MONITOR_TEST_DB_URL") else {
                eprintln!("skipping: MONITOR_TEST_DB_URL is not set");
                return None;
            };
            let opts = Opts::from_url(&url).expect("invalid MONITOR_TEST_DB_URL");
            let constraints = PoolConstraints::new(1, 2).expect("valid pool constraints");
            let small = PoolOpts::default().with_constraints(constraints);
            let admin = Pool::new(OptsBuilder::from_opts(opts.clone()).pool_opts(small.clone()))
                .expect("connect test MariaDB");
            let name = format!(
                "akurai_test_mon_{}_{}",
                std::process::id(),
                COUNTER.fetch_add(1, Ordering::Relaxed)
            );
            admin
                .get_conn()
                .expect("admin conn")
                .query_drop(format!("CREATE DATABASE `{name}`"))
                .expect("create scratch database");
            let scoped = OptsBuilder::from_opts(opts)
                .db_name(Some(name.clone()))
                .pool_opts(small);
            let pool = Pool::new(scoped).expect("connect scratch database");
            Some(TestDb { pool, admin, name })
        }

        pub fn conn(&self) -> PooledConn {
            self.pool.get_conn().expect("scratch connection")
        }
    }

    impl Drop for TestDb {
        fn drop(&mut self) {
            if let Ok(mut conn) = self.admin.get_conn() {
                conn.query_drop(format!("DROP DATABASE IF EXISTS `{}`", self.name))
                    .ok();
            }
        }
    }
}