yah-tower 0.8.20

//! Integration tests for `tower::storage` — RuleStore load, reload, and watch.

use std::time::Duration;

use tower::storage::{RuleStore, SkipReason, WatchEvent};
use tower_rules::{RuleId, SchemaVersion};

// ── Fixtures ──────────────────────────────────────────────────────────────────

const QUORUM_YAML: &str = r#"
schema_version: V1
id: yubaba-quorum-loss
name: "Yubaba raft quorum loss"
predicate:
  type: scryer_health
  signal: yubaba_raft_quorum_lost
trigger:
  type: notification
  channels:
    - type: desktop_badge
  severity: critical
debounce_ms: 30000
federation:
  type: local_only
enabled: true
"#;

const BURST_YAML: &str = r#"
schema_version: V1
id: db-error-burst
name: "Database error burst"
predicate:
  type: event_match
  scope:
    type: service
    ident: "noisetable-db.pdx"
  level: error
  target:
    type: glob
    pattern: "database.*"
  rate:
    min_count: 5
    window_ms: 60000
trigger:
  type: notification
  channels:
    - type: desktop_badge
  severity: critical
federation:
  type: local_only
enabled: true
"#;

const FUTURE_VERSION_YAML: &str = r#"
schema_version: V99
id: future-rule
name: "Future rule"
predicate:
  type: scryer_health
  signal: ingestion_lag
trigger:
  type: notification
  channels:
    - type: desktop_badge
  severity: info
federation:
  type: local_only
enabled: true
"#;

const MALFORMED_YAML: &str = "predicate: {type: [}";

const INVALID_RULE_YAML: &str = r#"
schema_version: V1
id: BAD_ID
name: "Invalid ID"
predicate:
  type: scryer_health
  signal: ingestion_lag
trigger:
  type: notification
  channels:
    - type: desktop_badge
  severity: info
federation:
  type: local_only
enabled: true
"#;

const QUORUM_JSON: &str = r#"{
    "schema_version": "V1",
    "id": "quorum-json",
    "name": "Quorum JSON",
    "predicate": {"type": "scryer_health", "signal": "yubaba_raft_quorum_lost"},
    "trigger": {
        "type": "notification",
        "channels": [{"type": "desktop_badge"}],
        "severity": "critical"
    },
    "federation": {"type": "local_only"},
    "enabled": true
}"#;

// ── Load tests ────────────────────────────────────────────────────────────────

#[test]
fn load_from_empty_dir() {
    let dir = tempfile::tempdir().unwrap();
    let store = RuleStore::new(dir.path());
    let result = store.load_all();
    assert!(result.loaded.is_empty());
    assert!(result.skipped.is_empty());
    assert!(result.warnings.is_empty());
}

#[test]
fn load_from_nonexistent_dir() {
    let store = RuleStore::new("/nonexistent/rules/path");
    let result = store.load_all();
    assert!(result.loaded.is_empty());
    assert!(result.skipped.is_empty());
}

#[test]
fn load_single_yaml_file() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("yubaba-quorum-loss.yaml"), QUORUM_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert_eq!(result.loaded.len(), 1);
    assert!(result.skipped.is_empty());
    let rule = &result.loaded[0];
    assert_eq!(rule.id, RuleId("yubaba-quorum-loss".into()));
    assert_eq!(rule.schema_version, SchemaVersion::V1);
    assert!(rule.enabled);
}

#[test]
fn load_json_file() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("quorum-json.json"), QUORUM_JSON).unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert_eq!(result.loaded.len(), 1);
    assert_eq!(result.loaded[0].id, RuleId("quorum-json".into()));
}

#[test]
fn load_multiple_yaml_files() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("yubaba-quorum-loss.yaml"), QUORUM_YAML).unwrap();
    std::fs::write(dir.path().join("db-error-burst.yaml"), BURST_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert_eq!(result.loaded.len(), 2);
    assert!(result.skipped.is_empty());
    let mut ids: Vec<String> = result.loaded.iter().map(|r| r.id.0.clone()).collect();
    ids.sort();
    assert_eq!(ids, ["db-error-burst", "yubaba-quorum-loss"]);
}

#[test]
fn non_yaml_files_ignored() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("yubaba-quorum-loss.yaml"), QUORUM_YAML).unwrap();
    std::fs::write(dir.path().join("README.md"), "# rules").unwrap();
    std::fs::write(dir.path().join("notes.txt"), "notes").unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert_eq!(result.loaded.len(), 1);
}

#[test]
fn skip_future_schema_version() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("future-rule.yaml"), FUTURE_VERSION_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert!(result.loaded.is_empty());
    assert_eq!(result.skipped.len(), 1);
    assert!(
        matches!(&result.skipped[0].reason, SkipReason::FutureVersion { found } if found == "V99")
    );
}

#[test]
fn skip_malformed_yaml() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("bad.yaml"), MALFORMED_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert!(result.loaded.is_empty());
    assert_eq!(result.skipped.len(), 1);
    assert!(matches!(result.skipped[0].reason, SkipReason::ParseError(_)));
}

#[test]
fn skip_invalid_rule_yields_validation_error() {
    let dir = tempfile::tempdir().unwrap();
    // INVALID_RULE_YAML has bad ID and empty channels — validation should reject it
    std::fs::write(dir.path().join("bad-rule.yaml"), INVALID_RULE_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert!(result.loaded.is_empty());
    assert_eq!(result.skipped.len(), 1);
    assert!(matches!(result.skipped[0].reason, SkipReason::ValidationError(_)));
}

#[test]
fn load_file_direct() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("quorum.yaml");
    std::fs::write(&path, QUORUM_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let (rule, warns) = store.load_file(&path).unwrap();

    assert_eq!(rule.id, RuleId("yubaba-quorum-loss".into()));
    assert!(warns.is_empty());
}

#[test]
fn load_file_missing_version_treated_as_v1() {
    let dir = tempfile::tempdir().unwrap();
    // No schema_version field — should fall back to V1
    let no_version_yaml = r#"
id: no-version-rule
name: "No version"
predicate:
  type: scryer_health
  signal: ingestion_lag
trigger:
  type: notification
  channels:
    - type: desktop_badge
  severity: info
federation:
  type: local_only
enabled: true
"#;
    let path = dir.path().join("no-version-rule.yaml");
    std::fs::write(&path, no_version_yaml).unwrap();

    let store = RuleStore::new(dir.path());
    let (rule, _) = store.load_file(&path).unwrap();
    assert_eq!(rule.schema_version, SchemaVersion::V1);
}

#[test]
fn mixed_good_and_bad_files() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("good.yaml"), QUORUM_YAML).unwrap();
    std::fs::write(dir.path().join("bad.yaml"), MALFORMED_YAML).unwrap();
    std::fs::write(dir.path().join("future.yaml"), FUTURE_VERSION_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let result = store.load_all();

    assert_eq!(result.loaded.len(), 1);
    assert_eq!(result.skipped.len(), 2);
}

// ── Reload tests ──────────────────────────────────────────────────────────────

#[test]
fn reload_changed_file_produces_updated_rule() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("quorum.yaml");
    std::fs::write(&path, QUORUM_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let (rule_v1, _) = store.load_file(&path).unwrap();
    assert_eq!(rule_v1.debounce_ms, Some(30_000));

    // Simulate operator editing the rule
    let updated_yaml = QUORUM_YAML.replace("30000", "60000");
    std::fs::write(&path, &updated_yaml).unwrap();

    let (rule_v2, _) = store.load_file(&path).unwrap();
    assert_eq!(rule_v2.debounce_ms, Some(60_000));
}

#[test]
fn reload_with_future_version_returns_skip_reason() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("future.yaml");
    std::fs::write(&path, FUTURE_VERSION_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let err = store.load_file(&path).unwrap_err();
    assert!(matches!(err, SkipReason::FutureVersion { .. }));
}

// ── Watch tests ───────────────────────────────────────────────────────────────

#[test]
fn watch_detects_new_file() {
    let dir = tempfile::tempdir().unwrap();
    let store = RuleStore::new(dir.path());
    let (_watcher, rx) = store.watch().unwrap();

    // Write a new rule file after the watcher is live
    std::thread::sleep(Duration::from_millis(50)); // let watcher initialise
    std::fs::write(dir.path().join("quorum.yaml"), QUORUM_YAML).unwrap();

    let event = rx.recv_timeout(Duration::from_secs(5)).expect("timed out waiting for WatchEvent");
    assert!(
        matches!(event, WatchEvent::Updated { rule, .. } if rule.id == RuleId("yubaba-quorum-loss".into()))
    );
}

#[test]
fn watch_detects_file_modification() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("quorum.yaml");
    std::fs::write(&path, QUORUM_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let (_watcher, rx) = store.watch().unwrap();

    std::thread::sleep(Duration::from_millis(50)); // let watcher initialise

    let updated_yaml = QUORUM_YAML.replace("30000", "60000");
    std::fs::write(&path, &updated_yaml).unwrap();

    // Drain until we see the event we care about (notify may emit multiple events)
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    loop {
        let remaining = deadline.saturating_duration_since(std::time::Instant::now());
        if remaining.is_zero() {
            panic!("timed out waiting for WatchEvent::Updated with debounce_ms=60000");
        }
        match rx.recv_timeout(remaining) {
            Ok(WatchEvent::Updated { rule, .. }) if rule.debounce_ms == Some(60_000) => break,
            Ok(_) => continue, // spurious or initial event
            Err(_) => panic!("timed out waiting for WatchEvent::Updated with debounce_ms=60000"),
        }
    }
}

#[test]
fn watch_emits_error_for_malformed_file() {
    let dir = tempfile::tempdir().unwrap();
    let store = RuleStore::new(dir.path());
    let (_watcher, rx) = store.watch().unwrap();

    std::thread::sleep(Duration::from_millis(50));
    std::fs::write(dir.path().join("bad.yaml"), MALFORMED_YAML).unwrap();

    let event = rx.recv_timeout(Duration::from_secs(5)).expect("timed out");
    assert!(matches!(event, WatchEvent::Error { reason: SkipReason::ParseError(_), .. }));
}

#[test]
fn watch_emits_removed_on_delete() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("quorum.yaml");
    std::fs::write(&path, QUORUM_YAML).unwrap();

    let store = RuleStore::new(dir.path());
    let (_watcher, rx) = store.watch().unwrap();

    std::thread::sleep(Duration::from_millis(50));
    std::fs::remove_file(&path).unwrap();

    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    loop {
        let remaining = deadline.saturating_duration_since(std::time::Instant::now());
        if remaining.is_zero() {
            panic!("timed out waiting for WatchEvent::Removed");
        }
        match rx.recv_timeout(remaining) {
            Ok(WatchEvent::Removed { rule_id, .. }) => {
                assert_eq!(rule_id, RuleId("quorum".into()));
                break;
            }
            Ok(_) => continue,
            Err(_) => panic!("timed out waiting for WatchEvent::Removed"),
        }
    }
}