fsmon 0.5.1

use libc;
use std::collections::{HashMap, HashSet, VecDeque};
use std::fs::{self, File, OpenOptions};
use std::io::{BufWriter, Write};
use std::path::PathBuf;
use std::time::Duration;

use crate::common::FileEvent;
use crate::common::metrics::MetricsRegistry;
use crate::{debug_log, warning_log};

/// Maximum number of open file handles to keep cached.
const MAX_OPEN_HANDLES: usize = 64;
/// Flush + sync interval: flush BufWriter to OS buffer and fdatasync to disk every second.
/// Ensures data is visible and persisted during runtime, not just on shutdown.
const FLUSH_SYNC_INTERVAL: Duration = Duration::from_secs(1);

// ---- FileLogWriter: unified event stream consumer for disk persistence ----

/// Async file writer consuming events from the broadcast stream and writing to JSONL files.
/// Runs as a tokio task. Handles disk-full buffering, fdatasync, and ENOENT retry.
pub(crate) struct FileLogWriter {
    log_dir: PathBuf,
    disk_buf: VecDeque<(FileEvent, String)>,
    disk_healthy: bool,
    last_disk_check: std::time::Instant,
    dirty_logs: HashSet<PathBuf>,
    debug: bool,
    local_time: bool,
    metrics: MetricsRegistry,
    /// Cached open file handles, keyed by log path.
    /// Avoids open+close per event for high-frequency writes.
    handles: HashMap<PathBuf, BufWriter<File>>,
}

impl std::fmt::Debug for FileLogWriter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FileLogWriter")
            .field("log_dir", &self.log_dir)
            .field("disk_buf_len", &self.disk_buf.len())
            .field("disk_healthy", &self.disk_healthy)
            .field("debug", &self.debug)
            .field("local_time", &self.local_time)
            .finish()
    }
}

impl FileLogWriter {
    pub(crate) fn new(
        log_dir: PathBuf,
        debug: bool,
        local_time: bool,
        metrics: MetricsRegistry,
    ) -> Self {
        Self {
            log_dir,
            disk_buf: VecDeque::with_capacity(10_000),
            disk_healthy: true,
            last_disk_check: std::time::Instant::now(),
            dirty_logs: HashSet::new(),
            metrics,
            debug,
            local_time,
            handles: HashMap::new(),
        }
    }

    /// Select UTC or local time serialization based on config.
    fn jsonl_string(&self, event: &FileEvent) -> String {
        if self.local_time {
            event.to_jsonl_string_local()
        } else {
            event.to_jsonl_string()
        }
    }

    /// Run the file writer event loop.
    pub(crate) async fn run(
        mut self,
        mut rx: tokio::sync::broadcast::Receiver<(FileEvent, String)>,
    ) {
        use tokio::time::interval;

        // Single timer: flush BufWriter → OS buffer + fdatasync → disk every second.
        let mut flush_sync_timer = interval(FLUSH_SYNC_INTERVAL);

        loop {
            tokio::select! {
                result = rx.recv() => {
                    match result {
                        Ok((event, cmd_name)) => {
                            if let Err(e) = self.write_event(&event, &cmd_name)
                                && self.debug {
                                    debug_log!(self.debug, "FileLogWriter write error: {}", e);
                                }
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                            warning_log!("FileLogWriter dropped {} events (disk too slow)", n);
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                            break;
                        }
                    }
                }
                // Flush + sync: BufWriter → OS buffer → disk
                _ = flush_sync_timer.tick() => {
                    self.flush_and_sync_dirty_logs();
                }
            }
        }

        self.flush_and_sync_dirty_logs();
        self.flush_all();
    }

    /// Get or open a BufWriter for the given log path.
    /// Returns None if the open fails (caller should buffer the event).
    fn get_or_open(&mut self, log_path: &PathBuf) -> std::io::Result<&mut BufWriter<File>> {
        if !self.handles.contains_key(log_path) {
            // Evict oldest handle if at capacity
            if self.handles.len() >= MAX_OPEN_HANDLES
                && let Some(evict_path) = self.handles.keys().next().cloned()
            {
                self.handles.remove(&evict_path);
            }
            let file = open_log_file(log_path, &self.log_dir)?;
            self.handles.insert(log_path.clone(), BufWriter::new(file));
        }
        Ok(self
            .handles
            .get_mut(log_path)
            .expect("handle should exist after insert or contains_key check"))
    }

    /// Write an event to the appropriate JSONL log file.
    /// Returns Ok(()) even if disk is full (event is buffered for retry).
    fn write_event(&mut self, event: &FileEvent, cmd_name: &str) -> std::io::Result<()> {
        let log_path = self
            .log_dir
            .join(crate::common::utils::cmd_to_log_name(cmd_name));

        // Try to flush buffer if disk was previously unhealthy
        if !self.disk_healthy
            && self.last_disk_check.elapsed() >= std::time::Duration::from_secs(10)
        {
            self.flush_disk_buf();
        }

        // Serialize before borrowing self.handles
        let line = self.jsonl_string(event);

        // If the file was deleted externally (e.g. by user or log rotation),
        // the cached BufWriter holds a stale fd pointing to a deleted inode.
        // Instead of path.exists() (TOCTOU), we catch ENOENT on write and
        // retry with a fresh handle.
        // Note: stale handle detection is done via write error handling below.

        match self.get_or_open(&log_path) {
            Ok(writer) => {
                writeln!(writer, "{}", line)?;
                self.dirty_logs.insert(log_path);
                self.disk_healthy = true;
                Ok(())
            }
            Err(e) => {
                // Disk might be full — buffer the event
                self.handles.remove(&log_path);
                self.disk_healthy = false;
                self.last_disk_check = std::time::Instant::now();
                if self.disk_buf.len() < 10_000 {
                    self.disk_buf
                        .push_back((event.clone(), cmd_name.to_string()));
                }
                self.metrics
                    .set_disk_buffer_events(self.disk_buf.len() as i64);
                Err(e)
            }
        }
    }

    /// Try to flush buffered events to disk.
    fn flush_disk_buf(&mut self) {
        if self.disk_buf.is_empty() {
            self.disk_healthy = true;
            return;
        }

        let mut remaining = VecDeque::new();
        while let Some((event, cmd_name)) = self.disk_buf.pop_front() {
            let log_path = self
                .log_dir
                .join(crate::common::utils::cmd_to_log_name(&cmd_name));
            let line = self.jsonl_string(&event);
            match self.get_or_open(&log_path) {
                Ok(writer) => {
                    if writeln!(writer, "{}", line).is_err() {
                        self.handles.remove(&log_path);
                        remaining.push_back((event, cmd_name));
                    }
                }
                Err(_) => {
                    remaining.push_back((event, cmd_name));
                }
            }
        }
        self.disk_buf = remaining;
        self.disk_healthy = self.disk_buf.is_empty();
        self.last_disk_check = std::time::Instant::now();
        self.metrics
            .set_disk_buffer_events(self.disk_buf.len() as i64);
    }

    /// Flush all open BufWriters to OS buffers.
    /// Called on shutdown to ensure no data is lost in user-space buffers.
    fn flush_all(&mut self) {
        for (path, writer) in &mut self.handles {
            if let Err(e) = writer.flush() {
                warning_log!("flush failed for '{}': {}", path.display(), e);
            }
        }
    }

    /// Flush + sync all dirty log files.
    /// Flush moves data from BufWriter (user-space) to OS buffer,
    /// then fdatasync persists OS buffer to disk.
    ///
    /// # Safety
    ///
    /// This function calls `libc::fstat` to check if a log file has been
    /// externally deleted (nlink == 0). The `fstat` call is safe because:
    /// - The fd is a valid, open file descriptor from `BufWriter<File>`
    /// - `std::mem::zeroed()` produces a valid `libc::stat` zero-initialized struct
    /// - The fd is not used concurrently (single-threaded access via `&mut self`)
    fn flush_and_sync_dirty_logs(&mut self) {
        if self.dirty_logs.is_empty() {
            return;
        }
        let paths: Vec<PathBuf> = self.dirty_logs.drain().collect();
        for path in &paths {
            // Try cached handle first
            if let Some(writer) = self.handles.get_mut(path) {
                // Check if file was externally deleted (nlink == 0).
                // On Unix, writing to an unlinked fd succeeds but data is lost
                // when the fd is closed. Detect this early and drop the stale handle.
                // fstat is ~100ns, negligible vs fdatasync's ms-level I/O.
                use std::os::fd::AsRawFd;
                let nlink = unsafe {
                    let mut stat: libc::stat = std::mem::zeroed();
                    libc::fstat(writer.get_ref().as_raw_fd(), &mut stat);
                    stat.st_nlink
                };
                if nlink == 0 {
                    if self.debug {
                        debug_log!(
                            self.debug,
                            "Log file '{}' was deleted externally, dropping stale handle (next write will recreate)",
                            path.display()
                        );
                    }
                    self.handles.remove(path);
                    continue;
                }

                // Flush BufWriter's user-space buffer to OS
                if let Err(e) = writer.flush() {
                    warning_log!("flush failed for '{}': {}", path.display(), e);
                    continue;
                }
                // Then sync OS buffer to disk
                if let Err(e) = writer.get_ref().sync_data() {
                    warning_log!("fdatasync failed for '{}': {}", path.display(), e);
                }
            } else {
                // Fallback: open for sync (handle was evicted)
                match File::options().write(true).open(path) {
                    Ok(file) => {
                        if let Err(e) = file.sync_data() {
                            warning_log!("fdatasync failed for '{}': {}", path.display(), e);
                        }
                    }
                    Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                    Err(e) => {
                        warning_log!("Could not open '{}' for sync: {}", path.display(), e);
                    }
                }
            }
        }
    }
}

/// Open a log file with create+append, retrying once on ENOENT.
/// Chowns the file to the original user immediately after creation using fchown
/// (fd-based) to avoid TOCTOU race: the fd is always valid even if the path is
/// deleted between open() and chown().
fn open_log_file(log_path: &PathBuf, log_dir: &PathBuf) -> std::io::Result<File> {
    use std::os::unix::fs::OpenOptionsExt;
    // O_NOFOLLOW prevents symlink attacks (F-018)
    // O_CLOEXEC prevents fd leaks to child processes
    let mut opts = OpenOptions::new();
    opts.create(true)
        .append(true)
        .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);
    let file = opts.open(log_path).or_else(|e| {
        if e.kind() == std::io::ErrorKind::NotFound {
            let _ = fs::create_dir_all(log_dir);
            let _ = crate::common::fid_parser::chown_to_user(log_dir);
            let mut opts2 = OpenOptions::new();
            opts2
                .create(true)
                .append(true)
                .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);
            opts2.open(log_path)
        } else {
            Err(e)
        }
    })?;
    // Chown via fchown (fd-based) — no TOCTOU race.
    // The path-based chown_to_user would fail with ENOENT if the file is
    // deleted between open() and chown(); fchown operates on the fd directly
    // and always succeeds as long as the fd is open.
    fchown_to_user(&file);
    Ok(file)
}

/// fchown a file descriptor to the original user (SUDO_UID/SUDO_GID).
/// Best-effort: errors are logged but not propagated — the file is still usable.
///
/// # Safety
///
/// This function calls `libc::fchown` to change file ownership. The call is safe
/// because:
/// - The fd is a valid, open file descriptor from `File::open`
/// - `uid` and `gid` are valid u32 values from `resolve_uid_gid()`
/// - The return value is checked; errors are logged but not propagated
/// - Expected errors (EPERM, EOPNOTSUPP, ENOSYS) are silently ignored for
///   compatibility with vfat/exfat/NFS filesystems
fn fchown_to_user(file: &File) {
    let (uid, gid) = crate::common::config::resolve_uid_gid();
    use std::os::fd::AsRawFd;
    let ret = unsafe { libc::fchown(file.as_raw_fd(), uid, gid) };
    if ret != 0 {
        let err = std::io::Error::last_os_error();
        // EPERM / EOPNOTSUPP / ENOSYS are expected on vfat/exfat/NFS — skip warning.
        let errno = err.raw_os_error().unwrap_or(0);
        if errno != libc::EPERM && errno != libc::EOPNOTSUPP && errno != libc::ENOSYS {
            warning_log!(
                "fchown failed for log file (fd {}): {}",
                file.as_raw_fd(),
                err
            );
        }
    }
}