rpki/src/tools/rpki_inter_rp_metrics.rs

use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::ccr::{CcrStateDigestComparison, compare_state_digests, decode_state_digest_summary};
use serde::Serialize;
use serde_json::{Value, json};

const RPS: [&str; 3] = ["ours-rp", "routinator", "rpki-client"];
const CCR_STATES: [&str; 5] = ["mfts", "vrps", "vaps", "tas", "rks"];

#[derive(Clone, Debug, PartialEq, Eq)]
struct Args {
    ours_run_root: PathBuf,
    peer_root: PathBuf,
    listen: String,
    poll_secs: u64,
    instance: String,
    once: bool,
    out_metrics: Option<PathBuf>,
    out_status: Option<PathBuf>,
}

fn usage() -> &'static str {
    "Usage: rpki_inter_rp_metrics --ours-run-root <path> --peer-root <path> [--listen <addr:port>] [--poll-secs <n>] [--instance <name>] [--once] [--out-metrics <path>] [--out-status <path>]"
}

pub fn main_entry() -> Result<(), String> {
    real_main()
}

fn real_main() -> Result<(), String> {
    let args = parse_args(&std::env::args().collect::<Vec<_>>())?;
    if args.once {
        let snapshot = scan(&args)?;
        let metrics = render_metrics(&snapshot);
        let status = render_status_json(&snapshot)?;
        if let Some(path) = args.out_metrics.as_ref() {
            write_file(path, metrics.as_bytes())?;
        } else {
            print!("{metrics}");
        }
        if let Some(path) = args.out_status.as_ref() {
            write_file(path, status.as_bytes())?;
        }
        return Ok(());
    }

    let shared = Arc::new(RwLock::new(scan(&args)?));
    let scanner = Arc::clone(&shared);
    let scan_args = args.clone();
    thread::spawn(move || {
        let poll_secs = scan_args.poll_secs.max(1);
        loop {
            thread::sleep(Duration::from_secs(poll_secs));
            match scan(&scan_args) {
                Ok(next) => *scanner.write().expect("metrics lock poisoned") = next,
                Err(err) => {
                    let mut previous = scanner.write().expect("metrics lock poisoned");
                    previous.service.last_reload_success = false;
                    previous.service.last_scan_timestamp_seconds = unix_now_seconds();
                    previous
                        .service
                        .parse_errors
                        .push(format!("scan failed: {err}"));
                }
            }
        }
    });

    serve_http(&args.listen, shared)
}

fn parse_args(argv: &[String]) -> Result<Args, String> {
    let mut ours_run_root = None;
    let mut peer_root = None;
    let mut listen = "127.0.0.1:9557".to_string();
    let mut poll_secs = 30u64;
    let mut instance = "inter-rp".to_string();
    let mut once = false;
    let mut out_metrics = None;
    let mut out_status = None;
    let mut index = 1usize;
    while index < argv.len() {
        match argv[index].as_str() {
            "--ours-run-root" => {
                index += 1;
                ours_run_root = Some(PathBuf::from(value_at(argv, index, "--ours-run-root")?));
            }
            "--peer-root" => {
                index += 1;
                peer_root = Some(PathBuf::from(value_at(argv, index, "--peer-root")?));
            }
            "--listen" => {
                index += 1;
                listen = value_at(argv, index, "--listen")?.to_string();
            }
            "--poll-secs" => {
                index += 1;
                let value = value_at(argv, index, "--poll-secs")?;
                poll_secs = value
                    .parse::<u64>()
                    .map_err(|_| format!("invalid --poll-secs: {value}"))?;
            }
            "--instance" => {
                index += 1;
                instance = value_at(argv, index, "--instance")?.to_string();
            }
            "--once" => once = true,
            "--out-metrics" => {
                index += 1;
                out_metrics = Some(PathBuf::from(value_at(argv, index, "--out-metrics")?));
            }
            "--out-status" => {
                index += 1;
                out_status = Some(PathBuf::from(value_at(argv, index, "--out-status")?));
            }
            "-h" | "--help" => return Err(usage().to_string()),
            other => return Err(format!("unknown argument: {other}\n{}", usage())),
        }
        index += 1;
    }
    Ok(Args {
        ours_run_root: ours_run_root
            .ok_or_else(|| format!("--ours-run-root is required\n{}", usage()))?,
        peer_root: peer_root.ok_or_else(|| format!("--peer-root is required\n{}", usage()))?,
        listen,
        poll_secs,
        instance,
        once,
        out_metrics,
        out_status,
    })
}

fn value_at<'a>(argv: &'a [String], index: usize, flag: &str) -> Result<&'a str, String> {
    argv.get(index)
        .map(|s| s.as_str())
        .ok_or_else(|| format!("{flag} requires a value"))
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct Snapshot {
    instance: String,
    service: ServiceMetrics,
    sync: SyncMetrics,
    samples: BTreeMap<String, RpSample>,
    ccr_compare: Option<CcrCompareMetrics>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct ServiceMetrics {
    last_scan_timestamp_seconds: f64,
    last_scan_duration_seconds: f64,
    last_reload_success: bool,
    parse_errors: Vec<String>,
    ours_run_root: String,
    peer_root: String,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct SyncMetrics {
    present: bool,
    success: bool,
    last_sync_timestamp_seconds: Option<f64>,
    age_seconds: Option<f64>,
    remote_host: Option<String>,
    message: Option<String>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RpSample {
    rp: String,
    present: bool,
    source_dir: Option<String>,
    run_id: Option<String>,
    run_seq: Option<u64>,
    success: bool,
    wall_seconds: Option<f64>,
    finish_timestamp_seconds: Option<f64>,
    artifact_age_seconds: Option<f64>,
    max_rss_bytes: BTreeMap<String, u64>,
    vrps: Option<u64>,
    vaps: Option<u64>,
    ccr_path: Option<String>,
    ccr: Option<CcrSampleMetrics>,
    parse_errors: Vec<String>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct CcrSampleMetrics {
    version: u32,
    hash_alg_oid: String,
    state_present: BTreeMap<String, bool>,
    state_hashes: BTreeMap<String, String>,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct CcrCompareMetrics {
    available: bool,
    match_all: bool,
    states: BTreeMap<String, CcrStateMetrics>,
    error: Option<String>,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct CcrStateMetrics {
    left_present: bool,
    right_present: bool,
    matches: bool,
    left_hash_hex: Option<String>,
    right_hash_hex: Option<String>,
}

fn scan(args: &Args) -> Result<Snapshot, String> {
    let start = Instant::now();
    let now = unix_now_seconds();
    let mut parse_errors = Vec::new();
    let mut samples = BTreeMap::new();

    let ours = scan_ours(&args.ours_run_root, now);
    parse_errors.extend(
        ours.parse_errors
            .iter()
            .map(|err| format!("ours-rp: {err}")),
    );
    samples.insert("ours-rp".to_string(), ours);

    for rp in ["routinator", "rpki-client"] {
        let peer = scan_peer(&args.peer_root, rp, now);
        parse_errors.extend(peer.parse_errors.iter().map(|err| format!("{rp}: {err}")));
        samples.insert(rp.to_string(), peer);
    }

    let sync = scan_sync_status(&args.peer_root, now);
    if let Some(message) = sync.message.as_ref().filter(|_| !sync.success) {
        parse_errors.push(format!("sync: {message}"));
    }

    let ccr_compare = build_ccr_compare(samples.get("ours-rp"), samples.get("rpki-client"));
    if let Some(compare) = ccr_compare
        .as_ref()
        .and_then(|compare| compare.error.as_ref())
    {
        parse_errors.push(format!("ccr compare: {compare}"));
    }

    Ok(Snapshot {
        instance: args.instance.clone(),
        service: ServiceMetrics {
            last_scan_timestamp_seconds: now,
            last_scan_duration_seconds: start.elapsed().as_secs_f64(),
            last_reload_success: parse_errors.is_empty(),
            parse_errors,
            ours_run_root: args.ours_run_root.display().to_string(),
            peer_root: args.peer_root.display().to_string(),
        },
        sync,
        samples,
        ccr_compare,
    })
}

fn scan_ours(run_root: &Path, now: f64) -> RpSample {
    let mut sample = RpSample {
        rp: "ours-rp".to_string(),
        ..RpSample::default()
    };
    let runs_root = if run_root.join("runs").is_dir() {
        run_root.join("runs")
    } else {
        run_root.to_path_buf()
    };
    let Some(run_dir) = latest_run_with_summary(&runs_root) else {
        sample.parse_errors.push(format!(
            "no run-summary.json found under {}",
            runs_root.display()
        ));
        return sample;
    };
    sample.present = true;
    sample.source_dir = Some(run_dir.display().to_string());

    let summary_path = run_dir.join("run-summary.json");
    let value = match read_json(&summary_path) {
        Ok(value) => value,
        Err(err) => {
            sample.parse_errors.push(err);
            return sample;
        }
    };
    sample.run_id = json_str(&value, &["runId"])
        .map(str::to_string)
        .or_else(|| {
            run_dir
                .file_name()
                .and_then(|v| v.to_str())
                .map(str::to_string)
        });
    sample.run_seq = json_u64(&value, &["runSeq"]).or_else(|| run_index_from_path(&run_dir));
    let status = json_str(&value, &["status"]).unwrap_or("unknown");
    sample.success = status == "success" && json_i64(&value, &["exitCode"]).unwrap_or(0) == 0;
    sample.wall_seconds = json_f64(&value, &["wallMs"]).map(|v| v / 1000.0);
    sample.finish_timestamp_seconds =
        json_str(&value, &["finishedAtRfc3339Utc"]).and_then(parse_rfc3339_to_unix);
    sample.artifact_age_seconds = sample
        .finish_timestamp_seconds
        .map(|finished| (now - finished).max(0.0));
    if let Some(max_rss_kb) = json_u64(&value, &["processMetrics", "maxRssKb"]) {
        sample
            .max_rss_bytes
            .insert("parent".to_string(), max_rss_kb.saturating_mul(1024));
        sample.max_rss_bytes.insert(
            "aggregate_peak".to_string(),
            max_rss_kb.saturating_mul(1024),
        );
    }
    sample.vrps =
        count_vrp_csv_unique_keys_opt(&run_dir.join("vrps.csv"), &mut sample.parse_errors)
            .or_else(|| json_u64(&value, &["reportCounts", "vrps"]));
    sample.vaps =
        count_vap_csv_unique_keys_opt(&run_dir.join("vaps.csv"), &mut sample.parse_errors)
            .or_else(|| json_u64(&value, &["reportCounts", "aspas"]));

    let ccr_path = run_dir.join("result.ccr");
    load_ccr_sample(&ccr_path, &mut sample);
    sample
}

fn scan_peer(peer_root: &Path, rp: &str, now: f64) -> RpSample {
    let mut sample = RpSample {
        rp: rp.to_string(),
        ..RpSample::default()
    };
    let rp_root = peer_root.join(rp);
    let latest = rp_root.join("latest");
    if !latest.exists() {
        sample
            .parse_errors
            .push(format!("missing latest directory: {}", latest.display()));
        return sample;
    }
    sample.present = true;
    sample.source_dir = Some(latest.display().to_string());

    let meta_path = ["run-meta.json", "meta.json"]
        .into_iter()
        .map(|name| latest.join(name))
        .find(|path| path.exists());
    let mut meta = Value::Null;
    if let Some(path) = meta_path.as_ref() {
        match read_json(path) {
            Ok(value) => meta = value,
            Err(err) => sample.parse_errors.push(err),
        }
    } else {
        sample
            .parse_errors
            .push(format!("missing run-meta.json under {}", latest.display()));
    }

    sample.run_id = json_str(&meta, &["runId"])
        .map(str::to_string)
        .or_else(|| json_str(&meta, &["run_id"]).map(str::to_string));
    sample.run_seq = json_u64(&meta, &["runSeq"]).or_else(|| json_u64(&meta, &["run_seq"]));
    sample.success = json_bool(&meta, &["success"]).unwrap_or(false);
    sample.wall_seconds = json_f64(&meta, &["wallSeconds"])
        .or_else(|| json_f64(&meta, &["wall_seconds"]))
        .or_else(|| json_f64(&meta, &["wallMs"]).map(|v| v / 1000.0))
        .or_else(|| json_f64(&meta, &["wall_ms"]).map(|v| v / 1000.0));
    sample.finish_timestamp_seconds = json_str(&meta, &["finishedAtRfc3339Utc"])
        .or_else(|| json_str(&meta, &["finished_at_rfc3339_utc"]))
        .and_then(parse_rfc3339_to_unix);
    sample.artifact_age_seconds = sample
        .finish_timestamp_seconds
        .map(|finished| (now - finished).max(0.0));
    read_rss_metric(&meta, &mut sample);

    let vrps_path = artifact_path(&latest, &meta, "vrpsCsv", "vrps.csv");
    let vaps_path = artifact_path(&latest, &meta, "vapsCsv", "vaps.csv");
    let ccr_path = artifact_path(&latest, &meta, "ccr", "result.ccr");
    load_ccr_sample(&ccr_path, &mut sample);

    sample.vrps = count_vrp_csv_unique_keys_opt(&vrps_path, &mut sample.parse_errors)
        .or_else(|| json_u64(&meta, &["counts", "vrps"]));
    sample.vaps = count_vap_csv_unique_keys_opt(&vaps_path, &mut sample.parse_errors)
        .or_else(|| json_u64(&meta, &["counts", "vaps"]))
        .or_else(|| json_u64(&meta, &["counts", "aspas"]));

    sample
}

fn read_rss_metric(meta: &Value, sample: &mut RpSample) {
    let fields = [
        ("parent", &["maxRssKb", "parent"][..]),
        ("child_max", &["maxRssKb", "childMax"][..]),
        ("aggregate_peak", &["maxRssKb", "aggregatePeak"][..]),
    ];
    for (name, path) in fields {
        if let Some(value) = json_u64(meta, path) {
            sample
                .max_rss_bytes
                .insert(name.to_string(), value.saturating_mul(1024));
        }
    }
    if sample.max_rss_bytes.is_empty() {
        if let Some(value) = json_u64(meta, &["maxRssKb"]) {
            sample
                .max_rss_bytes
                .insert("parent".to_string(), value.saturating_mul(1024));
            sample
                .max_rss_bytes
                .insert("aggregate_peak".to_string(), value.saturating_mul(1024));
        }
    }
}

fn artifact_path(run_dir: &Path, meta: &Value, key: &str, default_name: &str) -> PathBuf {
    json_str(meta, &["artifacts", key])
        .map(PathBuf::from)
        .filter(|path| !path.as_os_str().is_empty())
        .map(|path| {
            if path.is_absolute() {
                path
            } else {
                run_dir.join(path)
            }
        })
        .unwrap_or_else(|| run_dir.join(default_name))
}

fn load_ccr_sample(ccr_path: &Path, sample: &mut RpSample) {
    if !ccr_path.exists() {
        return;
    }
    sample.ccr_path = Some(ccr_path.display().to_string());
    let der = match fs::read(ccr_path) {
        Ok(der) => der,
        Err(err) => {
            sample
                .parse_errors
                .push(format!("read CCR failed: {}: {err}", ccr_path.display()));
            return;
        }
    };
    match decode_state_digest_summary(&der) {
        Ok(summary) => sample.ccr = Some(ccr_summary_to_metrics(summary)),
        Err(err) => sample.parse_errors.push(format!(
            "decode CCR digest failed: {}: {err}",
            ccr_path.display()
        )),
    }
}

fn ccr_summary_to_metrics(summary: crate::ccr::CcrStateDigestSummary) -> CcrSampleMetrics {
    let mut state_present = BTreeMap::new();
    let mut state_hashes = BTreeMap::new();
    for (name, hash) in [
        ("mfts", summary.mfts),
        ("vrps", summary.vrps),
        ("vaps", summary.vaps),
        ("tas", summary.tas),
        ("rks", summary.rks),
    ] {
        state_present.insert(name.to_string(), hash.is_some());
        if let Some(hash) = hash {
            state_hashes.insert(name.to_string(), hex::encode(hash));
        }
    }
    CcrSampleMetrics {
        version: summary.version,
        hash_alg_oid: summary.hash_alg_oid,
        state_present,
        state_hashes,
    }
}

fn scan_sync_status(peer_root: &Path, now: f64) -> SyncMetrics {
    let path = peer_root.join("sync-status.json");
    if !path.exists() {
        return SyncMetrics {
            present: false,
            success: false,
            message: Some(format!("missing {}", path.display())),
            ..SyncMetrics::default()
        };
    }
    let value = match read_json(&path) {
        Ok(value) => value,
        Err(err) => {
            return SyncMetrics {
                present: true,
                success: false,
                message: Some(err),
                ..SyncMetrics::default()
            };
        }
    };
    let timestamp = json_str(&value, &["lastSyncAtRfc3339Utc"])
        .or_else(|| json_str(&value, &["lastSyncTimestampRfc3339Utc"]))
        .or_else(|| json_str(&value, &["updatedAtRfc3339Utc"]))
        .and_then(parse_rfc3339_to_unix);
    SyncMetrics {
        present: true,
        success: json_bool(&value, &["success"])
            .or_else(|| json_bool(&value, &["lastSyncSuccess"]))
            .unwrap_or(false),
        last_sync_timestamp_seconds: timestamp,
        age_seconds: timestamp.map(|ts| (now - ts).max(0.0)),
        remote_host: json_str(&value, &["remoteHost"]).map(str::to_string),
        message: json_str(&value, &["message"])
            .or_else(|| json_str(&value, &["error"]))
            .map(str::to_string),
    }
}

fn build_ccr_compare(
    ours: Option<&RpSample>,
    rpki_client: Option<&RpSample>,
) -> Option<CcrCompareMetrics> {
    let left = ours?.ccr_path.as_ref()?;
    let right = rpki_client?.ccr_path.as_ref()?;
    let left = match fs::read(left) {
        Ok(value) => value,
        Err(err) => {
            return Some(CcrCompareMetrics {
                available: false,
                match_all: false,
                states: BTreeMap::new(),
                error: Some(format!("read ours CCR failed: {err}")),
            });
        }
    };
    let right = match fs::read(right) {
        Ok(value) => value,
        Err(err) => {
            return Some(CcrCompareMetrics {
                available: false,
                match_all: false,
                states: BTreeMap::new(),
                error: Some(format!("read rpki-client CCR failed: {err}")),
            });
        }
    };
    match compare_state_digests(&left, &right) {
        Ok(comparison) => Some(compare_to_metrics(comparison)),
        Err(err) => Some(CcrCompareMetrics {
            available: false,
            match_all: false,
            states: BTreeMap::new(),
            error: Some(err.to_string()),
        }),
    }
}

fn compare_to_metrics(comparison: CcrStateDigestComparison) -> CcrCompareMetrics {
    let states = comparison
        .states
        .iter()
        .map(|state| {
            (
                state.name.to_string(),
                CcrStateMetrics {
                    left_present: state.ours_present,
                    right_present: state.peer_present,
                    matches: state.matches,
                    left_hash_hex: state.ours_hash_hex.clone(),
                    right_hash_hex: state.peer_hash_hex.clone(),
                },
            )
        })
        .collect();
    CcrCompareMetrics {
        available: true,
        match_all: comparison.matches(),
        states,
        error: None,
    }
}

fn latest_run_with_summary(runs_root: &Path) -> Option<PathBuf> {
    let mut candidates = fs::read_dir(runs_root)
        .ok()?
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .filter(|path| path.is_dir() && path.join("run-summary.json").exists())
        .filter_map(|path| run_index_from_path(&path).map(|index| (index, path)))
        .collect::<Vec<_>>();
    candidates.sort_by_key(|(index, _)| *index);
    candidates.pop().map(|(_, path)| path)
}

fn count_vrp_csv_unique_keys_opt(path: &Path, errors: &mut Vec<String>) -> Option<u64> {
    if !path.exists() {
        return None;
    }
    match count_vrp_csv_unique_keys(path) {
        Ok(count) => Some(count),
        Err(err) => {
            errors.push(err);
            None
        }
    }
}

fn count_vap_csv_unique_keys_opt(path: &Path, errors: &mut Vec<String>) -> Option<u64> {
    if !path.exists() {
        return None;
    }
    match count_vap_csv_unique_keys(path) {
        Ok(count) => Some(count),
        Err(err) => {
            errors.push(err);
            None
        }
    }
}

fn count_vrp_csv_unique_keys(path: &Path) -> Result<u64, String> {
    let content = fs::read_to_string(path)
        .map_err(|e| format!("read VRP CSV failed: {}: {e}", path.display()))?;
    let mut unique = BTreeSet::new();
    for (index, line) in data_csv_lines(&content).enumerate() {
        if index == 0 {
            continue;
        }
        let columns = split_csv_simple(line);
        if columns.len() < 3 {
            return Err(format!(
                "invalid VRP CSV row in {}: expected at least 3 columns, got {}",
                path.display(),
                columns.len()
            ));
        }
        unique.insert((
            columns[0].to_string(),
            columns[1].to_string(),
            columns[2].to_string(),
        ));
    }
    Ok(unique.len() as u64)
}

fn count_vap_csv_unique_keys(path: &Path) -> Result<u64, String> {
    let content = fs::read_to_string(path)
        .map_err(|e| format!("read VAP CSV failed: {}: {e}", path.display()))?;
    let mut unique = BTreeSet::new();
    for (index, line) in data_csv_lines(&content).enumerate() {
        if index == 0 {
            continue;
        }
        let columns = split_csv_simple(line);
        if columns.len() < 2 {
            return Err(format!(
                "invalid VAP CSV row in {}: expected at least 2 columns, got {}",
                path.display(),
                columns.len()
            ));
        }
        unique.insert((columns[0].to_string(), columns[1].to_string()));
    }
    Ok(unique.len() as u64)
}

#[cfg(test)]
fn count_csv_rows(path: &Path) -> Result<u64, String> {
    let content = fs::read_to_string(path)
        .map_err(|e| format!("read CSV failed: {}: {e}", path.display()))?;
    let count = data_csv_lines(&content).skip(1).count();
    Ok(count as u64)
}

fn data_csv_lines(content: &str) -> impl Iterator<Item = &str> {
    content
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty())
        .filter(|line| !line.starts_with('#'))
}

fn split_csv_simple(line: &str) -> Vec<&str> {
    line.split(',').map(str::trim).collect()
}

fn render_metrics(snapshot: &Snapshot) -> String {
    let mut out = String::new();
    let mut writer = PromWriter::new(&mut out);
    let instance = snapshot.instance.as_str();
    writer.gauge(
        "inter_rp_service_up",
        "Inter-RP metrics service is up",
        &[label("instance", instance)],
        1.0,
    );
    writer.gauge(
        "inter_rp_service_last_scan_timestamp_seconds",
        "Last inter-RP scan timestamp",
        &[label("instance", instance)],
        snapshot.service.last_scan_timestamp_seconds,
    );
    writer.gauge(
        "inter_rp_service_last_scan_duration_seconds",
        "Last inter-RP scan duration",
        &[label("instance", instance)],
        snapshot.service.last_scan_duration_seconds,
    );
    writer.gauge(
        "inter_rp_service_last_reload_success",
        "Last inter-RP reload success",
        &[label("instance", instance)],
        bool_value(snapshot.service.last_reload_success),
    );
    writer.gauge(
        "inter_rp_parse_errors",
        "Current inter-RP parse error count",
        &[label("instance", instance)],
        snapshot.service.parse_errors.len() as f64,
    );
    writer.gauge(
        "inter_rp_sync_last_success",
        "Last remote200 artifact sync success",
        &[label("instance", instance)],
        bool_value(snapshot.sync.success),
    );
    if let Some(ts) = snapshot.sync.last_sync_timestamp_seconds {
        writer.gauge(
            "inter_rp_sync_last_timestamp_seconds",
            "Last remote200 artifact sync timestamp",
            &[label("instance", instance)],
            ts,
        );
    }
    if let Some(age) = snapshot.sync.age_seconds {
        writer.gauge(
            "inter_rp_sync_age_seconds",
            "Age of latest remote200 artifact sync",
            &[label("instance", instance)],
            age,
        );
    }

    for rp in RPS {
        let sample = snapshot.samples.get(rp);
        render_sample_metrics(&mut writer, instance, rp, sample);
    }
    render_count_diffs(&mut writer, instance, &snapshot.samples);
    render_ccr_compare_metrics(&mut writer, instance, snapshot.ccr_compare.as_ref());
    out
}

fn render_sample_metrics(
    writer: &mut PromWriter<'_>,
    instance: &str,
    rp: &str,
    sample: Option<&RpSample>,
) {
    let present = sample.map(|sample| sample.present).unwrap_or(false);
    let success = sample.map(|sample| sample.success).unwrap_or(false);
    writer.gauge(
        "inter_rp_run_present",
        "Whether latest RP run artifact is present",
        &[label("instance", instance), label("rp", rp)],
        bool_value(present),
    );
    writer.gauge(
        "inter_rp_run_success",
        "Whether latest RP run succeeded",
        &[label("instance", instance), label("rp", rp)],
        bool_value(success),
    );
    if let Some(sample) = sample {
        if let Some(seq) = sample.run_seq {
            writer.gauge(
                "inter_rp_run_seq",
                "Latest RP run sequence",
                &[label("instance", instance), label("rp", rp)],
                seq as f64,
            );
        }
        if let Some(wall) = sample.wall_seconds {
            writer.gauge(
                "inter_rp_run_wall_seconds",
                "Latest RP run wall clock seconds",
                &[label("instance", instance), label("rp", rp)],
                wall,
            );
        }
        if let Some(finish) = sample.finish_timestamp_seconds {
            writer.gauge(
                "inter_rp_run_finish_timestamp_seconds",
                "Latest RP run finish timestamp",
                &[label("instance", instance), label("rp", rp)],
                finish,
            );
        }
        if let Some(age) = sample.artifact_age_seconds {
            writer.gauge(
                "inter_rp_artifact_age_seconds",
                "Latest RP artifact age seconds",
                &[label("instance", instance), label("rp", rp)],
                age,
            );
        }
        for (kind, bytes) in &sample.max_rss_bytes {
            writer.gauge(
                "inter_rp_run_max_rss_bytes",
                "Latest RP run max RSS bytes",
                &[
                    label("instance", instance),
                    label("rp", rp),
                    label("kind", kind),
                ],
                *bytes as f64,
            );
        }
        if let Some(vrps) = sample.vrps {
            writer.gauge(
                "inter_rp_vrps",
                "Latest RP VRP count",
                &[label("instance", instance), label("rp", rp)],
                vrps as f64,
            );
        }
        if let Some(vaps) = sample.vaps {
            writer.gauge(
                "inter_rp_vaps",
                "Latest RP VAP/ASPA count",
                &[label("instance", instance), label("rp", rp)],
                vaps as f64,
            );
        }
        writer.gauge(
            "inter_rp_sample_parse_errors",
            "Current parse error count for an RP sample",
            &[label("instance", instance), label("rp", rp)],
            sample.parse_errors.len() as f64,
        );
        if let Some(ccr) = sample.ccr.as_ref() {
            writer.gauge(
                "inter_rp_ccr_version",
                "CCR version",
                &[label("instance", instance), label("rp", rp)],
                ccr.version as f64,
            );
            for state in CCR_STATES {
                writer.gauge(
                    "inter_rp_ccr_digest_present",
                    "CCR state digest presence",
                    &[
                        label("instance", instance),
                        label("rp", rp),
                        label("state", state),
                    ],
                    bool_value(ccr.state_present.get(state).copied().unwrap_or(false)),
                );
            }
        }
    }
}

fn render_count_diffs(
    writer: &mut PromWriter<'_>,
    instance: &str,
    samples: &BTreeMap<String, RpSample>,
) {
    for (left_index, left) in RPS.iter().enumerate() {
        for right in RPS.iter().skip(left_index + 1) {
            let left_sample = samples.get(*left);
            let right_sample = samples.get(*right);
            if let (Some(left_vrps), Some(right_vrps)) = (
                left_sample.and_then(|sample| sample.vrps),
                right_sample.and_then(|sample| sample.vrps),
            ) {
                writer.gauge(
                    "inter_rp_vrps_diff",
                    "Latest VRP count difference between two RPs; value is left minus right",
                    &[
                        label("instance", instance),
                        label("left", left),
                        label("right", right),
                    ],
                    left_vrps as f64 - right_vrps as f64,
                );
            }
            if let (Some(left_vaps), Some(right_vaps)) = (
                left_sample.and_then(|sample| sample.vaps),
                right_sample.and_then(|sample| sample.vaps),
            ) {
                writer.gauge(
                    "inter_rp_vaps_diff",
                    "Latest VAP/ASPA count difference between two RPs; value is left minus right",
                    &[
                        label("instance", instance),
                        label("left", left),
                        label("right", right),
                    ],
                    left_vaps as f64 - right_vaps as f64,
                );
            }
        }
    }
}

fn render_ccr_compare_metrics(
    writer: &mut PromWriter<'_>,
    instance: &str,
    compare: Option<&CcrCompareMetrics>,
) {
    let available = compare.map(|compare| compare.available).unwrap_or(false);
    let match_all = compare.map(|compare| compare.match_all).unwrap_or(false);
    writer.gauge(
        "inter_rp_ccr_digest_compare_available",
        "Whether ours-rp and rpki-client CCR digest comparison is available",
        &[
            label("instance", instance),
            label("left", "ours-rp"),
            label("right", "rpki-client"),
        ],
        bool_value(available),
    );
    writer.gauge(
        "inter_rp_ccr_digest_match",
        "Whether CCR state digest matches between ours-rp and rpki-client",
        &[
            label("instance", instance),
            label("left", "ours-rp"),
            label("right", "rpki-client"),
            label("state", "overall"),
        ],
        bool_value(match_all),
    );
    if let Some(compare) = compare {
        for state in CCR_STATES {
            let state_metrics = compare.states.get(state);
            writer.gauge(
                "inter_rp_ccr_digest_match",
                "Whether CCR state digest matches between ours-rp and rpki-client",
                &[
                    label("instance", instance),
                    label("left", "ours-rp"),
                    label("right", "rpki-client"),
                    label("state", state),
                ],
                bool_value(state_metrics.map(|state| state.matches).unwrap_or(false)),
            );
        }
    }
}

fn render_status_json(snapshot: &Snapshot) -> Result<String, String> {
    serde_json::to_string_pretty(&json!({
        "schemaVersion": 1,
        "generatedBy": "rpki_inter_rp_metrics",
        "instance": snapshot.instance,
        "service": snapshot.service,
        "sync": snapshot.sync,
        "samples": snapshot.samples,
        "ccrCompare": snapshot.ccr_compare,
    }))
    .map_err(|e| e.to_string())
}

#[derive(Clone, Debug)]
struct Label<'a> {
    key: &'a str,
    value: &'a str,
}

fn label<'a>(key: &'a str, value: &'a str) -> Label<'a> {
    Label { key, value }
}

struct PromWriter<'a> {
    out: &'a mut String,
    emitted_headers: BTreeSet<String>,
}

impl<'a> PromWriter<'a> {
    fn new(out: &'a mut String) -> Self {
        Self {
            out,
            emitted_headers: BTreeSet::new(),
        }
    }

    fn gauge(&mut self, name: &str, help: &str, labels: &[Label<'_>], value: f64) {
        self.header(name, help, "gauge");
        self.out.push_str(name);
        write_labels(self.out, labels);
        self.out.push(' ');
        self.out.push_str(&format_prom_value(value));
        self.out.push('\n');
    }

    fn header(&mut self, name: &str, help: &str, metric_type: &str) {
        if self.emitted_headers.insert(name.to_string()) {
            self.out.push_str("# HELP ");
            self.out.push_str(name);
            self.out.push(' ');
            self.out.push_str(&escape_help(help));
            self.out.push('\n');
            self.out.push_str("# TYPE ");
            self.out.push_str(name);
            self.out.push(' ');
            self.out.push_str(metric_type);
            self.out.push('\n');
        }
    }
}

fn write_labels(out: &mut String, labels: &[Label<'_>]) {
    if labels.is_empty() {
        return;
    }
    out.push('{');
    for (index, label) in labels.iter().enumerate() {
        if index > 0 {
            out.push(',');
        }
        out.push_str(label.key);
        out.push_str("=\"");
        out.push_str(&escape_label(label.value));
        out.push('"');
    }
    out.push('}');
}

fn serve_http(listen: &str, shared: Arc<RwLock<Snapshot>>) -> Result<(), String> {
    let listener = TcpListener::bind(listen).map_err(|e| format!("bind failed: {listen}: {e}"))?;
    for stream in listener.incoming() {
        match stream {
            Ok(mut stream) => {
                let snapshot = shared.read().expect("metrics lock poisoned").clone();
                if let Err(err) = handle_http_stream(&mut stream, &snapshot) {
                    eprintln!("http request failed: {err}");
                }
            }
            Err(err) => eprintln!("accept failed: {err}"),
        }
    }
    Ok(())
}

fn handle_http_stream(stream: &mut TcpStream, snapshot: &Snapshot) -> Result<(), String> {
    let mut buf = [0u8; 4096];
    let len = stream.read(&mut buf).map_err(|e| e.to_string())?;
    let req = String::from_utf8_lossy(&buf[..len]);
    let path = req
        .lines()
        .next()
        .and_then(|line| line.split_whitespace().nth(1))
        .unwrap_or("/");
    match path {
        "/metrics" => write_http_response(
            stream,
            "200 OK",
            "text/plain; version=0.0.4",
            &render_metrics(snapshot),
        ),
        "/status" => write_http_response(
            stream,
            "200 OK",
            "application/json",
            &render_status_json(snapshot)?,
        ),
        "/healthz" => write_http_response(stream, "200 OK", "text/plain", "ok\n"),
        _ => write_http_response(stream, "404 Not Found", "text/plain", "not found\n"),
    }
}

fn write_http_response(
    stream: &mut TcpStream,
    status: &str,
    content_type: &str,
    body: &str,
) -> Result<(), String> {
    let header = format!(
        "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        body.len()
    );
    stream
        .write_all(header.as_bytes())
        .and_then(|_| stream.write_all(body.as_bytes()))
        .map_err(|e| e.to_string())
}

fn read_json(path: &Path) -> Result<Value, String> {
    let bytes = fs::read(path).map_err(|e| format!("read json failed: {}: {e}", path.display()))?;
    serde_json::from_slice(&bytes)
        .map_err(|e| format!("parse json failed: {}: {e}", path.display()))
}

fn write_file(path: &Path, bytes: &[u8]) -> Result<(), String> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)
            .map_err(|e| format!("create parent dirs failed: {}: {e}", parent.display()))?;
    }
    fs::write(path, bytes).map_err(|e| format!("write file failed: {}: {e}", path.display()))
}

fn run_index_from_path(path: &Path) -> Option<u64> {
    path.file_name()
        .and_then(|name| name.to_str())
        .and_then(|name| name.strip_prefix("run_"))
        .and_then(|value| value.parse::<u64>().ok())
}

fn json_str<'a>(value: &'a Value, path: &[&str]) -> Option<&'a str> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_str()
}

fn json_u64(value: &Value, path: &[&str]) -> Option<u64> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_u64()
}

fn json_i64(value: &Value, path: &[&str]) -> Option<i64> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_i64()
}

fn json_f64(value: &Value, path: &[&str]) -> Option<f64> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current
        .as_f64()
        .or_else(|| current.as_i64().map(|value| value as f64))
        .or_else(|| current.as_u64().map(|value| value as f64))
}

fn json_bool(value: &Value, path: &[&str]) -> Option<bool> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_bool()
}

fn parse_rfc3339_to_unix(value: &str) -> Option<f64> {
    time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
        .ok()
        .map(|dt| dt.unix_timestamp() as f64)
}

fn unix_now_seconds() -> f64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs_f64())
        .unwrap_or(0.0)
}

fn bool_value(value: bool) -> f64 {
    if value { 1.0 } else { 0.0 }
}

fn format_prom_value(value: f64) -> String {
    if value.is_infinite() && value.is_sign_positive() {
        "+Inf".to_string()
    } else if value.fract() == 0.0 {
        format!("{value:.0}")
    } else {
        format!("{value:.6}")
            .trim_end_matches('0')
            .trim_end_matches('.')
            .to_string()
    }
}

fn escape_label(value: &str) -> String {
    value
        .replace('\\', "\\\\")
        .replace('\n', "\\n")
        .replace('"', "\\\"")
}

fn escape_help(value: &str) -> String {
    value.replace('\\', "\\\\").replace('\n', "\\n")
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ccr::{
        CcrContentInfo, CcrDigestAlgorithm, RpkiCanonicalCacheRepresentation,
        build_aspa_payload_state, build_roa_payload_state, encode_content_info,
    };
    use crate::data_model::roa::{IpPrefix, RoaAfi};
    use crate::validation::objects::{AspaAttestation, Vrp};
    use tempfile::TempDir;

    #[test]
    fn parse_args_accepts_required_paths() {
        let args = parse_args(&[
            "rpki_inter_rp_metrics".to_string(),
            "--ours-run-root".to_string(),
            "ours".to_string(),
            "--peer-root".to_string(),
            "peers".to_string(),
            "--once".to_string(),
        ])
        .expect("parse");
        assert_eq!(args.ours_run_root, PathBuf::from("ours"));
        assert_eq!(args.peer_root, PathBuf::from("peers"));
        assert!(args.once);
    }

    #[test]
    fn count_csv_rows_ignores_header_comments_and_blanks() {
        let temp = TempDir::new().expect("temp");
        let path = temp.path().join("vrps.csv");
        fs::write(
            &path,
            "# generated\nASN,IP Prefix,Max Length,Trust Anchor\nAS1,192.0.2.0/24,24,ta\n\n",
        )
        .expect("write");
        assert_eq!(count_csv_rows(&path).expect("count"), 1);
    }

    #[test]
    fn count_vrp_csv_uses_unique_vrp_keys() {
        let temp = TempDir::new().expect("temp");
        let path = temp.path().join("vrps.csv");
        fs::write(
            &path,
            "ASN,IP Prefix,Max Length,Trust Anchor,Expires\n\
             AS1,192.0.2.0/24,24,ta-a,100\n\
             AS1,192.0.2.0/24,24,ta-a,100\n\
             AS1,192.0.2.0/24,24,ta-b,200\n\
             AS2,198.51.100.0/24,24,ta-a,100\n",
        )
        .expect("write");
        assert_eq!(count_csv_rows(&path).expect("raw rows"), 4);
        assert_eq!(count_vrp_csv_unique_keys(&path).expect("unique vrps"), 2);
    }

    #[test]
    fn count_vap_csv_uses_unique_vap_keys() {
        let temp = TempDir::new().expect("temp");
        let path = temp.path().join("vaps.csv");
        fs::write(
            &path,
            "Customer ASN,Providers,Trust Anchor\n\
             AS64496,AS64497;AS64498,arin\n\
             AS64496,AS64497;AS64498,ripe\n\
             AS64496,AS64497,arin\n",
        )
        .expect("write");
        assert_eq!(count_csv_rows(&path).expect("raw rows"), 3);
        assert_eq!(count_vap_csv_unique_keys(&path).expect("unique vaps"), 2);
    }

    #[test]
    fn scan_fixture_renders_core_metrics_and_ccr_match() {
        let temp = TempDir::new().expect("temp");
        let ours_run = temp.path().join("ours/runs/run_0001");
        fs::create_dir_all(&ours_run).expect("mkdir");
        fs::write(
            ours_run.join("run-summary.json"),
            r#"{
              "runId":"run_0001",
              "runSeq":1,
              "status":"success",
              "exitCode":0,
              "wallMs":10000,
              "finishedAtRfc3339Utc":"2026-06-09T00:00:00Z",
              "processMetrics":{"maxRssKb":1000},
              "reportCounts":{"vrps":1,"aspas":1,"publicationPoints":1}
            }"#,
        )
        .expect("write summary");
        let ccr = sample_ccr(64496);
        fs::write(ours_run.join("result.ccr"), &ccr).expect("write ccr");

        let peer_root = temp.path().join("peers");
        let routinator = peer_root.join("routinator/latest");
        fs::create_dir_all(&routinator).expect("mkdir");
        fs::write(
            routinator.join("run-meta.json"),
            r#"{"runId":"run_0001","runSeq":1,"success":true,"wallMs":9000,"finishedAtRfc3339Utc":"2026-06-09T00:00:01Z","maxRssKb":{"aggregatePeak":900}}"#,
        )
        .expect("write meta");
        fs::write(
            routinator.join("vrps.csv"),
            "ASN,IP Prefix,Max Length,Trust Anchor\nAS64496,192.0.2.0/24,24,ta\n",
        )
        .expect("write vrps");
        fs::write(
            routinator.join("vaps.csv"),
            "Customer ASN,Providers,Trust Anchor\nAS64496,AS64497,ta\n",
        )
        .expect("write vaps");

        let rpki_client = peer_root.join("rpki-client/latest");
        fs::create_dir_all(&rpki_client).expect("mkdir");
        fs::write(
            rpki_client.join("run-meta.json"),
            r#"{"runId":"run_0001","runSeq":1,"success":true,"wallMs":8000,"finishedAtRfc3339Utc":"2026-06-09T00:00:02Z","maxRssKb":{"aggregatePeak":800},"artifacts":{"ccr":"result.ccr"},"counts":{"vrps":1,"vaps":1}}"#,
        )
        .expect("write meta");
        fs::write(rpki_client.join("result.ccr"), &ccr).expect("write ccr");
        fs::write(
            peer_root.join("sync-status.json"),
            r#"{"success":true,"lastSyncAtRfc3339Utc":"2026-06-09T00:00:03Z","remoteHost":"remote200"}"#,
        )
        .expect("write sync");

        let args = Args {
            ours_run_root: temp.path().join("ours"),
            peer_root,
            listen: "127.0.0.1:0".to_string(),
            poll_secs: 1,
            instance: "test".to_string(),
            once: true,
            out_metrics: None,
            out_status: None,
        };
        let snapshot = scan(&args).expect("scan");
        let metrics = render_metrics(&snapshot);
        assert!(metrics.contains("inter_rp_vrps{instance=\"test\",rp=\"ours-rp\"} 1"));
        assert!(metrics.contains("inter_rp_vaps{instance=\"test\",rp=\"rpki-client\"} 1"));
        assert!(metrics.contains(
            "inter_rp_ccr_digest_match{instance=\"test\",left=\"ours-rp\",right=\"rpki-client\",state=\"overall\"} 1"
        ));
        assert!(metrics.contains(
            "inter_rp_vrps_diff{instance=\"test\",left=\"routinator\",right=\"rpki-client\"} 0"
        ));
    }

    fn sample_ccr(customer_asn: u32) -> Vec<u8> {
        let vrps = build_roa_payload_state(&[Vrp {
            asn: customer_asn,
            prefix: IpPrefix {
                afi: RoaAfi::Ipv4,
                prefix_len: 24,
                addr: [192, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
            },
            max_length: 24,
        }])
        .expect("build vrps");
        let vaps = build_aspa_payload_state(&[AspaAttestation {
            customer_as_id: customer_asn,
            provider_as_ids: vec![64497],
        }])
        .expect("build vaps");
        let content = CcrContentInfo::new(RpkiCanonicalCacheRepresentation {
            version: 0,
            hash_alg: CcrDigestAlgorithm::Sha256,
            produced_at: time::OffsetDateTime::UNIX_EPOCH,
            mfts: None,
            vrps: Some(vrps),
            vaps: Some(vaps),
            tas: None,
            rks: None,
        });
        encode_content_info(&content).expect("encode")
    }
}