mecha10-cli 0.8.0

//! Relay `mecha10-node-runner`'s health-check tracing lines into the dev-mode TUI.
//!
//! [`HealthTailer`] remembers a byte offset per node's log file and relays newly appended
//! health/restart lines into [`crate::ui::LogBuffer`] on each [`HealthTailer::poll`].

use crate::paths;
use crate::ui::LogBuffer;
use mecha10_cli_core::utils::strip_node_scope;
use std::collections::HashMap;
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use std::path::Path;
use std::time::{Duration, Instant};

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum HealthEvent {
    Transition { node: String, running: bool },
    RestartAttempt { node: String, attempt: u32 },
    WillRestart { node: String, detail: String },
}

impl HealthEvent {
    /// The node name this event is about, exactly as it appeared in the log line (node-runner
    /// always logs its own short, scope-stripped name — see `spawn_node_runner`'s `--project`
    /// doc comment on why the CLI passes it that way).
    fn node(&self) -> &str {
        match self {
            HealthEvent::Transition { node, .. }
            | HealthEvent::RestartAttempt { node, .. }
            | HealthEvent::WillRestart { node, .. } => node,
        }
    }

    /// Render this event as a [`LogBuffer`] line. Tagged `HEALTH:` so `DevModeTui`'s log sidebar
    /// (`ui/dev_tui.rs::draw_logs_sidebar`) can style it distinctly from lifecycle `MODE:`/spawn
    /// `STARTED`/`STOPPED` lines already pushed there.
    fn format(&self) -> String {
        match self {
            HealthEvent::Transition { node, running: true } => format!("🩺 HEALTH: {} is Running", node),
            HealthEvent::Transition { node, running: false } => format!("🩺 HEALTH: {} is Stopped", node),
            HealthEvent::RestartAttempt { node, attempt } => {
                format!("🔄 HEALTH: {} restart attempt {}", node, attempt)
            }
            HealthEvent::WillRestart { node, detail } => format!("🔄 HEALTH: {} will restart in {}", node, detail),
        }
    }
}

/// Find `marker` in `line`, then split the text after it into (name-up-to-next-quote, rest).
/// Shared by the `"Node '<name>' ..."` and `"... for '<name>'"` shapes below.
fn extract_quoted_after<'a>(line: &'a str, marker: &str) -> Option<(&'a str, &'a str)> {
    let start = line.find(marker)? + marker.len();
    let rest = &line[start..];
    let end = rest.find('\'')?;
    Some((&rest[..end], &rest[end + 1..]))
}

/// Parse a `"Node '{node}' health: Running"` / `"...: Stopped"` line — see
/// `supervisor.rs::monitor_with_signals_unix`'s health-check branch.
fn parse_transition(line: &str) -> Option<HealthEvent> {
    let (node, after) = extract_quoted_after(line, "Node '")?;
    if after.starts_with(" health: Running") {
        Some(HealthEvent::Transition {
            node: node.to_string(),
            running: true,
        })
    } else if after.starts_with(" health: Stopped") {
        Some(HealthEvent::Transition {
            node: node.to_string(),
            running: false,
        })
    } else {
        None
    }
}

/// Parse a `"Restart attempt {attempt} for '{node}'"` line — see `supervisor.rs::supervise`'s
/// restart-attempt log.
fn parse_restart_attempt(line: &str) -> Option<HealthEvent> {
    let marker = "Restart attempt ";
    let start = line.find(marker)? + marker.len();
    let rest = &line[start..];
    let sep = rest.find(" for '")?;
    let attempt: u32 = rest[..sep].trim().parse().ok()?;
    let (node, _) = extract_quoted_after(&rest[sep..], "for '")?;
    Some(HealthEvent::RestartAttempt {
        node: node.to_string(),
        attempt,
    })
}

/// Parse a `"Node '{node}' will restart in {detail}"` line — see `supervisor.rs::supervise`'s
/// backoff log, logged just before `sleep(backoff_duration)`.
fn parse_will_restart(line: &str) -> Option<HealthEvent> {
    let (node, after) = extract_quoted_after(line, "Node '")?;
    let detail = after.strip_prefix(" will restart in ")?;
    Some(HealthEvent::WillRestart {
        node: node.to_string(),
        detail: detail.trim().to_string(),
    })
}

/// Parse a single tracing log line for any of the node-runner health/restart events this module
/// relays.
pub(crate) fn parse_health_line(line: &str) -> Option<HealthEvent> {
    parse_transition(line)
        .or_else(|| parse_restart_attempt(line))
        .or_else(|| parse_will_restart(line))
}

pub struct HealthTailer {
    offsets: HashMap<String, u64>,
    min_interval: Duration,
    last_poll: Option<Instant>,
}

impl Default for HealthTailer {
    fn default() -> Self {
        Self::new(Duration::from_millis(500))
    }
}

impl HealthTailer {
    mecha10_core::new_with_defaults!(
        /// Create a tailer that does real file I/O at most once per `min_interval`.
        pub fn new(min_interval: Duration) -> Self {
            offsets: HashMap::new(),
            last_poll: None
        }
    );

    /// Check every node in `node_identifiers` for newly appended health/restart lines and relay
    /// them into `log_buffer`.
    pub fn poll(&mut self, node_identifiers: &[String], log_buffer: &LogBuffer) {
        self.poll_in_dir(node_identifiers, Path::new(paths::project::LOGS_DIR), log_buffer);
    }

    /// Throttle-checked poll against an explicit `logs_dir` - split out from `poll` so tests can
    /// exercise the throttling behavior against a temp directory instead of the process CWD's
    /// real `logs/` folder.
    fn poll_in_dir(&mut self, node_identifiers: &[String], logs_dir: &Path, log_buffer: &LogBuffer) {
        if let Some(last) = self.last_poll {
            if last.elapsed() < self.min_interval {
                return;
            }
        }
        self.last_poll = Some(Instant::now());

        for node in node_identifiers {
            self.poll_node(node, logs_dir, log_buffer);
        }
    }

    /// Read and process whatever's new in one node's log file. Split out from `poll` with an
    /// explicit `logs_dir` so it's unit-testable against a temp directory instead of the process
    /// CWD's real `logs/` folder.
    fn poll_node(&mut self, node_identifier: &str, logs_dir: &Path, log_buffer: &LogBuffer) {
        let short_name = strip_node_scope(node_identifier);
        let log_path = logs_dir.join(format!("{}.log", short_name));

        let Ok(mut file) = File::open(&log_path) else {
            return;
        };
        let Ok(metadata) = file.metadata() else { return };
        let len = metadata.len();

        let offset = self.offsets.entry(node_identifier.to_string()).or_insert(0);
        if len < *offset {
            *offset = 0;
        }
        if len == *offset {
            return;
        }

        if file.seek(SeekFrom::Start(*offset)).is_err() {
            return;
        }
        let mut buf = Vec::new();
        if file.read_to_end(&mut buf).is_err() {
            return;
        }
        let text = String::from_utf8_lossy(&buf);

        let Some(last_newline) = text.rfind('\n') else {
            return;
        };
        *offset += (last_newline + 1) as u64;

        for line in text[..last_newline].lines() {
            if let Some(event) = parse_health_line(line) {
                if event.node() == short_name {
                    log_buffer.push(event.format());
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;

    #[test]
    fn parses_running_transition() {
        let line = "2024-01-01T00:00:00Z  INFO mecha10_node_runner::supervisor: Node 'camera' health: Running";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::Transition {
                node: "camera".to_string(),
                running: true
            })
        );
    }

    #[test]
    fn parses_stopped_transition() {
        let line = "2024-01-01T00:00:00Z  WARN mecha10_node_runner::supervisor: Node 'lidar' health: Stopped";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::Transition {
                node: "lidar".to_string(),
                running: false
            })
        );
    }

    #[test]
    fn parses_bare_message_without_tracing_prefix() {
        assert_eq!(
            parse_health_line("Node 'camera' health: Running"),
            Some(HealthEvent::Transition {
                node: "camera".to_string(),
                running: true
            })
        );
    }

    #[test]
    fn parses_restart_attempt() {
        let line = "INFO mecha10_node_runner::supervisor: Restart attempt 2 for 'camera'";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::RestartAttempt {
                node: "camera".to_string(),
                attempt: 2
            })
        );
    }

    #[test]
    fn parses_will_restart_with_debug_formatted_duration() {
        let line = "WARN mecha10_node_runner::supervisor: Node 'camera' will restart in 2s";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::WillRestart {
                node: "camera".to_string(),
                detail: "2s".to_string()
            })
        );
    }

    #[test]
    fn ignores_unrelated_log_lines() {
        assert_eq!(parse_health_line("Node 'camera' started (PID: Some(1234))"), None);
        assert_eq!(parse_health_line("Starting supervision for 'camera'"), None);
        assert_eq!(parse_health_line(""), None);
    }

    #[test]
    fn ignores_health_line_with_unknown_status_word() {
        assert_eq!(parse_health_line("Node 'camera' health: Unknown"), None);
    }

    #[test]
    fn formats_running_transition_with_health_tag() {
        let event = HealthEvent::Transition {
            node: "camera".to_string(),
            running: true,
        };
        assert_eq!(event.format(), "🩺 HEALTH: camera is Running");
    }

    #[test]
    fn formats_stopped_transition_with_health_tag() {
        let event = HealthEvent::Transition {
            node: "camera".to_string(),
            running: false,
        };
        assert_eq!(event.format(), "🩺 HEALTH: camera is Stopped");
    }

    fn write_log(dir: &Path, node: &str, content: &str) {
        let path = dir.join(format!("{}.log", node));
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)
            .unwrap();
        file.write_all(content.as_bytes()).unwrap();
    }

    #[test]
    fn poll_node_relays_new_health_transition_into_log_buffer() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'camera' health: Running\n");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);

        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);

        assert_eq!(log_buffer.get_all(), vec!["🩺 HEALTH: camera is Running".to_string()]);
    }

    #[test]
    fn poll_node_only_relays_lines_appended_since_last_poll() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'camera' health: Running\n");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_all().len(), 1);

        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_all().len(), 1);

        write_log(dir.path(), "camera", "Node 'camera' health: Stopped\n");
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(
            log_buffer.get_all(),
            vec![
                "🩺 HEALTH: camera is Running".to_string(),
                "🩺 HEALTH: camera is Stopped".to_string(),
            ]
        );
    }

    #[test]
    fn poll_node_leaves_incomplete_trailing_line_for_next_poll() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'camera' health: Run");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert!(log_buffer.get_all().is_empty(), "partial line should not be parsed yet");

        write_log(dir.path(), "camera", "ning\n");
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_all(), vec!["🩺 HEALTH: camera is Running".to_string()]);
    }

    #[test]
    fn poll_node_is_a_noop_when_log_file_does_not_exist() {
        let dir = tempfile::tempdir().unwrap();
        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);

        tailer.poll_node("@mecha10/nonexistent", dir.path(), &log_buffer);

        assert!(log_buffer.get_all().is_empty());
    }

    #[test]
    fn poll_node_ignores_events_for_a_different_node_in_the_same_file() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'lidar' health: Running\n");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);

        assert!(log_buffer.get_all().is_empty());
    }

    #[test]
    fn poll_throttles_real_polls_to_min_interval() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'camera' health: Running\n");

        let mut tailer = HealthTailer::new(Duration::from_secs(3600));
        let log_buffer = LogBuffer::new(100);
        let nodes = vec!["@mecha10/camera".to_string()];

        tailer.poll_in_dir(&nodes, dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_all().len(), 1);

        write_log(dir.path(), "camera", "Node 'camera' health: Stopped\n");
        tailer.poll_in_dir(&nodes, dir.path(), &log_buffer);
        assert_eq!(
            log_buffer.get_all().len(),
            1,
            "throttled poll should not have read the new line"
        );
    }
}