hayson-engine 0.17.0

use std::fs;
use std::io::{self, BufRead, BufReader};
use std::path::{Path, PathBuf};

use serde_json::Value;

use crate::catalog::{parse_catalog, Catalog, TableEntry};
use crate::storage::{parse_entry, partition_names_in, LOG_SUFFIX};

use super::database::{Database, LockMode, OpenOptions};
use super::layout::{Layout, LayoutOptions};

/// How much a finding matters to someone about to open the directory.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Severity {
    /// The directory cannot be opened as it is, or data in it cannot be read back.
    Error,
    /// Opening works; the engine repairs it on the next writable open, or it is only worth a look.
    Warning,
}

/// One finding of [`verify`]. `code` is a stable kebab-case identifier callers can match on;
/// `message` is for people.
#[derive(Debug, Clone, PartialEq)]
pub struct Problem {
    pub severity: Severity,
    pub code: &'static str,
    pub table: Option<String>,
    /// Path of the file concerned, relative to the database directory.
    pub file: Option<String>,
    /// Byte offset of the offending line in `file`.
    pub offset: Option<u64>,
    pub message: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VerifyOptions {
    /// Also open the database read-only (no lock) and report relation orphans. Skipped when the
    /// structural checks already found an error, because the open would only repeat it.
    pub relations: bool,
}

impl Default for VerifyOptions {
    fn default() -> Self {
        VerifyOptions { relations: true }
    }
}

#[derive(Debug, Clone, PartialEq, Default)]
pub struct VerifyReport {
    pub problems: Vec<Problem>,
    pub tables_checked: usize,
    pub log_lines_checked: u64,
}

impl VerifyReport {
    pub fn has_errors(&self) -> bool {
        self.problems.iter().any(|p| p.severity == Severity::Error)
    }

    pub fn is_clean(&self) -> bool {
        self.problems.is_empty()
    }
}

/// Checks a database directory without taking its lock or changing a byte, and reports what is
/// wrong with it instead of failing on it: every damaged, missing or unreadable file becomes a
/// [`Problem`], never an `Err` and never a panic.
///
/// Looks at the metadata document, each declared table's log files line by line, logs the catalog
/// does not declare, an unfinished unit-of-work journal and (optionally) declared relations.
pub fn verify(dir: &Path, options: &VerifyOptions) -> VerifyReport {
    let mut report = VerifyReport::default();
    let mut checker = Checker {
        root: dir.to_path_buf(),
        report: &mut report,
    };
    checker.run(options);
    report
}

struct Checker<'a> {
    root: PathBuf,
    report: &'a mut VerifyReport,
}

impl Checker<'_> {
    fn push(&mut self, severity: Severity, code: &'static str, message: String) -> &mut Problem {
        self.report.problems.push(Problem {
            severity,
            code,
            table: None,
            file: None,
            offset: None,
            message,
        });
        self.report.problems.last_mut().expect("just pushed")
    }

    fn relative(&self, path: &Path) -> String {
        path.strip_prefix(&self.root).unwrap_or(path).display().to_string()
    }

    fn run(&mut self, options: &VerifyOptions) {
        if !self.root.is_dir() {
            let message = format!("\"{}\" is not a directory", self.root.display());
            self.push(Severity::Error, "directory-missing", message);
            return;
        }
        let layout = match Database::resolve_layout(&self.root, &LayoutOptions::default()) {
            Ok(layout) => layout,
            Err(error) => {
                self.push(Severity::Error, "layout-unresolved", error.to_string());
                return;
            }
        };
        let Some(catalog) = self.read_catalog(&layout) else {
            return;
        };

        for (name, entry) in &catalog.tables {
            self.check_table(&layout, name, entry);
        }
        self.check_orphan_logs(&layout, &catalog);
        self.check_journal(&layout);

        if options.relations && !self.report.has_errors() {
            self.check_relations();
        }
    }

    fn read_catalog(&mut self, layout: &Layout) -> Option<Catalog> {
        let path = layout.metadata_path();
        let file = self.relative(path);
        let raw = match fs::read_to_string(path) {
            Ok(raw) => raw,
            Err(error) if error.kind() == io::ErrorKind::NotFound => {
                let holds_logs = !self.log_files_under(layout).is_empty();
                let severity = if holds_logs { Severity::Error } else { Severity::Warning };
                self.push(severity, "catalog-missing", format!("\"{file}\" does not exist"))
                    .file = Some(file.clone());
                return None;
            }
            Err(error) => {
                self.push(Severity::Error, "catalog-unreadable", format!("\"{file}\": {error}"))
                    .file = Some(file.clone());
                return None;
            }
        };
        let value: Value = match serde_json::from_str(&raw) {
            Ok(value) => value,
            Err(error) => {
                self.push(
                    Severity::Error,
                    "catalog-invalid-json",
                    format!("\"{file}\" is not valid JSON: {error}"),
                )
                .file = Some(file.clone());
                return None;
            }
        };
        match parse_catalog(&value) {
            Ok(catalog) => Some(catalog),
            Err(error) => {
                self.push(Severity::Error, "catalog-invalid", format!("\"{file}\": {error}"))
                    .file = Some(file.clone());
                None
            }
        }
    }

    /// Every `*.jsonl` file directly in the tables directory or one level below it.
    fn log_files_under(&self, layout: &Layout) -> Vec<PathBuf> {
        let is_log = |path: &Path| path.is_file() && path.to_string_lossy().ends_with(LOG_SUFFIX);
        let mut found = Vec::new();
        let Ok(top) = fs::read_dir(layout.tables()) else {
            return found;
        };
        for path in top.filter_map(Result::ok).map(|entry| entry.path()) {
            if is_log(&path) {
                found.push(path);
            } else if path.is_dir() {
                if let Ok(inner) = fs::read_dir(&path) {
                    found.extend(
                        inner
                            .filter_map(Result::ok)
                            .map(|entry| entry.path())
                            .filter(|p| is_log(p)),
                    );
                }
            }
        }
        found.sort();
        found
    }

    fn logs_of(&self, layout: &Layout, name: &str, entry: &TableEntry) -> Vec<PathBuf> {
        let flat = layout.log_path(name);
        if entry.partition_by.is_some() {
            let dir = layout.tables().join(name);
            let parts = partition_names_in(&dir);
            if !parts.is_empty() {
                return parts
                    .iter()
                    .map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
                    .collect();
            }
        }
        if fs::symlink_metadata(&flat).is_ok() {
            vec![flat]
        } else {
            Vec::new()
        }
    }

    fn check_table(&mut self, layout: &Layout, name: &str, entry: &TableEntry) {
        self.report.tables_checked += 1;
        for path in self.logs_of(layout, name, entry) {
            self.check_log(name, &path);
        }
    }

    fn check_log(&mut self, table: &str, path: &Path) {
        let file = self.relative(path);
        let opened = match fs::File::open(path) {
            Ok(opened) => opened,
            Err(error) => {
                let problem = self.push(Severity::Error, "log-unreadable", format!("\"{file}\": {error}"));
                problem.table = Some(table.to_string());
                problem.file = Some(file.clone());
                return;
            }
        };
        let mut reader = BufReader::new(opened);
        let mut buffer = Vec::new();
        let mut offset = 0u64;
        loop {
            buffer.clear();
            let read = match reader.read_until(b'\n', &mut buffer) {
                Ok(0) => break,
                Ok(read) => read,
                Err(error) => {
                    let problem = self.push(
                        Severity::Error,
                        "log-unreadable",
                        format!("\"{file}\" stopped being readable at byte {offset}: {error}"),
                    );
                    problem.table = Some(table.to_string());
                    problem.file = Some(file.clone());
                    problem.offset = Some(offset);
                    return;
                }
            };
            let terminated = buffer.last() == Some(&b'\n');
            if !terminated {
                let problem = self.push(
                    Severity::Warning,
                    "log-truncated-tail",
                    format!("\"{file}\" ends with {read} bytes of an unfinished line at byte {offset}; a writable open cuts it off"),
                );
                problem.table = Some(table.to_string());
                problem.file = Some(file.clone());
                problem.offset = Some(offset);
                return;
            }
            let content = &buffer[..buffer.len() - 1];
            if !content.is_empty() {
                self.report.log_lines_checked += 1;
                let reason = match std::str::from_utf8(content) {
                    Ok(line) => parse_entry(line, &file, offset).err().map(|error| error.reason),
                    Err(error) => Some(format!("line is not valid UTF-8 ({error})")),
                };
                if let Some(reason) = reason {
                    let problem = self.push(
                        Severity::Error,
                        "log-corrupt-line",
                        format!("\"{file}\" at byte {offset}: {reason}"),
                    );
                    problem.table = Some(table.to_string());
                    problem.file = Some(file.clone());
                    problem.offset = Some(offset);
                }
            }
            offset += read as u64;
        }
    }

    fn check_orphan_logs(&mut self, layout: &Layout, catalog: &Catalog) {
        for path in self.log_files_under(layout) {
            let Some(table) = table_of_log(layout, &path) else {
                continue;
            };
            if !catalog.tables.contains_key(&table) {
                let file = self.relative(&path);
                let problem = self.push(
                    Severity::Warning,
                    "log-undeclared",
                    format!("\"{file}\" belongs to table \"{table}\", which the catalog does not declare"),
                );
                problem.table = Some(table);
                problem.file = Some(file.clone());
            }
        }
    }

    fn check_journal(&mut self, layout: &Layout) {
        let path = layout.journal_path();
        if fs::symlink_metadata(&path).is_ok() {
            let file = self.relative(&path);
            self.push(
                Severity::Warning,
                "journal-unfinished",
                format!(
                    "\"{file}\" is a unit of work that never finished; the next writable open completes or discards it"
                ),
            )
            .file = Some(file.clone());
        }
    }

    fn check_relations(&mut self) {
        let options = OpenOptions {
            readonly: true,
            lock: LockMode::Skip,
            ..OpenOptions::default()
        };
        let database = match Database::open_with(&self.root, options) {
            Ok(database) => database,
            Err(error) => {
                self.push(Severity::Error, "open-failed", error.to_string());
                return;
            }
        };
        match database.check_integrity() {
            Ok(integrity) => {
                for orphan in integrity.orphans {
                    let problem = self.push(
                        Severity::Warning,
                        "relation-orphan",
                        format!(
                            "relation \"{}\": {}.{} = {} of record {} points at nothing",
                            orphan.relation, orphan.table, orphan.field, orphan.value, orphan.primary_key
                        ),
                    );
                    problem.table = Some(orphan.table);
                }
            }
            Err(error) => {
                self.push(Severity::Error, "integrity-check-failed", error.to_string());
            }
        }
    }
}

/// The table a log file belongs to: `<table>.jsonl`, or `<table>/<part>.jsonl` one level down.
fn table_of_log(layout: &Layout, path: &Path) -> Option<String> {
    let relative = path.strip_prefix(layout.tables()).ok()?;
    let mut parts = relative.components();
    let first = parts.next()?.as_os_str().to_string_lossy().to_string();
    if parts.next().is_some() {
        Some(first)
    } else {
        first.strip_suffix(LOG_SUFFIX).map(str::to_string)
    }
}