openlatch-client 0.1.18

/// JSONL event audit logger with daily rotation and retention cleanup.
///
/// This is a leaf module — it has zero imports from other modules in this crate.
/// The writer task runs in its own tokio task and is the ONLY writer for log files,
/// preventing concurrent write corruption (architecture rule: single writer per file).
pub mod daemon_log;
pub mod tamper_log;

use std::io::Write;
use std::path::{Path, PathBuf};

use tokio::sync::mpsc;

// ---------------------------------------------------------------------------
// Log file naming (LOG-01, LOG-02)
// ---------------------------------------------------------------------------

/// Returns the log file name for the given date.
///
/// # Examples
///
/// ```
/// use chrono::NaiveDate;
/// // Returns "events-2026-04-07.jsonl"
/// ```
pub fn log_file_name(date: &chrono::NaiveDate) -> String {
    format!("events-{}.jsonl", date.format("%Y-%m-%d"))
}

/// Returns the full path for today's event log file (UTC date).
pub fn current_log_path(log_dir: &Path) -> PathBuf {
    let today = chrono::Utc::now().date_naive();
    log_dir.join(log_file_name(&today))
}

// ---------------------------------------------------------------------------
// Synchronous append helper (used by writer task only)
// ---------------------------------------------------------------------------

/// Append a single JSON event as one line to the log file.
///
/// Creates the file if it does not exist (append mode).
/// Returns an error if the file cannot be opened or written.
///
/// # Errors
///
/// Returns `std::io::Error` if the file cannot be opened or if serialization fails.
pub fn append_event_sync(path: &Path, event: &serde_json::Value) -> std::io::Result<()> {
    let line = serde_json::to_string(event).map_err(std::io::Error::other)?;
    append_line_sync(path, &line)
}

/// Append a pre-serialized JSON line to the log file.
///
/// Creates the file if it does not exist (append mode).
fn append_line_sync(path: &Path, line: &str) -> std::io::Result<()> {
    let mut file = open_append(path)?;
    writeln!(file, "{line}")?;
    Ok(())
}

/// Open a file in create-if-missing append mode. Shared by the synchronous
/// append helper and the background writer's `BufWriter`.
fn open_append(path: &Path) -> std::io::Result<std::fs::File> {
    std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
}

// ---------------------------------------------------------------------------
// EventLogger with mpsc channel (PERFORMANCE: non-blocking on hot path)
// ---------------------------------------------------------------------------

/// Sends pre-serialized event JSON lines to the background writer task via an mpsc channel.
///
/// `log()` is non-blocking: if the channel is full (capacity 4096), the event
/// is dropped with a warning rather than blocking the verdict path. The
/// capacity holds a full startup inventory walk (~1k config-snapshot events
/// emitted in one sub-100ms burst) so the boot burst is buffered rather than
/// dropped; tokio's mpsc allocates per queued message, so the higher bound
/// costs memory only while events are actually in flight.
///
/// # Architecture
///
/// `EventLogger` is the sender side; `EventLoggerHandle` owns the background task.
/// The caller must drop `EventLogger` before calling `EventLoggerHandle::shutdown()`
/// so the writer task sees the channel close and exits cleanly.
///
/// Events are accepted as pre-serialized JSON strings to avoid double serialization
/// (struct → Value → String). The caller serializes once; the writer appends as-is.
#[derive(Clone)]
pub struct EventLogger {
    tx: mpsc::Sender<String>,
}

/// Owns the background writer task. MUST be held until shutdown.
pub struct EventLoggerHandle {
    join_handle: tokio::task::JoinHandle<()>,
}

impl EventLogger {
    /// Create a new `EventLogger` and its background writer task.
    ///
    /// Returns the logger (sender) and the handle (task owner) as a pair.
    /// The caller must hold the `EventLoggerHandle` until the daemon shuts down.
    pub fn new(log_dir: PathBuf) -> (Self, EventLoggerHandle) {
        let (logger, mut rx) = Self::channel();
        let handle = EventLoggerHandle {
            join_handle: tokio::spawn(async move { run_event_writer(log_dir, &mut rx).await }),
        };
        (logger, handle)
    }

    /// Build the sender + receiver **without** spawning the writer.
    ///
    /// The daemon uses this so it can run [`run_event_writer`] under its
    /// in-process task supervisor (`core::supervision::task`) instead of a bare
    /// detached `tokio::spawn`. Keeping the spawn out of this module is what
    /// lets `logging/` stay a leaf: it exposes the receiver and the writer
    /// routine, and the supervision policy lives with the daemon.
    pub fn channel() -> (Self, mpsc::Receiver<String>) {
        let (tx, rx) = mpsc::channel(4096);
        (Self { tx }, rx)
    }

    /// Send a pre-serialized JSON line to the background writer (non-blocking).
    ///
    /// Uses `try_send` so the hot path never awaits channel capacity.
    /// If the channel is full or closed, the event is dropped with a warning.
    pub fn log(&self, event_json: String) {
        // PERFORMANCE: try_send is synchronous — never blocks the verdict path
        if self.tx.try_send(event_json).is_err() {
            tracing::warn!("event log channel full or closed, event dropped");
        }
    }

    /// Send a pre-serialized JSON line, awaiting channel capacity if needed.
    ///
    /// For BACKGROUND producers only — the configuration monitor's inventory
    /// walk emits its whole result set in one tight loop with no other await
    /// point, so `try_send` there does not degrade gracefully: it overruns the
    /// 1024-slot buffer before the writer task is next scheduled and drops
    /// audit lines that nothing is latency-sensitive about. Applying
    /// backpressure paces the walk to the writer instead.
    ///
    /// NEVER call this from the verdict path — that path has a <5 ms budget and
    /// must keep using `log()`.
    pub async fn log_backpressured(&self, event_json: String) {
        if self.tx.send(event_json).await.is_err() {
            tracing::warn!("event log channel closed, event dropped");
        }
    }
}

impl EventLoggerHandle {
    /// Wrap an already-spawned writer task.
    ///
    /// Used by the daemon, which spawns the writer through its task supervisor
    /// so a panic in the writer restarts it instead of silently ending event
    /// logging for the rest of the process lifetime.
    pub fn from_task(join_handle: tokio::task::JoinHandle<()>) -> Self {
        Self { join_handle }
    }

    /// Wait for the background writer to drain and exit.
    ///
    /// The caller must drop (or let go of) the `EventLogger` sender BEFORE calling
    /// this method, otherwise the writer task will block waiting for more events.
    pub async fn shutdown(self) {
        let _ = self.join_handle.await;
    }
}

/// Background task that drains the mpsc channel and appends pre-serialized JSON lines to disk.
///
/// This is the ONLY writer for event log files — single-writer design prevents
/// file corruption without requiring a mutex.
///
/// The file handle is held open across events inside a `BufWriter`, and each
/// wake-up drains every line currently queued before a single `flush()`.
/// Re-opening the file per event (the original design) could not keep up with a
/// startup inventory burst — the channel overflowed and audit events were
/// dropped (`event log channel full`). The dated path is derived once per drain
/// batch (it only changes at a midnight rollover); a low-rate stream still
/// flushes after every event because the drain loop empties at once.
///
/// Takes the receiver by `&mut` rather than by value so a supervisor can
/// re-invoke it on the **same** channel after a panic — the queued lines an
/// owned receiver would have taken to the grave survive the restart.
pub async fn run_event_writer(log_dir: PathBuf, rx: &mut mpsc::Receiver<String>) {
    // Ensure log directory exists before writing any events.
    if let Err(e) = tokio::fs::create_dir_all(&log_dir).await {
        tracing::error!(error = %e, "failed to create log directory");
        return;
    }

    let mut writer: Option<(PathBuf, std::io::BufWriter<std::fs::File>)> = None;

    while let Some(first) = rx.recv().await {
        // Derive the dated path once per wake-up: it can only change at a
        // midnight rollover, so recomputing it per line would waste a clock
        // read plus two allocations on every event of a boot-inventory burst.
        let path = current_log_path(&log_dir);
        append_log_line(&mut writer, &path, &first);
        while let Ok(line) = rx.try_recv() {
            append_log_line(&mut writer, &path, &line);
        }
        // Flush the whole batch at once; the next wake-up re-derives the path.
        flush_writer(&mut writer);
    }

    // Flush any buffered tail on graceful shutdown (sender dropped).
    flush_writer(&mut writer);
    tracing::debug!("event logger writer task shutting down");
}

/// Append one line to the dated event-log `BufWriter`, opening it (or reopening
/// on a date rollover) as needed. Fail-open: a failed open or write drops the
/// line with a warning rather than blocking the channel.
fn append_log_line(
    writer: &mut Option<(PathBuf, std::io::BufWriter<std::fs::File>)>,
    path: &Path,
    line: &str,
) {
    let need_open = writer.as_ref().is_none_or(|(p, _)| p.as_path() != path);
    if need_open {
        flush_writer(writer);
        match open_log_writer(path) {
            Ok(w) => *writer = Some((path.to_path_buf(), w)),
            Err(e) => {
                tracing::warn!(error = %e, "failed to open event log");
                *writer = None;
                return;
            }
        }
    }
    if let Some((_, w)) = writer.as_mut() {
        if let Err(e) = writeln!(w, "{line}") {
            tracing::warn!(error = %e, "failed to append event to log");
        }
    }
}

/// Flush the buffered writer if one is open. No-op otherwise.
fn flush_writer(writer: &mut Option<(PathBuf, std::io::BufWriter<std::fs::File>)>) {
    if let Some((_, w)) = writer.as_mut() {
        let _ = w.flush();
    }
}

/// Open the dated event-log file in append mode, wrapped in a `BufWriter`.
fn open_log_writer(path: &Path) -> std::io::Result<std::io::BufWriter<std::fs::File>> {
    Ok(std::io::BufWriter::new(open_append(path)?))
}

// ---------------------------------------------------------------------------
// Retention cleanup (LOG-03)
// ---------------------------------------------------------------------------

/// Delete log files older than `retention_days` from `log_dir`.
///
/// Two filename patterns are recognised:
/// - `events-YYYY-MM-DD.jsonl` — audit event log written by `EventLogger`
/// - `daemon.log.YYYY-MM-DD` — operational log written by
///   `tracing_appender::rolling::daily`
///
/// Files with unrecognized names are left untouched.
///
/// Returns the number of files deleted.
///
/// # Errors
///
/// Returns `std::io::Error` if the directory cannot be read or a file cannot be deleted.
pub fn cleanup_old_logs(log_dir: &Path, retention_days: u32) -> std::io::Result<u32> {
    let cutoff = chrono::Utc::now().date_naive() - chrono::Duration::days(retention_days as i64);
    let mut deleted = 0u32;

    for entry in std::fs::read_dir(log_dir)? {
        let entry = entry?;
        let name = entry.file_name();
        let name_str = name.to_string_lossy();
        let date_str = parse_dated_log_name(&name_str);
        let Some(date_str) = date_str else { continue };
        if let Ok(date) = chrono::NaiveDate::parse_from_str(date_str, "%Y-%m-%d") {
            if date < cutoff {
                std::fs::remove_file(entry.path())?;
                deleted += 1;
            }
        }
    }
    Ok(deleted)
}

/// Extract the `YYYY-MM-DD` substring from a recognised log filename.
/// Returns `None` for any other name so unrelated files (token files,
/// `.bak` siblings, etc.) are left alone.
fn parse_dated_log_name(name: &str) -> Option<&str> {
    if let Some(date_str) = name
        .strip_prefix("events-")
        .and_then(|s| s.strip_suffix(".jsonl"))
    {
        return Some(date_str);
    }
    name.strip_prefix("daemon.log.")
}

// ---------------------------------------------------------------------------
// Unit tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use std::io::BufRead;
    use tempfile::TempDir;

    fn make_date(year: i32, month: u32, day: u32) -> chrono::NaiveDate {
        chrono::NaiveDate::from_ymd_opt(year, month, day).unwrap()
    }

    // Test 1: log_file_name() for 2026-04-07 returns "events-2026-04-07.jsonl"
    #[test]
    fn test_log_file_name_format() {
        let date = make_date(2026, 4, 7);
        assert_eq!(log_file_name(&date), "events-2026-04-07.jsonl");
    }

    // Test 2: append_event writes a valid JSON line to the file (parseable by serde_json)
    #[test]
    fn test_append_event_sync_writes_valid_json() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("test.jsonl");
        let event = json!({"tool": "bash", "session": "abc123"});

        append_event_sync(&path, &event).unwrap();

        let content = std::fs::read_to_string(&path).unwrap();
        let line = content.lines().next().unwrap();
        // Must be parseable as JSON
        let parsed: serde_json::Value = serde_json::from_str(line).unwrap();
        assert_eq!(parsed["tool"], "bash");
    }

    // Test 3: append_event writes exactly one line (ends with \n, no embedded newlines)
    #[test]
    fn test_append_event_sync_single_line() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("test.jsonl");
        let event = json!({"key": "value"});

        append_event_sync(&path, &event).unwrap();

        let content = std::fs::read_to_string(&path).unwrap();
        // Exactly one line (content ends with \n and has no embedded newlines in the JSON)
        assert!(content.ends_with('\n'), "file must end with newline");
        let lines: Vec<&str> = content.lines().collect();
        assert_eq!(lines.len(), 1, "must produce exactly one line");
    }

    // Test 4: Two calls to append_event produce two lines in the file
    #[test]
    fn test_append_event_sync_two_calls_two_lines() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("test.jsonl");
        let event1 = json!({"seq": 1});
        let event2 = json!({"seq": 2});

        append_event_sync(&path, &event1).unwrap();
        append_event_sync(&path, &event2).unwrap();

        let file = std::fs::File::open(&path).unwrap();
        let lines: Vec<_> = std::io::BufReader::new(file)
            .lines()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(lines.len(), 2, "must produce exactly two lines");

        let parsed1: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
        let parsed2: serde_json::Value = serde_json::from_str(&lines[1]).unwrap();
        assert_eq!(parsed1["seq"], 1);
        assert_eq!(parsed2["seq"], 2);
    }

    // Test 5: cleanup_old_logs with retention_days=0 deletes all events-*.jsonl files
    #[test]
    fn test_cleanup_old_logs_retention_zero_deletes_all() {
        let dir = TempDir::new().unwrap();
        // Create files for yesterday and two days ago (both older than today when retention=0)
        let yesterday = chrono::Utc::now().date_naive() - chrono::Duration::days(1);
        let two_days_ago = chrono::Utc::now().date_naive() - chrono::Duration::days(2);

        std::fs::write(dir.path().join(log_file_name(&yesterday)), b"line\n").unwrap();
        std::fs::write(dir.path().join(log_file_name(&two_days_ago)), b"line\n").unwrap();

        let deleted = cleanup_old_logs(dir.path(), 0).unwrap();
        assert_eq!(deleted, 2, "retention_days=0 must delete all past files");

        // Directory should now be empty
        let remaining: Vec<_> = std::fs::read_dir(dir.path())
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert!(remaining.is_empty(), "no files should remain");
    }

    // Test 6: cleanup_old_logs with retention_days=30 keeps files newer than 30 days
    #[test]
    fn test_cleanup_old_logs_retention_keeps_recent() {
        let dir = TempDir::new().unwrap();
        let recent = chrono::Utc::now().date_naive() - chrono::Duration::days(5);
        let old = chrono::Utc::now().date_naive() - chrono::Duration::days(40);

        let recent_file = dir.path().join(log_file_name(&recent));
        let old_file = dir.path().join(log_file_name(&old));

        std::fs::write(&recent_file, b"line\n").unwrap();
        std::fs::write(&old_file, b"line\n").unwrap();

        let deleted = cleanup_old_logs(dir.path(), 30).unwrap();
        assert_eq!(deleted, 1, "only the old file should be deleted");
        assert!(recent_file.exists(), "recent file must not be deleted");
        assert!(!old_file.exists(), "old file must be deleted");
    }

    // Test 6b: cleanup_old_logs also deletes daemon.log.YYYY-MM-DD files
    // produced by tracing_appender::rolling::daily.
    #[test]
    fn test_cleanup_old_logs_deletes_old_daemon_logs() {
        let dir = TempDir::new().unwrap();
        let recent = chrono::Utc::now().date_naive() - chrono::Duration::days(5);
        let old = chrono::Utc::now().date_naive() - chrono::Duration::days(40);

        let recent_daemon = dir
            .path()
            .join(format!("daemon.log.{}", recent.format("%Y-%m-%d")));
        let old_daemon = dir
            .path()
            .join(format!("daemon.log.{}", old.format("%Y-%m-%d")));
        let recent_events = dir.path().join(log_file_name(&recent));
        let old_events = dir.path().join(log_file_name(&old));

        std::fs::write(&recent_daemon, b"line\n").unwrap();
        std::fs::write(&old_daemon, b"line\n").unwrap();
        std::fs::write(&recent_events, b"line\n").unwrap();
        std::fs::write(&old_events, b"line\n").unwrap();

        let deleted = cleanup_old_logs(dir.path(), 30).unwrap();
        assert_eq!(
            deleted, 2,
            "both old daemon and event files must be deleted"
        );
        assert!(recent_daemon.exists(), "recent daemon log must be kept");
        assert!(recent_events.exists(), "recent event log must be kept");
        assert!(!old_daemon.exists(), "old daemon log must be deleted");
        assert!(!old_events.exists(), "old event log must be deleted");
    }

    // Test 6c: cleanup_old_logs leaves unrelated files alone (e.g. the
    // currently-active daemon.log without a date suffix, sentinel files,
    // .bak siblings).
    #[test]
    fn test_cleanup_old_logs_ignores_unrelated_files() {
        let dir = TempDir::new().unwrap();
        let unrelated_files = [
            "daemon.log",
            "fallback.jsonl",
            "fallback.jsonl.offset",
            "random.txt",
        ];
        for name in &unrelated_files {
            std::fs::write(dir.path().join(name), b"x").unwrap();
        }

        let deleted = cleanup_old_logs(dir.path(), 0).unwrap();
        assert_eq!(deleted, 0, "no files matching the dated patterns to delete");
        for name in &unrelated_files {
            assert!(
                dir.path().join(name).exists(),
                "{name} must remain untouched"
            );
        }
    }

    // Test 7: EventLogger sends events via mpsc channel and writer task appends them
    #[tokio::test]
    async fn test_event_logger_sends_and_appends() {
        let dir = TempDir::new().unwrap();
        let log_dir = dir.path().to_path_buf();

        let (logger, handle) = EventLogger::new(log_dir.clone());

        let event = json!({"tool": "write_file", "session": "sess_001"});
        let event_str = serde_json::to_string(&event).unwrap();
        logger.log(event_str);

        // Drop the logger to close the sender channel, then wait for writer to drain.
        drop(logger);
        handle.shutdown().await;

        let today = chrono::Utc::now().date_naive();
        let log_path = log_dir.join(log_file_name(&today));

        assert!(log_path.exists(), "log file must be created");
        let content = std::fs::read_to_string(&log_path).unwrap();
        let parsed: serde_json::Value =
            serde_json::from_str(content.lines().next().unwrap()).unwrap();
        assert_eq!(parsed["tool"], "write_file");
    }

    // Test 8: a burst of events drains without loss — the buffered writer must
    // write every queued line and shutdown must flush the buffered tail. Guards
    // the batched-drain rewrite that fixed the startup `event log channel full`
    // drops.
    #[tokio::test]
    async fn test_event_logger_drains_burst_without_loss() {
        let dir = TempDir::new().unwrap();
        let log_dir = dir.path().to_path_buf();

        let (logger, handle) = EventLogger::new(log_dir.clone());

        const N: usize = 500;
        for i in 0..N {
            logger.log(format!(r#"{{"n":{i}}}"#));
        }

        drop(logger);
        handle.shutdown().await;

        let log_path = log_dir.join(log_file_name(&chrono::Utc::now().date_naive()));
        let content = std::fs::read_to_string(&log_path).unwrap();
        let lines: Vec<&str> = content.lines().collect();
        assert_eq!(
            lines.len(),
            N,
            "every burst line must be written and flushed"
        );

        // FIFO order preserved end to end.
        let first = serde_json::from_str::<serde_json::Value>(lines[0]).unwrap();
        let last = serde_json::from_str::<serde_json::Value>(lines[N - 1]).unwrap();
        assert_eq!(first["n"].as_i64(), Some(0));
        assert_eq!(last["n"].as_i64(), Some((N - 1) as i64));
    }
}