mcp/src/logs/follower.rs

use crate::{
    auth::AuthorizationService,
    logs::{index::HistoryStore, stable_line_id, timestamp_from_line, LogLine},
};
use notify::{Event, RecommendedWatcher, RecursiveMode, Watcher};
use serde::Serialize;
use std::{
    fs::File,
    io::{Read, Seek, SeekFrom},
    path::PathBuf,
};
use tokio::sync::broadcast;

#[derive(Debug, Clone, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum LiveMessage {
    Hello,
    LogLines { lines: Vec<LogLine> },
    Rotation,
    HistoryChanged,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct FileIdentity {
    device: u64,
    inode: u64,
}
#[cfg(unix)]
fn identity(meta: &std::fs::Metadata) -> FileIdentity {
    use std::os::unix::fs::MetadataExt;
    FileIdentity {
        device: meta.dev(),
        inode: meta.ino(),
    }
}

pub struct FileFollower {
    path: PathBuf,
    file: Option<File>,
    identity: Option<FileIdentity>,
    offset: u64,
    partial: Vec<u8>,
    next_line: usize,
    redact: bool,
}
#[derive(Debug, Default)]
pub struct FollowResult {
    pub lines: Vec<LogLine>,
    pub rotated: bool,
    pub truncated: bool,
}

impl FileFollower {
    pub fn at_eof(path: PathBuf, redact: bool) -> std::io::Result<Self> {
        let mut s = Self {
            path,
            file: None,
            identity: None,
            offset: 0,
            partial: Vec::new(),
            next_line: 0,
            redact,
        };
        s.open(true)?;
        Ok(s)
    }
    pub fn from_start(path: PathBuf, redact: bool) -> std::io::Result<Self> {
        let mut s = Self {
            path,
            file: None,
            identity: None,
            offset: 0,
            partial: Vec::new(),
            next_line: 0,
            redact,
        };
        s.open(false)?;
        Ok(s)
    }
    fn open(&mut self, eof: bool) -> std::io::Result<()> {
        let mut file = File::open(&self.path)?;
        let meta = file.metadata()?;
        self.identity = Some(identity(&meta));
        self.partial.clear();
        self.next_line = 0;
        if eof {
            let mut chunk = [0_u8; 8192];
            loop {
                let read = file.read(&mut chunk)?;
                if read == 0 {
                    break;
                }
                for byte in &chunk[..read] {
                    if *byte == b'\n' {
                        self.next_line += 1;
                        self.partial.clear();
                    } else {
                        self.partial.push(*byte);
                    }
                }
            }
            self.offset = meta.len();
        } else {
            self.offset = 0;
        }
        self.file = Some(file);
        Ok(())
    }
    pub fn read_new(&mut self) -> std::io::Result<FollowResult> {
        let mut result = FollowResult::default();
        let current = match std::fs::metadata(&self.path) {
            Ok(m) => m,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(result),
            Err(e) => return Err(e),
        };
        let current_id = identity(&current);
        if self.identity != Some(current_id) {
            if let Some(old) = self.file.as_mut() {
                let mut tail = Vec::new();
                old.seek(SeekFrom::Start(self.offset))?;
                old.read_to_end(&mut tail)?;
                result.lines.extend(decode_lines(
                    &mut self.partial,
                    &tail,
                    self.identity.expect("open file identity"),
                    &mut self.next_line,
                    self.redact,
                ));
            }
            self.open(false)?;
            result.rotated = true;
        } else if current.len() < self.offset {
            self.offset = 0;
            self.partial.clear();
            self.next_line = 0;
            result.truncated = true;
        }
        let file = self.file.as_mut().expect("opened follower");
        file.seek(SeekFrom::Start(self.offset))?;
        let mut bytes = Vec::new();
        file.read_to_end(&mut bytes)?;
        self.offset += bytes.len() as u64;
        result.lines.extend(decode_lines(
            &mut self.partial,
            &bytes,
            self.identity.expect("open file identity"),
            &mut self.next_line,
            self.redact,
        ));
        Ok(result)
    }
}
fn decode_lines(
    partial: &mut Vec<u8>,
    bytes: &[u8],
    identity: FileIdentity,
    next_line: &mut usize,
    redact: bool,
) -> Vec<LogLine> {
    partial.extend_from_slice(bytes);
    let mut completed = Vec::new();
    let mut consumed = 0;
    for (i, b) in partial.iter().enumerate() {
        if *b == b'\n' {
            let raw = String::from_utf8_lossy(&partial[consumed..i])
                .trim_end_matches('\r')
                .to_owned();
            let id = stable_line_id(identity.device, identity.inode, *next_line, &raw);
            let text = if redact {
                crate::logs::redact::redact_player_ips(&raw)
            } else {
                raw
            };
            completed.push(LogLine {
                id,
                timestamp: timestamp_from_line(&text),
                text,
                source: "latest.log".into(),
            });
            *next_line += 1;
            consumed = i + 1;
        }
    }
    partial.drain(..consumed);
    completed
}

pub fn spawn_watcher(
    log_dir: PathBuf,
    latest: PathBuf,
    whitelist_path: PathBuf,
    auth: AuthorizationService,
    history: HistoryStore,
    tx: broadcast::Sender<LiveMessage>,
    redact: bool,
) -> notify::Result<RecommendedWatcher> {
    // Open first, then register the directory watch, then recover bytes written
    // between those two operations. Events queued after registration are harmless:
    // the follower's byte offset makes the later read a no-op.
    let mut follower = FileFollower::at_eof(latest.clone(), false).map_err(notify::Error::io)?;
    let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel::<notify::Result<Event>>();
    let mut watcher = notify::recommended_watcher(move |event| {
        let _ = event_tx.send(event);
    })?;
    watcher.watch(&log_dir, RecursiveMode::NonRecursive)?;
    if whitelist_path.parent() != Some(log_dir.as_path()) {
        if let Some(parent) = whitelist_path.parent() {
            watcher.watch(parent, RecursiveMode::NonRecursive)?;
        }
    }
    let startup_delta = follower.read_new();
    tokio::spawn(async move {
        if let Ok(result) = startup_delta {
            if !result.lines.is_empty() {
                let mut lines = result.lines;
                for line in &lines {
                    auth.observe_line(&line.text).await;
                }
                if redact {
                    for line in &mut lines {
                        line.text = crate::logs::redact::redact_player_ips(&line.text);
                    }
                }
                let _ = tx.send(LiveMessage::LogLines { lines });
            }
        }
        while let Some(event) = event_rx.recv().await {
            match event {
                Ok(event) => {
                    let touches_whitelist = event.paths.iter().any(|p| p == &whitelist_path);
                    let touches_logs = event
                        .paths
                        .iter()
                        .any(|p| p == &latest || p.parent() == Some(log_dir.as_path()));
                    if touches_whitelist {
                        match crate::auth::whitelist::Whitelist::load(&whitelist_path) {
                            Ok(w) => {
                                auth.replace_whitelist(w).await;
                                tracing::info!(event = "whitelist_changed")
                            }
                            Err(e) => tracing::error!(event="whitelist_reload_failed",error=%e),
                        }
                    }
                    if touches_logs {
                        match follower.read_new() {
                            Ok(result) => {
                                if result.rotated || result.truncated {
                                    auth.clear_for_rotation().await;
                                    let _ = auth.rebuild().await;
                                    let _ = history.refresh(&log_dir).await;
                                    let _ = tx.send(LiveMessage::Rotation);
                                }
                                if !result.lines.is_empty() {
                                    let mut lines = result.lines;
                                    for line in &lines {
                                        auth.observe_line(&line.text).await;
                                    }
                                    if redact {
                                        for line in &mut lines {
                                            line.text =
                                                crate::logs::redact::redact_player_ips(&line.text);
                                        }
                                    }
                                    let _ = tx.send(LiveMessage::LogLines { lines });
                                } else {
                                    let _ = history.refresh(&log_dir).await;
                                    let _ = tx.send(LiveMessage::HistoryChanged);
                                }
                            }
                            Err(e) => tracing::warn!(event="watcher_recovery",error=%e),
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!(event="watcher_overflow_recovery",error=%e);
                    let _ = auth.rebuild().await;
                    let _ = history.refresh(&log_dir).await;
                    let _ = follower.read_new();
                }
            }
        }
    });
    Ok(watcher)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    #[test]
    fn buffers_partial_and_handles_multiple_lines() {
        let d = tempfile::tempdir().unwrap();
        let p = d.path().join("latest.log");
        std::fs::write(&p, b"").unwrap();
        let mut f = FileFollower::from_start(p.clone(), false).unwrap();
        std::fs::write(&p, b"one").unwrap();
        assert!(f.read_new().unwrap().lines.is_empty());
        std::fs::OpenOptions::new()
            .append(true)
            .open(&p)
            .unwrap()
            .write_all(b"\ntwo\n")
            .unwrap();
        let r = f.read_new().unwrap();
        assert_eq!(
            r.lines.iter().map(|l| l.text.as_str()).collect::<Vec<_>>(),
            vec!["one", "two"]
        );
    }
    #[test]
    fn detects_truncation() {
        let d = tempfile::tempdir().unwrap();
        let p = d.path().join("latest.log");
        std::fs::write(&p, b"old\n").unwrap();
        let mut f = FileFollower::from_start(p.clone(), false).unwrap();
        assert_eq!(f.read_new().unwrap().lines.len(), 1);
        std::fs::write(&p, b"n\n").unwrap();
        let r = f.read_new().unwrap();
        assert!(r.truncated);
        assert_eq!(r.lines[0].text, "n");
    }
    #[test]
    fn detects_file_replacement_and_reads_the_new_generation() {
        let d = tempfile::tempdir().unwrap();
        let p = d.path().join("latest.log");
        std::fs::write(&p, b"old\n").unwrap();
        let mut follower = FileFollower::at_eof(p.clone(), false).unwrap();
        std::fs::rename(&p, d.path().join("previous.log")).unwrap();
        std::fs::write(&p, b"new\n").unwrap();
        let result = follower.read_new().unwrap();
        assert!(result.rotated);
        assert_eq!(result.lines.last().unwrap().text, "new");
    }
    #[test]
    fn eof_startup_retains_an_existing_partial_line() {
        let d = tempfile::tempdir().unwrap();
        let p = d.path().join("latest.log");
        std::fs::write(&p, b"complete\npartial").unwrap();
        let mut follower = FileFollower::at_eof(p.clone(), false).unwrap();
        std::fs::OpenOptions::new()
            .append(true)
            .open(&p)
            .unwrap()
            .write_all(b" rest\n")
            .unwrap();
        let result = follower.read_new().unwrap();
        assert_eq!(result.lines[0].text, "partial rest");
    }
    #[tokio::test]
    async fn live_and_recent_snapshots_share_line_ids() {
        let d = tempfile::tempdir().unwrap();
        let p = d.path().join("latest.log");
        std::fs::write(&p, b"").unwrap();
        let mut follower = FileFollower::from_start(p.clone(), false).unwrap();
        std::fs::write(&p, b"same line\n").unwrap();
        let live = follower.read_new().unwrap().lines.remove(0);
        let index = crate::logs::index::HistoryIndex::scan(d.path()).unwrap();
        let history = HistoryStore::new(index, 0, 0, 1024, false)
            .page(None, 1)
            .await
            .unwrap()
            .lines
            .remove(0);
        assert_eq!(live.id, history.id);
    }
}