ottyl/src/ingest.rs

use std::path::{Path, PathBuf};
use std::time::Duration;

use crate::engine::{EngineMessage, IngestedLine};
use crate::timestamp::Timestamp;
use tokio::fs::File;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::sync::mpsc;

const POLL_INTERVAL: Duration = Duration::from_millis(200);

/// Reads stdin line by line, sequences and timestamps each line, and forwards it
/// to the engine task.
pub async fn run(engine: mpsc::Sender<EngineMessage>) {
    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.send(EngineMessage::Ingest(line)).await.is_err() {
            break;
        }
    }
}

/// Identifies a specific file instance on disk, so a recreated path (e.g. a
/// tailed process restarting and reopening its log file) can be told apart from
/// the same file simply growing.
#[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)))
}

/// Tails `path` like `tail -f`: reads existing content, then keeps reading as
/// new lines are appended. If the file is deleted and recreated at the same
/// path (or truncated in place) — e.g. the tailed process restarted and
/// reopened its log — tailing resumes against the new file from its start,
/// rather than exiting or getting stuck. `sequence` keeps incrementing across
/// a recreation event; it's ottyl's own line counter, not tied to the source
/// file.
pub async fn run_file(engine: mpsc::Sender<EngineMessage>, 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) => {
                    // EOF: decide whether the file was truncated/recreated
                    // (tail the new instance from scratch) or just hasn't
                    // grown yet (keep polling).
                    match tokio::fs::metadata(&path).await {
                        Ok(meta) => {
                            let new_id = file_id(&meta);
                            if new_id != id || meta.len() < offset {
                                continue 'reopen;
                            }
                            id = new_id;
                        }
                        Err(_) => {
                            // Path gone (mid-recreation); wait for it to
                            // reappear and reopen fresh.
                        }
                    }
                    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.send(EngineMessage::Ingest(line)).await.is_err() {
                            return;
                        }
                    }
                    // else: partial line at EOF (write in progress) — keep
                    // buffering and read again without treating it as final.
                }
                Err(_) => continue 'reopen,
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    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) = mpsc::channel(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) = mpsc::channel(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) = mpsc::channel(16);
        tokio::spawn(run_file(tx, path.clone()));

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

        // Truncate in place (same inode, shorter content) then write new data.
        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) = mpsc::channel(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);
    }
}