wsb 0.253.6349

//! One operation's output, on disk, in the format it arrived in.
//!
//! # What the file is
//!
//! JSONL, and every line but the first and last is exactly what the child
//! process emitted on its packaged stream — an `wsb::output::Line`, seq, ts and
//! record. Nothing is reformatted on the way in and nothing is interpreted on
//! the way out, so the run a page folds from a live operation and the run it
//! folds from a file three days later are folded from the same records.
//!
//! The first line is a header and the last is a footer, both marked with a
//! `log` key that no record has:
//!
//! ```text
//! {"log":"header","id":"…","kind":"test_run","label":"the declared suite",…}
//! {"seq":1,"ts":"…","type":"command_started",…}
//! …
//! {"log":"footer","status":"completed","ended_at":"…"}
//! ```
//!
//! The header is what lets the daemon rebuild its history from the directory
//! after a restart: an operation that ran yesterday can be listed, opened and
//! read without the process that ran it ever having existed. A file with no
//! footer is an operation that did not finish — the daemon was killed while it
//! ran — and it says so rather than reading as a run that ended cleanly.
//!
//! # Why not the database
//!
//! Because the log is the one artifact somebody wants OUTSIDE the workbench:
//! pasted into an issue, grepped from a terminal, tailed while a build runs,
//! sent to somebody who does not have this daemon. A row in a database is none
//! of those, and the page already has to tell an operator where the log is.

use crate::prelude::*;
use crate::output::Line;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

/// The format marker. Bumped only if the shape of a line changes; a reader
/// that meets a version it does not know says so instead of guessing.
const FORMAT: &str = "wsb-op-log/1";

/// How an operation ended, as the footer records it.
///
/// Two states, not three: a log with a footer is a log of something that
/// finished. "Running" is the ABSENCE of a footer, which is a fact about the
/// file rather than a value written into it — and it is the only encoding that
/// survives the process being killed, which is exactly when it matters.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Status {
    Completed,
    Failed,
}

/// The first line of an operation's log: what ran, and what started it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Header {
    /// Always `"header"`.
    pub log: String,
    pub format: String,
    pub id: uuid::Uuid,
    /// What kind of operation this was, in the vocabulary of whoever started
    /// it — the workbench's operation kinds, or a command name from a command
    /// line. A string because the two vocabularies are not one enum and
    /// pretending they were would make one of them lossy.
    pub kind: String,
    pub label: String,
    /// The command line, as it was actually run.
    pub argv: Vec<String>,
    pub started_at: chrono::DateTime<chrono::Utc>,
}

/// The last line: how it ended. Absent when the daemon died mid-run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Footer {
    /// Always `"footer"`.
    pub log: String,
    pub status: Status,
    pub ended_at: chrono::DateTime<chrono::Utc>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    /// Lines written. A reader can say "3 of 4210 lines shown" without
    /// counting the file.
    pub lines: u64,
}

/// An open log file, appended to as the operation speaks.
///
/// Writes are buffered by the OS and flushed per line: an operation that is
/// still running is a file somebody may be tailing in a terminal, and a log
/// that appears twenty lines behind the page is one that gets called broken.
pub struct OpLogWriter {
    path: PathBuf,
    file: Mutex<Option<std::fs::File>>,
    lines: std::sync::atomic::AtomicU64,
}

impl OpLogWriter {
    /// Create the file and write its header.
    ///
    /// A failure to open is returned, never swallowed: the caller decides
    /// whether an operation runs without a log (it does — losing the output
    /// file must not lose the operation) and says so where somebody sees it.
    pub fn create(
        logs_root: &Path,
        id: uuid::Uuid,
        kind: &str,
        label: &str,
        argv: &[String],
        started_at: chrono::DateTime<chrono::Utc>,
    ) -> Result<Self, String> {
        let path = super::op_log_path(logs_root, kind, &id, started_at);
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|e| format!("cannot create {}: {e}", parent.display()))?;
        }
        let mut file = std::fs::File::create(&path)
            .map_err(|e| format!("cannot create {}: {e}", path.display()))?;
        let header = Header {
            log: "header".to_string(),
            format: FORMAT.to_string(),
            id,
            kind: kind.to_string(),
            label: label.to_string(),
            argv: argv.to_vec(),
            started_at,
        };
        writeln!(
            file,
            "{}",
            serde_json::to_string(&header).map_err(|e| e.to_string())?
        )
        .map_err(|e| format!("cannot write to {}: {e}", path.display()))?;
        Ok(Self {
            path,
            file: Mutex::new(Some(file)),
            lines: std::sync::atomic::AtomicU64::new(0),
        })
    }

    /// Where this operation's output is, absolutely — the path the page shows.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Append one captured line.
    pub fn write_line(&self, line: &Line) {
        let Ok(encoded) = serde_json::to_string(line) else {
            return;
        };
        let mut guard = match self.file.lock() {
            Ok(guard) => guard,
            // A poisoned lock means a previous write panicked. The file is
            // still open and still correct up to that point; refusing to
            // write any more of it is the honest response.
            Err(_) => return,
        };
        let Some(file) = guard.as_mut() else {
            return;
        };
        if let Err(error) = writeln!(file, "{encoded}") {
            tracing::warn!(path = %self.path.display(), %error, "an operation's log could not be written");
            // One failed write closes the file: a log with a hole in it is
            // worse than a log that says where it stopped, because nothing
            // downstream can tell the two apart.
            *guard = None;
            return;
        }
        let _ = file.flush();
        self.lines
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    }

    /// Write the footer and close. Called however the operation ended.
    pub fn finish(&self, status: Status, error: Option<String>) {
        let footer = Footer {
            log: "footer".to_string(),
            status,
            ended_at: chrono::Utc::now(),
            error,
            lines: self.lines.load(std::sync::atomic::Ordering::Relaxed),
        };
        let mut guard = match self.file.lock() {
            Ok(guard) => guard,
            // A poisoned lock means a previous write panicked. The file is
            // still open and still correct up to that point; refusing to
            // write any more of it is the honest response.
            Err(_) => return,
        };
        let Some(file) = guard.as_mut() else {
            return;
        };
        if let Ok(encoded) = serde_json::to_string(&footer) {
            let _ = writeln!(file, "{encoded}");
            let _ = file.flush();
        }
        *guard = None;
    }
}

/// One operation's log, read back from disk.
#[derive(Debug, Clone)]
pub struct ReadLog {
    pub header: Header,
    pub footer: Option<Footer>,
    pub lines: Vec<Line>,
    /// Lines that were not readable as records — a truncated final line from a
    /// daemon that was killed mid-write, most often. Counted, not hidden.
    pub unreadable: u64,
    pub path: PathBuf,
}

/// Read a whole operation log.
///
/// Everything is read: the file exists precisely so that the ring buffer's
/// limit does not apply to it, and a reader that re-imposed a limit would
/// reintroduce the truncation the file was written to avoid.
pub fn read(path: &Path) -> Result<ReadLog, String> {
    let content = std::fs::read_to_string(path)
        .map_err(|e| format!("cannot read {}: {e}", path.display()))?;
    let mut header: Option<Header> = None;
    let mut footer: Option<Footer> = None;
    let mut lines = Vec::new();
    let mut unreadable = 0u64;

    for raw in content.lines() {
        if raw.trim().is_empty() {
            continue;
        }
        // The envelope lines are the ones carrying `log`; everything else is a
        // record. Checked by parsing rather than by looking for a substring:
        // a record whose text mentions the word would otherwise be swallowed.
        match serde_json::from_str::<serde_json::Value>(raw) {
            Ok(value) => match value.get("log").and_then(serde_json::Value::as_str) {
                Some("header") => header = serde_json::from_value(value).ok(),
                Some("footer") => footer = serde_json::from_value(value).ok(),
                _ => match serde_json::from_value::<Line>(value) {
                    Ok(line) => lines.push(line),
                    Err(_) => unreadable += 1,
                },
            },
            Err(_) => unreadable += 1,
        }
    }

    let header = header.ok_or_else(|| {
        format!(
            "{} has no header line, so nothing can say what it is a log of",
            path.display()
        )
    })?;
    if header.format != FORMAT {
        return Err(format!(
            "{} is in format {}, which this daemon does not read (it writes {FORMAT})",
            path.display(),
            header.format
        ));
    }
    Ok(ReadLog {
        header,
        footer,
        lines,
        unreadable,
        path: path.to_path_buf(),
    })
}

/// Read only the envelope of a log — what it was, and how it ended.
///
/// This is what rebuilding history after a restart needs. Reading the whole of
/// every file to list them would mean parsing hundreds of megabytes to draw a
/// list nobody has clicked on yet.
pub fn read_envelope(path: &Path) -> Result<(Header, Option<Footer>), String> {
    use std::io::{BufRead, BufReader, Seek, SeekFrom};

    let file = std::fs::File::open(path)
        .map_err(|e| format!("cannot open {}: {e}", path.display()))?;
    let size = file
        .metadata()
        .map_err(|e| format!("cannot stat {}: {e}", path.display()))?
        .len();
    let mut reader = BufReader::new(file);
    let mut first = String::new();
    reader
        .read_line(&mut first)
        .map_err(|e| format!("cannot read {}: {e}", path.display()))?;
    let header: Header = serde_json::from_str(first.trim()).map_err(|_| {
        format!(
            "{} does not begin with a log header",
            path.display()
        )
    })?;

    // The footer is the last line. Read a tail rather than the file: a
    // half-gigabyte build log would otherwise be read in full to answer
    // "did it finish".
    const TAIL: u64 = 8 * 1024;
    let from = size.saturating_sub(TAIL);
    let mut file = reader.into_inner();
    let footer = file
        .seek(SeekFrom::Start(from))
        .ok()
        .and_then(|_| {
            let mut tail = String::new();
            use std::io::Read;
            file.read_to_string(&mut tail).ok().map(|_| tail)
        })
        .and_then(|tail| {
            tail.lines()
                .rev()
                .find_map(|line| serde_json::from_str::<Footer>(line).ok())
        })
        .filter(|f: &Footer| f.log == "footer");

    Ok((header, footer))
}

/// Every operation log under the root, newest first.
///
/// Bounded by `limit` because this runs at startup, in front of a page waiting
/// to be drawn: a year of operations is a directory listing, not a list
/// anybody scrolls.
pub fn recent_logs(logs_root: &Path, limit: usize) -> Vec<PathBuf> {
    let ops_dir = super::stream_dir(logs_root, super::Stream::Ops);
    let mut days: Vec<PathBuf> = std::fs::read_dir(&ops_dir)
        .into_iter()
        .flatten()
        .flatten()
        .map(|entry| entry.path())
        .filter(|path| path.is_dir())
        .collect();
    // The dated directories sort lexically into date order, newest last.
    days.sort();

    let mut found = Vec::new();
    for day in days.into_iter().rev() {
        let mut files: Vec<PathBuf> = std::fs::read_dir(&day)
            .into_iter()
            .flatten()
            .flatten()
            .map(|entry| entry.path())
            .filter(|path| path.extension().is_some_and(|ext| ext == "jsonl"))
            .collect();
        files.sort();
        for path in files.into_iter().rev() {
            found.push(path);
            if found.len() >= limit {
                return found;
            }
        }
    }
    found
}

#[cfg(test)]
#[path = "op_log_tests.rs"]
mod tests;