ottyl/src/engine/ingest.rs

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use crate::consts::{METRICS_POLL_INTERVAL, STDIN_POLL_INTERVAL as POLL_INTERVAL};
use crate::engine::timestamp::Timestamp;
use crate::engine::{EngineHandle, EngineMessage, IngestedLine, MetricSample};
use tokio::fs::File;
use tokio::io::{AsyncBufReadExt, BufReader};

pub async fn run(engine: EngineHandle) {
    let stdin = tokio::io::stdin();
    let mut lines = BufReader::new(stdin).lines();
    let mut sequence: u64 = 0;

    while let Ok(Some(text)) = lines.next_line().await {
        let line = IngestedLine {
            sequence,
            timestamp: Timestamp::now(),
            text,
        };
        sequence += 1;

        if engine.tx.send(EngineMessage::Ingest(line)).await.is_err() {
            break;
        }
    }
}

#[derive(PartialEq, Eq, Clone, Copy)]
struct FileId(#[cfg(unix)] u64, #[cfg(not(unix))] ());

fn file_id(meta: &std::fs::Metadata) -> FileId {
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        FileId(meta.ino())
    }
    #[cfg(not(unix))]
    {
        let _ = meta;
        FileId(())
    }
}

async fn open(path: &Path) -> Option<(File, FileId)> {
    let file = File::open(path).await.ok()?;
    let meta = file.metadata().await.ok()?;
    Some((file, file_id(&meta)))
}

pub async fn run_file(engine: EngineHandle, path: PathBuf) {
    let mut sequence: u64 = 0;

    'reopen: loop {
        let (file, mut id) = loop {
            if let Some(opened) = open(&path).await {
                break opened;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        };

        let mut reader = BufReader::new(file);
        let mut offset: u64 = 0;
        let mut pending = String::new();

        loop {
            let mut chunk = String::new();
            match reader.read_line(&mut chunk).await {
                Ok(0) => {
                    if let Ok(meta) = tokio::fs::metadata(&path).await {
                        let new_id = file_id(&meta);
                        if new_id != id || meta.len() < offset {
                            continue 'reopen;
                        }
                        id = new_id;
                    }
                    tokio::time::sleep(POLL_INTERVAL).await;
                }
                Ok(n) => {
                    offset += n as u64;
                    pending.push_str(&chunk);
                    if pending.ends_with('\n') {
                        let text = pending.trim_end_matches(['\n', '\r']).to_string();
                        pending.clear();

                        let line = IngestedLine {
                            sequence,
                            timestamp: Timestamp::now(),
                            text,
                        };
                        sequence += 1;

                        if engine.tx.send(EngineMessage::Ingest(line)).await.is_err() {
                            return;
                        }
                    }
                }
                Err(_) => continue 'reopen,
            }
        }
    }
}

async fn fetch_metrics_body(client: &reqwest::Client, url: &str) -> Result<String, String> {
    let response = client
        .get(url)
        .send()
        .await
        .map_err(|err| format!("warning: failed to poll metrics endpoint {url}: {err}"))?;
    let response = response
        .error_for_status()
        .map_err(|err| format!("warning: metrics endpoint {url} returned an error: {err}"))?;
    response
        .text()
        .await
        .map_err(|err| format!("warning: failed to read metrics response from {url}: {err}"))
}

pub async fn run_metrics_poll(engine: EngineHandle, url: String) {
    let client = reqwest::Client::new();
    loop {
        match fetch_metrics_body(&client, &url).await {
            Ok(body) => {
                let now = Timestamp::now();
                for sample in parse_openmetrics(&body, now) {
                    if engine
                        .tx
                        .send(EngineMessage::IngestMetricSample(sample))
                        .await
                        .is_err()
                    {
                        return;
                    }
                }
            }
            Err(warning) => eprintln!("{warning}"),
        }
        tokio::time::sleep(METRICS_POLL_INTERVAL).await;
    }
}

fn parse_openmetrics(body: &str, now: Timestamp) -> Vec<MetricSample> {
    body.lines()
        .filter_map(|line| parse_openmetrics_line(line, now))
        .collect()
}

fn parse_openmetrics_line(line: &str, now: Timestamp) -> Option<MetricSample> {
    let line = line.trim();
    if line.is_empty() || line.starts_with('#') {
        return None;
    }

    let (name_and_labels, rest) = line.split_once(char::is_whitespace)?;
    let mut fields = rest.split_whitespace();
    let value: f64 = fields.next()?.parse().ok()?;
    let timestamp = fields
        .next()
        .and_then(|raw| raw.parse::<f64>().ok())
        .map(|ms| Timestamp::from_millis(ms as i64))
        .unwrap_or(now);

    let (metric_name, labels) = match name_and_labels.find('{') {
        Some(open) => {
            let close = name_and_labels.rfind('}')?;
            (
                name_and_labels[..open].to_string(),
                parse_openmetrics_labels(&name_and_labels[open + 1..close]),
            )
        }
        None => (name_and_labels.to_string(), BTreeMap::new()),
    };
    if metric_name.is_empty() {
        return None;
    }

    Some(MetricSample {
        metric_name,
        labels,
        value,
        timestamp,
    })
}

fn parse_openmetrics_labels(raw: &str) -> BTreeMap<String, String> {
    raw.split(',')
        .filter_map(|pair| pair.split_once('='))
        .map(|(name, value)| {
            (
                name.trim().to_string(),
                value.trim().trim_matches('"').to_string(),
            )
        })
        .collect()
}

#[cfg(test)]
mod openmetrics_tests {
    use super::*;

    #[test]
    fn parses_bare_metric_without_labels_or_timestamp() {
        let now = Timestamp::from_millis(1000);
        let samples = parse_openmetrics("requests_total 42\n", now);
        assert_eq!(samples.len(), 1);
        assert_eq!(samples[0].metric_name, "requests_total");
        assert_eq!(samples[0].value, 42.0);
        assert!(samples[0].labels.is_empty());
        assert_eq!(samples[0].timestamp, now);
    }

    #[test]
    fn parses_labels_and_explicit_timestamp() {
        let now = Timestamp::from_millis(1000);
        let samples =
            parse_openmetrics(r#"http_requests_total{method="GET",path="/a"} 7 5000"#, now);
        assert_eq!(samples.len(), 1);
        let sample = &samples[0];
        assert_eq!(sample.metric_name, "http_requests_total");
        assert_eq!(sample.value, 7.0);
        assert_eq!(sample.labels.get("method").map(String::as_str), Some("GET"));
        assert_eq!(sample.labels.get("path").map(String::as_str), Some("/a"));
        assert_eq!(sample.timestamp, Timestamp::from_millis(5000));
    }

    #[test]
    fn skips_comments_blank_lines_and_malformed_lines() {
        let now = Timestamp::from_millis(0);
        let body = "# HELP requests_total docs\n# TYPE requests_total counter\n\nnot a valid line\nrequests_total 1\n";
        let samples = parse_openmetrics(body, now);
        assert_eq!(samples.len(), 1);
        assert_eq!(samples[0].metric_name, "requests_total");
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine::test_handle;
    use std::io::Write;
    use std::time::Duration;
    use tokio::sync::mpsc;

    async fn recv_text(rx: &mut mpsc::Receiver<EngineMessage>) -> String {
        match tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timed out waiting for ingested line")
            .expect("engine channel closed")
        {
            EngineMessage::Ingest(line) => line.text,
            _ => panic!("expected Ingest"),
        }
    }

    #[tokio::test]
    async fn tails_existing_and_appended_lines() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("app.log");
        std::fs::write(&path, "first\n").unwrap();

        let (tx, mut rx) = test_handle(16);
        tokio::spawn(run_file(tx, path.clone()));

        assert_eq!(recv_text(&mut rx).await, "first");

        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        writeln!(f, "second").unwrap();
        assert_eq!(recv_text(&mut rx).await, "second");
    }

    #[tokio::test]
    async fn resumes_after_delete_and_recreate() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("app.log");
        std::fs::write(&path, "before-restart\n").unwrap();

        let (tx, mut rx) = test_handle(16);
        tokio::spawn(run_file(tx, path.clone()));

        assert_eq!(recv_text(&mut rx).await, "before-restart");

        std::fs::remove_file(&path).unwrap();
        std::fs::write(&path, "after-restart\n").unwrap();

        assert_eq!(recv_text(&mut rx).await, "after-restart");
    }

    #[tokio::test]
    async fn resumes_after_truncation_in_place() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("app.log");
        std::fs::write(&path, "long line before truncation\n").unwrap();

        let (tx, mut rx) = test_handle(16);
        tokio::spawn(run_file(tx, path.clone()));

        assert_eq!(recv_text(&mut rx).await, "long line before truncation");

        let mut f = std::fs::OpenOptions::new()
            .write(true)
            .truncate(true)
            .open(&path)
            .unwrap();
        writeln!(f, "new").unwrap();

        assert_eq!(recv_text(&mut rx).await, "new");
    }

    #[tokio::test]
    async fn sequence_keeps_incrementing_across_recreation() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("app.log");
        std::fs::write(&path, "one\n").unwrap();

        let (tx, mut rx) = test_handle(16);
        tokio::spawn(run_file(tx, path.clone()));

        let first = match tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap()
        {
            EngineMessage::Ingest(line) => line.sequence,
            _ => panic!("expected Ingest"),
        };
        assert_eq!(first, 0);

        std::fs::remove_file(&path).unwrap();
        std::fs::write(&path, "two\n").unwrap();

        let second = match tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap()
        {
            EngineMessage::Ingest(line) => line.sequence,
            _ => panic!("expected Ingest"),
        };
        assert_eq!(second, 1);
    }
}