//! Finds the newest file in a directory and follows it, switching whenever a new file appears.
//!
//! The directory is polled rather than watched with inotify: it is one directory, the cost is a
//! `readdir` per interval, and polling behaves the same on every filesystem.
//!
//! A file counts as new when its path has not been seen before, when the path now names a
//! different inode, or when it has shrunk since the last scan. The last case catches a harness
//! that reruns a task and truncates the old log with `>` rather than creating a fresh file.

use std::collections::{HashMap, HashSet};
use std::fs::{self, File, Metadata};
use std::io::{self, Read, Seek, SeekFrom};
use std::mem;
use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};
use std::time::SystemTime;

/// The most bytes handed out by one poll, so a huge backlog cannot stall a front end.
const MAX_READ: u64 = 4 << 20;

#[derive(Clone)]
pub struct Options {
    pub dir: PathBuf,
    pub recursive: bool,
    /// Matched against the file name only, with `*` and `?` wildcards.
    pub glob: Option<String>,
}

/// Where to begin reading a file that is being opened.
#[derive(Clone, Copy, Debug)]
pub enum Start {
    Beginning,
    LastLines(usize),
}

#[derive(Debug, PartialEq)]
pub enum Event {
    /// Now following this file; any data that follows belongs to it.
    Switched(PathBuf),
    /// The followed file was truncated or replaced; its data starts over.
    Restarted(PathBuf),
    Data(Vec<u8>),
}

#[derive(Clone, Debug)]
pub struct Entry {
    pub path: PathBuf,
    /// When the file's current contents began: its creation time, or when it was seen to restart.
    pub started: SystemTime,
    id: (u64, u64),
    size: u64,
}

struct Current {
    path: PathBuf,
    id: (u64, u64),
    file: File,
    offset: u64,
}

pub struct Follower {
    opts: Options,
    initial: Start,
    /// What the previous scan saw; `None` until the first scan.
    known: Option<HashMap<PathBuf, Entry>>,
    /// Oldest first.
    entries: Vec<Entry>,
    current: Option<Current>,
    pinned: bool,
    pending: Vec<Event>,
}

impl Follower {
    /// `initial` is where to begin in the file that is newest at startup. Files that appear later
    /// are always read from the beginning.
    pub fn new(opts: Options, initial: Start) -> Self {
        Follower {
            opts,
            initial,
            known: None,
            entries: Vec::new(),
            current: None,
            pinned: false,
            pending: Vec::new(),
        }
    }

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

    #[cfg_attr(not(feature = "gui"), allow(dead_code))]
    pub fn options(&self) -> &Options {
        &self.opts
    }

    /// Files seen by the last scan, oldest first.
    pub fn entries(&self) -> &[Entry] {
        &self.entries
    }

    pub fn current_path(&self) -> Option<&Path> {
        self.current.as_ref().map(|c| c.path.as_path())
    }

    pub fn pinned(&self) -> bool {
        self.pinned
    }

    /// While pinned, new files do not take over from the current one.
    pub fn set_pinned(&mut self, pinned: bool) {
        self.pinned = pinned;
    }

    /// Follows `path` from now on. The `Switched` event arrives with the next poll.
    pub fn open(&mut self, path: &Path, start: Start) -> io::Result<()> {
        let mut file = File::open(path)?;
        let meta = file.metadata()?;
        let offset = match start {
            Start::Beginning => 0,
            Start::LastLines(n) => tail_offset(&mut file, meta.len(), n)?,
        };
        self.current = Some(Current {
            path: path.to_path_buf(),
            id: file_id(&meta),
            file,
            offset,
        });
        self.pending.push(Event::Switched(path.to_path_buf()));
        Ok(())
    }

    /// Rescans the directory, switches files if a new one appeared, and returns what happened
    /// along with any data written since the last poll.
    pub fn poll(&mut self) -> io::Result<Vec<Event>> {
        let mut entries = scan(&self.opts)?;
        let mut fresh = HashSet::new();
        if let Some(known) = &self.known {
            let now = SystemTime::now();
            for e in &mut entries {
                match known.get(&e.path) {
                    Some(k) if k.id == e.id && e.size >= k.size => {
                        e.started = k.started;
                        continue;
                    }
                    // Truncated in place keeps its creation time, so date the restart ourselves.
                    Some(k) if k.id == e.id => e.started = now,
                    // New, or replaced by another file: its creation time is right.
                    _ => {}
                }
                fresh.insert(e.path.clone());
            }
            entries.sort_by(|a, b| (a.started, &a.path).cmp(&(b.started, &b.path)));
        }
        let target = match &self.known {
            None => entries.last().map(|e| (e.path.clone(), self.initial)),
            Some(_) => entries
                .iter()
                .rev()
                .find(|e| fresh.contains(&e.path))
                .map(|e| (e.path.clone(), Start::Beginning)),
        };
        self.known = Some(
            entries
                .iter()
                .map(|e| (e.path.clone(), e.clone()))
                .collect(),
        );
        self.entries = entries;

        if let Some((path, start)) = target {
            // A restart of the current file is handled when it is read, below.
            let is_current = self.current_path() == Some(path.as_path());
            if !is_current && (!self.pinned || self.current.is_none()) {
                match self.open(&path, start) {
                    Ok(()) => {}
                    // Deleted between the scan and the open; the next scan will sort it out.
                    Err(e) if e.kind() == io::ErrorKind::NotFound => {}
                    Err(e) => return Err(e),
                }
            }
        }

        let mut events = mem::take(&mut self.pending);
        self.read_current(&mut events)?;
        Ok(events)
    }

    fn read_current(&mut self, events: &mut Vec<Event>) -> io::Result<()> {
        let Some(cur) = self.current.as_mut() else {
            return Ok(());
        };
        if let Ok(meta) = fs::metadata(&cur.path)
            && file_id(&meta) != cur.id
            && let Ok(file) = File::open(&cur.path)
        {
            cur.id = file_id(&file.metadata()?);
            cur.file = file;
            cur.offset = 0;
            events.push(Event::Restarted(cur.path.clone()));
        }
        let len = cur.file.metadata()?.len();
        if len < cur.offset {
            cur.offset = 0;
            events.push(Event::Restarted(cur.path.clone()));
        }
        if len > cur.offset {
            cur.file.seek(SeekFrom::Start(cur.offset))?;
            let mut buf = Vec::new();
            (&cur.file)
                .take((len - cur.offset).min(MAX_READ))
                .read_to_end(&mut buf)?;
            cur.offset += buf.len() as u64;
            if !buf.is_empty() {
                events.push(Event::Data(buf));
            }
        }
        Ok(())
    }
}

fn file_id(meta: &Metadata) -> (u64, u64) {
    (meta.dev(), meta.ino())
}

/// Lists the matching regular files under `opts.dir`, oldest first. A missing directory is
/// treated as empty so that the follower can be started before the harness creates it.
fn scan(opts: &Options) -> io::Result<Vec<Entry>> {
    let mut out = Vec::new();
    match walk(&opts.dir, opts, &mut out) {
        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
        other => other?,
    }
    out.sort_by(|a, b| (a.started, &a.path).cmp(&(b.started, &b.path)));
    Ok(out)
}

fn walk(dir: &Path, opts: &Options, out: &mut Vec<Entry>) -> io::Result<()> {
    for dirent in fs::read_dir(dir)? {
        let Ok(dirent) = dirent else { continue };
        let name = dirent.file_name();
        let name = name.to_string_lossy();
        if name.starts_with('.') {
            continue;
        }
        let path = dirent.path();
        let Ok(kind) = dirent.file_type() else {
            continue;
        };
        if kind.is_dir() {
            if opts.recursive {
                // Unreadable or vanished subdirectories are skipped, not fatal.
                let _ = walk(&path, opts, out);
            }
            continue;
        }
        if let Some(glob) = &opts.glob
            && !wildcard(glob.as_bytes(), name.as_bytes())
        {
            continue;
        }
        // Follows symlinks, so a `latest.log` link to a real file is still followed.
        let Ok(meta) = fs::metadata(&path) else {
            continue;
        };
        if !meta.is_file() {
            continue;
        }
        let created = meta
            .created()
            .or_else(|_| meta.modified())
            .unwrap_or(SystemTime::UNIX_EPOCH);
        out.push(Entry {
            path,
            started: created,
            id: file_id(&meta),
            size: meta.len(),
        });
    }
    Ok(())
}

/// Matches `s` against `pat`, where `*` matches any run of bytes and `?` any single byte.
pub fn wildcard(pat: &[u8], s: &[u8]) -> bool {
    let (mut p, mut i) = (0, 0);
    let mut star: Option<(usize, usize)> = None;
    while i < s.len() {
        if p < pat.len() && (pat[p] == b'?' || pat[p] == s[i]) {
            p += 1;
            i += 1;
        } else if p < pat.len() && pat[p] == b'*' {
            star = Some((p, i));
            p += 1;
        } else if let Some((sp, si)) = star {
            p = sp + 1;
            i = si + 1;
            star = Some((sp, si + 1));
        } else {
            return false;
        }
    }
    pat[p..].iter().all(|&c| c == b'*')
}

/// The offset at which the last `n` lines of a file of length `len` begin. A final newline does
/// not count as starting an empty last line.
fn tail_offset(file: &mut File, len: u64, n: usize) -> io::Result<u64> {
    if n == 0 {
        return Ok(len);
    }
    let mut buf = vec![0u8; 64 * 1024];
    let mut pos = len;
    let mut seen = 0;
    while pos > 0 {
        let chunk = pos.min(buf.len() as u64);
        pos -= chunk;
        file.seek(SeekFrom::Start(pos))?;
        file.read_exact(&mut buf[..chunk as usize])?;
        for i in (0..chunk as usize).rev() {
            let at = pos + i as u64;
            if buf[i] == b'\n' && at != len - 1 {
                seen += 1;
                if seen == n {
                    return Ok(at + 1);
                }
            }
        }
    }
    Ok(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs::OpenOptions;
    use std::io::Write;
    use std::thread;
    use std::time::Duration;

    fn follower(dir: &Path, initial: Start) -> Follower {
        Follower::new(
            Options {
                dir: dir.to_path_buf(),
                recursive: false,
                glob: None,
            },
            initial,
        )
    }

    /// Creation times need to differ for "newest" to mean anything.
    fn write_new(path: &Path, text: &str) {
        thread::sleep(Duration::from_millis(15));
        fs::write(path, text).unwrap();
    }

    fn append(path: &Path, text: &str) {
        let mut f = OpenOptions::new().append(true).open(path).unwrap();
        f.write_all(text.as_bytes()).unwrap();
    }

    fn data(events: &[Event]) -> String {
        events
            .iter()
            .filter_map(|e| match e {
                Event::Data(d) => Some(String::from_utf8_lossy(d).into_owned()),
                _ => None,
            })
            .collect()
    }

    #[test]
    fn starts_on_newest_file_with_last_lines() {
        let tmp = tempfile::tempdir().unwrap();
        write_new(&tmp.path().join("00-a.log"), "old\n");
        write_new(&tmp.path().join("01-b.log"), "1\n2\n3\n4\n");
        let mut f = follower(tmp.path(), Start::LastLines(2));
        let ev = f.poll().unwrap();
        assert_eq!(ev[0], Event::Switched(tmp.path().join("01-b.log")));
        assert_eq!(data(&ev), "3\n4\n");
    }

    #[test]
    fn follows_appended_data() {
        let tmp = tempfile::tempdir().unwrap();
        let p = tmp.path().join("a.log");
        write_new(&p, "one\n");
        let mut f = follower(tmp.path(), Start::Beginning);
        f.poll().unwrap();
        assert!(f.poll().unwrap().is_empty());
        append(&p, "two\npart");
        assert_eq!(data(&f.poll().unwrap()), "two\npart");
    }

    #[test]
    fn switches_to_a_new_file_from_its_beginning() {
        let tmp = tempfile::tempdir().unwrap();
        write_new(&tmp.path().join("00.log"), "a\n");
        let mut f = follower(tmp.path(), Start::LastLines(0));
        f.poll().unwrap();
        let p = tmp.path().join("01.log");
        write_new(&p, "first\nsecond\n");
        let ev = f.poll().unwrap();
        assert_eq!(ev[0], Event::Switched(p));
        assert_eq!(data(&ev), "first\nsecond\n");
    }

    #[test]
    fn waits_for_a_directory_that_does_not_exist_yet() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("runs/M2a");
        let mut f = follower(&dir, Start::LastLines(10));
        assert!(f.poll().unwrap().is_empty());
        fs::create_dir_all(&dir).unwrap();
        write_new(&dir.join("00.log"), "hello\n");
        let ev = f.poll().unwrap();
        assert_eq!(ev[0], Event::Switched(dir.join("00.log")));
        assert_eq!(data(&ev), "hello\n");
    }

    #[test]
    fn restarts_when_the_current_file_is_truncated() {
        let tmp = tempfile::tempdir().unwrap();
        let p = tmp.path().join("a.log");
        write_new(&p, "a long first run\n");
        let mut f = follower(tmp.path(), Start::Beginning);
        f.poll().unwrap();
        fs::write(&p, "rerun\n").unwrap();
        let ev = f.poll().unwrap();
        assert_eq!(ev[0], Event::Restarted(p));
        assert_eq!(data(&ev), "rerun\n");
    }

    #[test]
    fn restarts_when_the_current_file_is_replaced() {
        let tmp = tempfile::tempdir().unwrap();
        let p = tmp.path().join("a.log");
        write_new(&p, "one\n");
        let mut f = follower(tmp.path(), Start::Beginning);
        f.poll().unwrap();
        let q = tmp.path().join(".tmp");
        fs::write(&q, "a whole new file\n").unwrap();
        fs::rename(&q, &p).unwrap();
        let ev = f.poll().unwrap();
        assert_eq!(ev[0], Event::Restarted(p));
        assert_eq!(data(&ev), "a whole new file\n");
    }

    #[test]
    fn switches_to_an_older_file_that_is_rewritten() {
        let tmp = tempfile::tempdir().unwrap();
        let old = tmp.path().join("00.log");
        write_new(&old, "a long first run of task zero\n");
        write_new(&tmp.path().join("01.log"), "b\n");
        let mut f = follower(tmp.path(), Start::Beginning);
        f.poll().unwrap();
        fs::write(&old, "rerun\n").unwrap();
        let ev = f.poll().unwrap();
        assert_eq!(ev[0], Event::Switched(old.clone()));
        assert_eq!(data(&ev), "rerun\n");
        // It now ranks as the newest file, and keeps that rank on later scans.
        assert_eq!(f.entries().last().unwrap().path, old);
        append(&old, "more\n");
        f.poll().unwrap();
        assert_eq!(f.entries().last().unwrap().path, old);
    }

    #[test]
    fn appends_to_other_files_do_not_switch() {
        let tmp = tempfile::tempdir().unwrap();
        let old = tmp.path().join("00.log");
        write_new(&old, "a\n");
        write_new(&tmp.path().join("01.log"), "b\n");
        let mut f = follower(tmp.path(), Start::Beginning);
        f.poll().unwrap();
        append(&old, "more\n");
        assert!(f.poll().unwrap().is_empty());
    }

    #[test]
    fn pinned_stays_put() {
        let tmp = tempfile::tempdir().unwrap();
        let a = tmp.path().join("00.log");
        write_new(&a, "a\n");
        let mut f = follower(tmp.path(), Start::Beginning);
        f.poll().unwrap();
        f.set_pinned(true);
        write_new(&tmp.path().join("01.log"), "b\n");
        append(&a, "more\n");
        assert_eq!(f.poll().unwrap(), vec![Event::Data(b"more\n".to_vec())]);
        assert_eq!(f.entries().len(), 2);
    }

    #[test]
    fn open_switches_by_hand() {
        let tmp = tempfile::tempdir().unwrap();
        let a = tmp.path().join("00.log");
        write_new(&a, "x\ny\n");
        write_new(&tmp.path().join("01.log"), "b\n");
        let mut f = follower(tmp.path(), Start::Beginning);
        f.poll().unwrap();
        f.open(&a, Start::LastLines(1)).unwrap();
        let ev = f.poll().unwrap();
        assert_eq!(ev[0], Event::Switched(a));
        assert_eq!(data(&ev), "y\n");
    }

    #[test]
    fn filters_by_glob_skips_hidden_and_recurses_on_request() {
        let tmp = tempfile::tempdir().unwrap();
        write_new(&tmp.path().join("a.log"), "log\n");
        write_new(&tmp.path().join(".hidden.log"), "hidden\n");
        write_new(&tmp.path().join("b.txt"), "txt\n");
        let mut f = Follower::new(
            Options {
                dir: tmp.path().to_path_buf(),
                recursive: false,
                glob: Some("*.log".into()),
            },
            Start::Beginning,
        );
        assert_eq!(data(&f.poll().unwrap()), "log\n");

        let sub = tmp.path().join("M2b");
        fs::create_dir(&sub).unwrap();
        write_new(&sub.join("00.log"), "nested\n");
        assert!(f.poll().unwrap().is_empty());

        let mut r = Follower::new(
            Options {
                dir: tmp.path().to_path_buf(),
                recursive: true,
                glob: Some("*.log".into()),
            },
            Start::Beginning,
        );
        assert_eq!(data(&r.poll().unwrap()), "nested\n");
    }

    #[test]
    fn wildcard_matching() {
        assert!(wildcard(b"*.log", b"04-task.log"));
        assert!(!wildcard(b"*.log", b"04-task.log.swp"));
        assert!(wildcard(b"??-*", b"04-task.log"));
        assert!(!wildcard(b"??-*", b"4-task.log"));
        assert!(wildcard(b"*", b""));
        assert!(wildcard(b"a*b*c", b"aXbYbZc"));
        assert!(!wildcard(b"a*b*c", b"aXbYbZ"));
    }

    #[test]
    fn tail_offset_counts_lines_from_the_end() {
        let tmp = tempfile::tempdir().unwrap();
        let p = tmp.path().join("f");
        let check = |text: &str, n: usize, want: &str| {
            fs::write(&p, text).unwrap();
            let mut file = File::open(&p).unwrap();
            let off = tail_offset(&mut file, text.len() as u64, n).unwrap();
            assert_eq!(&text[off as usize..], want, "text {text:?} n {n}");
        };
        check("a\nb\nc\n", 2, "b\nc\n");
        check("a\nb\nc", 2, "b\nc");
        check("a\nb\nc\n", 10, "a\nb\nc\n");
        check("a\nb\nc\n", 0, "");
        check("", 3, "");
        check("\n\n", 1, "\n");
    }
}