tower/

storage.rs

//! Rule storage: per-machine `.yah/cloud/rules/*.yaml` with config-file watcher.
//!
//! `RuleStore` loads all rule files from the rules directory at tower boot.
//! Schema versioning is enforced: future-version files are skipped with a warning;
//! older files (lower minor than the current `SchemaVersion`) migrate lazily.
//! Today that migration is a no-op since V1 is the only version.
//!
//! `RuleStore::watch` starts a `notify`-backed file watcher that emits
//! `WatchEvent`s whenever a rule file is created, modified, or removed. The
//! caller (the tower daemon) maps these events onto `Supervisor::update_rule`
//! and `Supervisor::disable_rule` calls per the arch doc §Rule storage.
//!
//! Architecture: `.yah/docs/architecture/A052-yah-tower.md`

use std::path::{Path, PathBuf};
use std::sync::mpsc;

use notify::{EventKind, RecursiveMode, Watcher};
use tower_rules::{RuleId, SchemaVersion, TowerRule};

use crate::rules::{
    parse::{parse_rule_json, parse_rule_yaml, ParseError},
    validate::{validate, RuleError, RuleWarning},
};

// ── Skip reason ───────────────────────────────────────────────────────────────

/// Why a rule file was not loaded.
#[derive(Debug, Clone, thiserror::Error)]
pub enum SkipReason {
    #[error("parse error: {0}")]
    ParseError(String),

    #[error("validation error: {0}")]
    ValidationError(String),

    /// The file declares a `schema_version` the running tower doesn't know about.
    /// Skipped instead of failing so a mixed-version fleet can coexist.
    #[error("future schema version '{found}' — upgrade tower to load this rule")]
    FutureVersion { found: String },
}

/// A rule file that was skipped during `load_all`.
#[derive(Debug)]
pub struct SkippedFile {
    pub path: PathBuf,
    pub reason: SkipReason,
}

// ── Load result ───────────────────────────────────────────────────────────────

/// Output of `RuleStore::load_all`.
pub struct LoadResult {
    /// Successfully parsed and validated rules.
    pub loaded: Vec<TowerRule>,
    /// Per-rule non-fatal warnings from semantic validation (empty IDs, etc.).
    pub warnings: Vec<(RuleId, Vec<RuleWarning>)>,
    /// Files that were present but skipped.
    pub skipped: Vec<SkippedFile>,
}

// ── Watch event ───────────────────────────────────────────────────────────────

/// Event emitted by the `RuleWatcher` background thread.
///
/// Tower maps these onto `Supervisor::update_rule` (Updated) and
/// `Supervisor::disable_rule` (Removed). On `Error`, log the reason and
/// keep the previously-loaded rule active.
#[derive(Debug)]
pub enum WatchEvent {
    /// A rule file was created or modified; the new parsed rule is included.
    Updated { path: PathBuf, rule: TowerRule },
    /// A rule file was removed.
    /// `rule_id` is derived from the file-stem (convention: filename == rule ID).
    Removed { path: PathBuf, rule_id: RuleId },
    /// A rule file changed but could not be reloaded.
    Error { path: PathBuf, reason: SkipReason },
}

// ── RuleWatcher ───────────────────────────────────────────────────────────────

/// Live config-file watcher for the rules directory.
///
/// Kept alive for as long as the caller holds this handle. Dropping it stops
/// the watcher and the background translation thread exits when its channel closes.
pub struct RuleWatcher {
    // Held to keep the notify watcher alive. Dropped together with this struct.
    _watcher: notify::RecommendedWatcher,
}

// ── RuleStore ─────────────────────────────────────────────────────────────────

/// Loader and watcher for per-machine tower rule files.
///
/// Rules live in `<rules_dir>/*.yaml` (or `.yml` / `.json`). The store is
/// stateless — call `load_all` at boot and `watch` to receive subsequent changes.
#[derive(Clone)]
pub struct RuleStore {
    rules_dir: PathBuf,
}

impl RuleStore {
    pub fn new(rules_dir: impl Into<PathBuf>) -> Self {
        Self { rules_dir: rules_dir.into() }
    }

    pub fn rules_dir(&self) -> &Path {
        &self.rules_dir
    }

    /// Load all rule files from the rules directory.
    ///
    /// Non-rule files (wrong extension) are silently ignored. Files that fail
    /// parsing, validation, or version checks are collected in `skipped`.
    /// I/O errors reading the directory are logged and an empty result returned.
    pub fn load_all(&self) -> LoadResult {
        let mut loaded = Vec::new();
        let mut warnings = Vec::new();
        let mut skipped = Vec::new();

        let entries = match std::fs::read_dir(&self.rules_dir) {
            Ok(it) => it,
            Err(_) => return LoadResult { loaded, warnings, skipped },
        };

        for entry in entries.flatten() {
            let path = entry.path();
            if !is_rule_file(&path) {
                continue;
            }
            match self.load_file(&path) {
                Ok((rule, warns)) => {
                    if !warns.is_empty() {
                        warnings.push((rule.id.clone(), warns));
                    }
                    loaded.push(rule);
                }
                Err(reason) => skipped.push(SkippedFile { path, reason }),
            }
        }

        LoadResult { loaded, warnings, skipped }
    }

    /// Parse and validate a single rule file.
    ///
    /// YAML and JSON are both accepted. The schema version is probed before
    /// full deserialisation so that future-version files produce a
    /// `SkipReason::FutureVersion` rather than a serde unknown-variant error.
    pub fn load_file(&self, path: &Path) -> Result<(TowerRule, Vec<RuleWarning>), SkipReason> {
        let src = std::fs::read_to_string(path)
            .map_err(|e| SkipReason::ParseError(e.to_string()))?;

        // Probe the schema version before full parse.
        let version = probe_schema_version(path, &src)?;
        let _ = version; // V1 is the only version; migration hook is a no-op.

        let rule = parse_src(path, &src).map_err(|e| SkipReason::ParseError(e.to_string()))?;

        let warns = validate(&rule).map_err(|e: RuleError| SkipReason::ValidationError(e.to_string()))?;

        Ok((rule, warns))
    }

    /// Start a config-file watcher on the rules directory.
    ///
    /// Returns a `RuleWatcher` handle (keep alive for as long as watching is
    /// desired) and a `mpsc::Receiver<WatchEvent>` for consuming events.
    ///
    /// The watcher translates notify events into `WatchEvent`s on a background
    /// thread. The thread exits when the watcher handle is dropped (channel close).
    pub fn watch(&self) -> Result<(RuleWatcher, mpsc::Receiver<WatchEvent>), notify::Error> {
        let (watch_tx, watch_rx) = mpsc::channel::<WatchEvent>();
        let (notify_tx, notify_rx) = mpsc::channel::<notify::Result<notify::Event>>();

        let mut watcher = notify::recommended_watcher(notify_tx)?;
        watcher.watch(&self.rules_dir, RecursiveMode::NonRecursive)?;

        let store = self.clone();
        // Explicit stack size (R330-S26): this thread previously relied on
        // the platform pthread default — 8 MiB on glibc, but only ~128 KiB
        // on musl. Pin it so a musl build doesn't silently shrink the stack
        // this loop runs on.
        std::thread::Builder::new()
            .name("tower-rule-watcher".into())
            .stack_size(8 * 1024 * 1024)
            .spawn(move || {
                for result in notify_rx {
                    let event = match result {
                        Ok(e) => e,
                        Err(e) => {
                            eprintln!("tower::storage: notify error: {e}");
                            continue;
                        }
                    };
                    for path in event.paths {
                        if !is_rule_file(&path) {
                            continue;
                        }
                        match event.kind {
                            EventKind::Create(_) | EventKind::Modify(_) => {
                                let ev = match store.load_file(&path) {
                                    Ok((rule, _)) => WatchEvent::Updated { path, rule },
                                    Err(reason) => WatchEvent::Error { path, reason },
                                };
                                let _ = watch_tx.send(ev);
                            }
                            EventKind::Remove(_) => {
                                let rule_id = rule_id_from_path(&path)
                                    .unwrap_or_else(|| RuleId("unknown".into()));
                                let _ = watch_tx.send(WatchEvent::Removed { path, rule_id });
                            }
                            _ => {}
                        }
                    }
                }
            })
            .expect("spawn tower-rule-watcher thread");

        Ok((RuleWatcher { _watcher: watcher }, watch_rx))
    }
}

// ── Helpers ───────────────────────────────────────────────────────────────────

/// Returns `true` if the path looks like a rule file (yaml, yml, or json).
fn is_rule_file(path: &Path) -> bool {
    matches!(
        path.extension().and_then(|e| e.to_str()),
        Some("yaml") | Some("yml") | Some("json")
    )
}

/// Derive a `RuleId` from the filename stem.
///
/// Convention: the stem is the rule ID (e.g. `yubaba-quorum-loss.yaml` → `yubaba-quorum-loss`).
/// Used for `WatchEvent::Removed` where the file no longer exists.
fn rule_id_from_path(path: &Path) -> Option<RuleId> {
    path.file_stem().and_then(|s| s.to_str()).map(|s| RuleId(s.to_string()))
}

/// Probe the `schema_version` field from raw YAML/JSON without deserialising
/// the full rule. Returns the known version on success, or a `SkipReason` if
/// the version is unrecognised or the file cannot even be parsed as YAML.
fn probe_schema_version(path: &Path, src: &str) -> Result<SchemaVersion, SkipReason> {
    // Parse as a generic map to read just the schema_version string.
    let raw: serde_yaml::Value = if is_json_path(path) {
        serde_json::from_str::<serde_json::Value>(src)
            .map_err(|e| SkipReason::ParseError(e.to_string()))
            .map(|v| serde_yaml::to_value(v).unwrap_or(serde_yaml::Value::Null))?
    } else {
        serde_yaml::from_str::<serde_yaml::Value>(src)
            .map_err(|e| SkipReason::ParseError(e.to_string()))?
    };

    let version_str = raw
        .get("schema_version")
        .and_then(|v| v.as_str())
        .unwrap_or("V1"); // missing version → treat as V1 for backward compat

    match version_str {
        "V1" => Ok(SchemaVersion::V1),
        other => Err(SkipReason::FutureVersion { found: other.to_string() }),
    }
}

fn parse_src(path: &Path, src: &str) -> Result<TowerRule, ParseError> {
    if is_json_path(path) { parse_rule_json(src) } else { parse_rule_yaml(src) }
}

fn is_json_path(path: &Path) -> bool {
    matches!(path.extension().and_then(|e| e.to_str()), Some("json"))
}