hayson-engine 0.27.0

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

use serde_json::Value;

use crate::catalog::{parse_catalog, read_catalog_text, Catalog, ReadCatalogError, TableEntry};
use crate::crypto::{CipherOptions, Sealer};
use crate::schema::physical_path_of;
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, 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,
    /// Also check every table's image files: a referenced file that is missing, a file whose bytes
    /// no longer hash to its name, a file no record references and, with `max_image_bytes`, a file
    /// over that size. Reads every image file, and shares the read-only open `relations` uses.
    pub images: bool,
    /// Size in bytes above which an existing image file is reported as `image-oversize`.
    pub max_image_bytes: Option<u64>,
    /// The keys of an encrypted database. Without them its catalog cannot be opened, and verifying
    /// it ends with one `catalog-encrypted` problem; with them every sealed line and image is
    /// opened and checked like a plain one.
    pub encryption: Option<CipherOptions>,
}

impl Default for VerifyOptions {
    fn default() -> Self {
        VerifyOptions {
            relations: true,
            images: true,
            max_image_bytes: None,
            encryption: None,
        }
    }
}

#[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 sealer = match options.encryption.as_ref().map(Sealer::new).transpose() {
        Ok(sealer) => sealer.map(Arc::new),
        Err(error) => {
            report.problems.push(Problem {
                severity: Severity::Error,
                code: "encryption-key-invalid",
                table: None,
                file: None,
                offset: None,
                message: error.to_string(),
            });
            return report;
        }
    };
    let mut checker = Checker {
        root: dir.to_path_buf(),
        report: &mut report,
        sealer,
    };
    checker.run(options);
    report
}

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

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_sealed(&self.root, &LayoutOptions::default(), self.sealer.as_deref()) {
                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 || options.images) && !self.report.has_errors() {
            self.check_open(options);
        }
    }

    fn read_catalog(&mut self, layout: &Layout) -> Option<Catalog> {
        let path = layout.metadata_path();
        let file = self.relative(path);
        let raw = match read_catalog_text(path, self.sealer.as_deref()) {
            Ok(Some(raw)) => raw,
            Ok(None) => {
                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) => {
                let code = error.problem_code();
                let reason = match error {
                    ReadCatalogError::Io(error) => error.to_string(),
                    ReadCatalogError::Cipher(error)
                    | ReadCatalogError::Encrypted(error)
                    | ReadCatalogError::NotEncrypted(error) => error.to_string(),
                };
                self.push(Severity::Error, code, format!("\"{file}\": {reason}")).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() && !path.join(layout.metadata_file()).is_file() {
                // NOTE: a folder with its own metadata document is another database, not a partition directory.
                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) => match &self.sealer {
                        Some(sealer) => match sealer.open_line(&format!("log:{table}"), line) {
                            Ok(plain) => parse_entry(&plain, &file, offset).err().map(|error| error.reason),
                            Err(error) => Some(error.reason),
                        },
                        None => 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_images(&mut self, database: &Database, verify: &VerifyOptions) {
        for name in database.table_names() {
            let Some(table) = database.table(name) else {
                continue;
            };
            let images = match table.list_images() {
                Ok(images) => images,
                Err(error) => {
                    let problem = self.push(
                        Severity::Error,
                        "image-unreadable",
                        format!("table \"{name}\": {error}"),
                    );
                    problem.table = Some(name.to_string());
                    continue;
                }
            };
            for image in images {
                let file = self.relative(&table.image_dir().join(&image.physical_path));
                let mut report = |checker: &mut Self, severity, code, message: String| {
                    let problem = checker.push(severity, code, message);
                    problem.table = Some(name.to_string());
                    problem.file = Some(file.clone());
                };
                if image.is_missing() {
                    let who = &image.referenced_by[0];
                    report(
                        self,
                        Severity::Error,
                        "image-missing",
                        format!(
                            "record \"{}\" (field \"{}\") references \"{}\", which is not on disk",
                            who.key, who.field, image.physical_path
                        ),
                    );
                    continue;
                }
                match table.read_image(&image.physical_path) {
                    // NOTE: in an encrypted database the seal's own tag already authenticates the bytes.
                    Ok(Some(bytes)) if self.sealer.is_none() && physical_path_of(&bytes) != image.physical_path => {
                        report(
                            self,
                            Severity::Error,
                            "image-corrupt",
                            format!("\"{}\" no longer hashes to its own name", image.physical_path),
                        )
                    }
                    Ok(_) => {}
                    Err(error) => report(
                        self,
                        Severity::Error,
                        "image-unreadable",
                        format!("\"{}\": {error}", image.physical_path),
                    ),
                }
                if image.referenced_by.is_empty() {
                    report(
                        self,
                        Severity::Warning,
                        "image-orphan",
                        format!("\"{}\" is referenced by no record", image.physical_path),
                    );
                }
                if verify.max_image_bytes.is_some_and(|max| image.bytes > max) {
                    report(
                        self,
                        Severity::Warning,
                        "image-oversize",
                        format!(
                            "\"{}\" is {} bytes, over the configured maximum",
                            image.physical_path, image.bytes
                        ),
                    );
                }
            }
        }
    }

    fn check_open(&mut self, verify: &VerifyOptions) {
        let options = OpenOptions {
            readonly: true,
            lock: LockMode::Skip,
            encryption: verify.encryption.clone(),
            ..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;
            }
        };
        if verify.images {
            self.check_images(&database, verify);
        }
        if !verify.relations {
            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)
    }
}