hayson-engine 0.24.0

use std::collections::BTreeSet;
use std::fs;
use std::io;
use std::path::{Component, Path, PathBuf};

use serde_json::{json, Value};

use crate::atomic::write_atomic;
use crate::catalog::{parse_catalog, Catalog};
use crate::errors::HaysonSchemaError;
use crate::lock::DirLock;
use crate::storage::{partition_names_in, LOG_SUFFIX};

use super::database::{Database, DatabaseError, HeldLock, LockMode};
use super::layout::{Layout, LayoutOptions};
use super::verify::{verify, VerifyOptions};

/// The manifest a backup carries next to the files it copied.
pub const BACKUP_MANIFEST: &str = "hayson-backup.json";
const BACKUP_FORMAT: u64 = 1;
const RESTORE_OLD: &str = ".restore-old";
const RESTORE_NEW: &str = ".restore-new";

#[derive(Debug, Clone)]
pub struct BackupOptions {
    /// How the writer lock is taken for the length of the copy. [`LockMode::Skip`] copies a database
    /// a writer may be appending to, which can capture a half-written last line (the engine cuts
    /// such a tail off on the next writable open) — but nothing worse.
    pub lock: LockMode,
    /// Also copy the `<table>.idx` snapshots. Off by default: they are caches the engine rebuilds
    /// from the logs, so leaving them out makes a backup smaller and no less complete.
    pub indexes: bool,
    /// Tables whose data (log files, partitions, images, snapshot) is left out; their declarations
    /// stay in the metadata document, so the copy opens with them empty. A name the database does
    /// not declare is refused.
    pub exclude_tables: Vec<String>,
}

impl Default for BackupOptions {
    fn default() -> Self {
        BackupOptions {
            lock: LockMode::default(),
            indexes: false,
            exclude_tables: Vec::new(),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BackupResult {
    pub dest: PathBuf,
    pub files: usize,
    pub bytes: u64,
}

#[derive(Debug, Clone, Default)]
pub struct RestoreOptions {
    /// How the writer lock on the directory being restored into is taken. As in [`BackupOptions`].
    pub lock: LockMode,
    /// Refuses a backup holding an image file larger than this, before anything is replaced.
    pub max_image_bytes: Option<u64>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RestoreResult {
    pub dir: PathBuf,
    pub files: usize,
    pub bytes: u64,
    /// Files of the database that was there before and were replaced.
    pub replaced: usize,
}

pub(super) fn refuse(message: String) -> DatabaseError {
    DatabaseError::Schema(HaysonSchemaError(message))
}

pub(super) fn read_catalog_of(layout: &Layout) -> Result<Catalog, DatabaseError> {
    let path = layout.metadata_path();
    let raw = fs::read_to_string(path).map_err(|error| {
        if error.kind() == io::ErrorKind::NotFound {
            refuse(format!(
                "\"{}\" does not exist — there is no database to back up",
                path.display()
            ))
        } else {
            DatabaseError::Io(error)
        }
    })?;
    let value: Value = serde_json::from_str(&raw)
        .map_err(|error| DatabaseError::Catalog(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
    Ok(parse_catalog(&value)?)
}

fn relative_safely(root: &Path, path: &Path) -> Result<PathBuf, DatabaseError> {
    let relative = path
        .strip_prefix(root)
        .map_err(|_| refuse(format!("\"{}\" is outside the database directory", path.display())))?;
    check_relative(relative)?;
    Ok(relative.to_path_buf())
}

/// A path a backup or a restore may join onto a root: relative, no `..`, no root, no prefix.
pub(super) fn check_relative(path: &Path) -> Result<(), DatabaseError> {
    let sane = path.components().all(|part| matches!(part, Component::Normal(_)));
    if sane && path.components().next().is_some() {
        Ok(())
    } else {
        Err(refuse(format!(
            "\"{}\" is not a path inside the database directory",
            path.display()
        )))
    }
}

fn files_in(dir: &Path) -> Result<Vec<PathBuf>, DatabaseError> {
    let mut found = Vec::new();
    let Ok(entries) = fs::read_dir(dir) else {
        return Ok(found);
    };
    for entry in entries {
        let entry = entry?;
        let kind = entry.file_type()?;
        if kind.is_symlink() {
            return Err(refuse(format!(
                "\"{}\" is a symbolic link — refusing to follow it",
                entry.path().display()
            )));
        }
        if kind.is_file() {
            found.push(entry.path());
        }
    }
    found.sort();
    Ok(found)
}

/// Every file the database at `root` accounts for, as paths relative to `root`: its metadata
/// document, the logs of its tables (one file, or one per partition), the files under each table's
/// image directory and, when asked, the index snapshots. Anything else in the directory — another
/// database in a subfolder, say — is not the database's and is left out.
pub(super) fn owned_files(
    root: &Path,
    layout: &Layout,
    catalog: &Catalog,
    indexes: bool,
    exclude: &[String],
) -> Result<Vec<PathBuf>, DatabaseError> {
    if let Some(unknown) = exclude.iter().find(|name| !catalog.tables.contains_key(*name)) {
        return Err(refuse(format!(
            "cannot leave out table \"{unknown}\" — the database declares no such table"
        )));
    }
    let mut owned: BTreeSet<PathBuf> = BTreeSet::new();
    let mut add = |path: &Path| -> Result<(), DatabaseError> {
        owned.insert(relative_safely(root, path)?);
        Ok(())
    };
    add(layout.metadata_path())?;
    for (name, entry) in &catalog.tables {
        if exclude.contains(name) {
            continue;
        }
        let flat = layout.log_path(name);
        match fs::symlink_metadata(&flat) {
            Ok(meta) if meta.file_type().is_symlink() => {
                return Err(refuse(format!(
                    "\"{}\" is a symbolic link — refusing to follow it",
                    flat.display()
                )))
            }
            Ok(meta) if meta.is_file() => add(&flat)?,
            _ => {}
        }
        let partitions = layout.tables().join(name);
        for part in partition_names_in(&partitions) {
            add(&partitions.join(format!("{part}{LOG_SUFFIX}")))?;
        }
        if indexes && layout.index_path(name).is_file() {
            add(&layout.index_path(name))?;
        }
        let image_dir = entry
            .options
            .as_ref()
            .and_then(|options| options.image_dir.clone())
            .or_else(|| catalog.defaults.as_ref().and_then(|d| d.image_dir.clone()))
            .map(|dir| root.join(dir))
            .unwrap_or_else(|| layout.tables().join(format!("{name}.images")));
        for file in files_in(&image_dir)? {
            add(&file)?;
        }
    }
    Ok(owned.into_iter().collect())
}

/// The flat log files of the tables `exclude` leaves out, as paths relative to `root`. They are
/// packed or copied empty: the catalog still declares those tables, and an unpacked database whose
/// declared table has no log at all does not open read-only.
pub(super) fn emptied_logs(root: &Path, layout: &Layout, catalog: &Catalog, exclude: &[String]) -> Vec<PathBuf> {
    exclude
        .iter()
        .filter(|name| {
            catalog
                .tables
                .get(*name)
                .is_some_and(|entry| entry.partition_by.is_none())
        })
        .filter_map(|name| relative_safely(root, &layout.log_path(name)).ok())
        .collect()
}

pub(super) fn hold(root: &Path, layout: &Layout, lock: &LockMode) -> Result<HeldLock, DatabaseError> {
    let LockMode::Enabled(options) = lock else {
        return Ok(HeldLock(None));
    };
    if root.is_dir() {
        fs::create_dir_all(layout.runtime())?;
    }
    Ok(HeldLock(Some(DirLock::acquire_as(
        layout.lock_path(),
        options.clone(),
        root.to_path_buf(),
    )?)))
}

fn copy_synced(from: &Path, to: &Path) -> io::Result<u64> {
    if let Some(parent) = to.parent() {
        fs::create_dir_all(parent)?;
    }
    let bytes = fs::copy(from, to)?;
    fs::File::open(to)?.sync_all()?;
    Ok(bytes)
}

/// Copies a database directory into `dest`: the files the database accounts for (see
/// [`owned_files`]) plus a `hayson-backup.json` manifest listing them with their sizes.
///
/// `dest` must not exist or must be an empty directory. The copy is built next to it and renamed
/// into place, so a backup that fails halfway leaves no half-filled `dest`. The directory is held
/// under the writer lock for the length of the copy unless `options.lock` is [`LockMode::Skip`];
/// a database with an unfinished unit of work is refused, because its log and journal disagree
/// until a writable open finishes it.
///
/// History kept in a database of its own is not part of this database's files, so it is not copied.
pub fn backup_database(dir: &Path, dest: &Path, options: &BackupOptions) -> Result<BackupResult, DatabaseError> {
    let layout = Database::resolve_layout(dir, &LayoutOptions::default())?;
    let catalog = read_catalog_of(&layout)?;
    if dest.exists()
        && fs::read_dir(dest)
            .map(|mut entries| entries.next().is_some())
            .unwrap_or(true)
    {
        return Err(refuse(format!(
            "\"{}\" already exists and is not an empty directory",
            dest.display()
        )));
    }
    let partial = sibling(dest, ".partial");
    if partial.exists() {
        fs::remove_dir_all(&partial)?;
    }

    let held = hold(dir, &layout, &options.lock)?;
    let outcome = copy_database(dir, &layout, &catalog, &partial, options);
    drop(held);
    match outcome {
        Ok(result) => {
            if dest.exists() {
                fs::remove_dir(dest)?;
            }
            fs::rename(&partial, dest)?;
            Ok(BackupResult {
                dest: dest.to_path_buf(),
                ..result
            })
        }
        Err(error) => {
            let _ = fs::remove_dir_all(&partial);
            Err(error)
        }
    }
}

pub(super) fn sibling(path: &Path, suffix: &str) -> PathBuf {
    let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
    name.push(suffix);
    path.with_file_name(name)
}

fn copy_database(
    root: &Path,
    layout: &Layout,
    catalog: &Catalog,
    partial: &Path,
    options: &BackupOptions,
) -> Result<BackupResult, DatabaseError> {
    if layout.journal_path().exists() {
        return Err(refuse(format!(
            "\"{}\" holds an unfinished unit of work — open the database for writing once so it is finished, then back it up",
            layout.journal_path().display()
        )));
    }
    let mut files = owned_files(root, layout, catalog, options.indexes, &options.exclude_tables)?;
    let emptied = emptied_logs(root, layout, catalog, &options.exclude_tables);
    files.extend(emptied.iter().cloned());
    files.sort();
    fs::create_dir_all(partial)?;
    let mut listed = Vec::new();
    let mut total = 0u64;
    for relative in &files {
        let bytes = if emptied.contains(relative) {
            let target = partial.join(relative);
            if let Some(parent) = target.parent() {
                fs::create_dir_all(parent)?;
            }
            fs::File::create(&target)?.sync_all()?;
            0
        } else {
            copy_synced(&root.join(relative), &partial.join(relative))?
        };
        total += bytes;
        listed.push(json!({ "path": relative.to_string_lossy().replace('\\', "/"), "bytes": bytes }));
    }
    let manifest = json!({
        "format": BACKUP_FORMAT,
        "createdAt": chrono::Utc::now().timestamp_millis(),
        "engine": env!("CARGO_PKG_VERSION"),
        "metadataFile": layout.metadata_file(),
        "files": listed,
    });
    let mut text = serde_json::to_string_pretty(&manifest).map_err(|e| DatabaseError::Catalog(e.to_string()))?;
    text.push('\n');
    write_atomic(&partial.join(BACKUP_MANIFEST), text.as_bytes(), None)?;
    Ok(BackupResult {
        dest: partial.to_path_buf(),
        files: files.len(),
        bytes: total,
    })
}

pub(super) struct ManifestEntry {
    pub(super) relative: PathBuf,
    pub(super) bytes: u64,
}

fn read_manifest(backup: &Path) -> Result<(String, Vec<ManifestEntry>), DatabaseError> {
    let path = backup.join(BACKUP_MANIFEST);
    let raw = fs::read_to_string(&path).map_err(|error| {
        refuse(format!(
            "\"{}\" is not a readable backup manifest: {error}",
            path.display()
        ))
    })?;
    parse_manifest(&raw, &path.display().to_string())
}

pub(super) fn parse_manifest(raw: &str, label: &str) -> Result<(String, Vec<ManifestEntry>), DatabaseError> {
    let manifest: Value =
        serde_json::from_str(raw).map_err(|error| refuse(format!("\"{label}\" is not valid JSON: {error}")))?;
    if manifest.get("format").and_then(Value::as_u64) != Some(BACKUP_FORMAT) {
        return Err(refuse(format!("\"{label}\" is not a backup format this engine reads")));
    }
    let metadata_file = manifest
        .get("metadataFile")
        .and_then(Value::as_str)
        .ok_or_else(|| refuse("backup manifest names no \"metadataFile\"".to_string()))?
        .to_string();
    let listed = manifest
        .get("files")
        .and_then(Value::as_array)
        .ok_or_else(|| refuse("backup manifest lists no \"files\"".to_string()))?;
    let mut entries = Vec::new();
    for item in listed {
        let relative = PathBuf::from(
            item.get("path")
                .and_then(Value::as_str)
                .ok_or_else(|| refuse("a backup manifest entry has no \"path\"".to_string()))?,
        );
        check_relative(&relative)?;
        let bytes = item
            .get("bytes")
            .and_then(Value::as_u64)
            .ok_or_else(|| refuse(format!("manifest entry \"{}\" has no \"bytes\"", relative.display())))?;
        entries.push(ManifestEntry { relative, bytes });
    }
    Ok((metadata_file, entries))
}

/// Restores a backup made by [`backup_database`] into `dir`, replacing the database that is there.
///
/// Recoverable: nothing is deleted until the replacement is in place. The backup is copied into
/// `<dir>/.restore-new` and checked (sizes against the manifest, then [`verify`] with no error);
/// the files of the database being replaced are moved into `<dir>/.restore-old`; the new files are
/// moved in; and only then is `.restore-old` removed. If a move fails, the files already moved are
/// put back, and when even that fails both errors are reported. A `.restore-old` left by an
/// interrupted restore is refused, because it may hold the only good copy.
///
/// Only the files the database accounts for are replaced, so another database living in a
/// subfolder of `dir` is untouched.
pub fn restore_database(backup: &Path, dir: &Path, options: &RestoreOptions) -> Result<RestoreResult, DatabaseError> {
    let (metadata_file, entries) = read_manifest(backup)?;
    for entry in &entries {
        let path = backup.join(&entry.relative);
        let meta = fs::symlink_metadata(&path).map_err(|error| {
            refuse(format!(
                "backup file \"{}\" is missing: {error}",
                entry.relative.display()
            ))
        })?;
        if !meta.is_file() || meta.len() != entry.bytes {
            return Err(refuse(format!(
                "backup file \"{}\" does not match its manifest entry ({} bytes expected)",
                entry.relative.display(),
                entry.bytes
            )));
        }
    }
    restore_staged(
        dir,
        &metadata_file,
        &entries,
        &options.lock,
        options.max_image_bytes,
        |staged| {
            let mut total = 0u64;
            for entry in &entries {
                total += copy_synced(&backup.join(&entry.relative), &staged.join(&entry.relative))?;
            }
            Ok(total)
        },
    )
}

/// The recoverable half of a restore, shared with unpacking: `stage` fills the given staging
/// directory with the `entries` (returning their byte total) while the writer lock is held.
pub(super) fn restore_staged(
    dir: &Path,
    metadata_file: &str,
    entries: &[ManifestEntry],
    lock: &LockMode,
    max_image_bytes: Option<u64>,
    stage: impl FnOnce(&Path) -> Result<u64, DatabaseError>,
) -> Result<RestoreResult, DatabaseError> {
    let old = dir.join(RESTORE_OLD);
    if old.exists() {
        return Err(refuse(format!(
            "\"{}\" exists — an earlier restore was interrupted and it may hold the only good copy of the database; move its files back (or delete it) before restoring again",
            old.display()
        )));
    }
    let created = !dir.exists();
    fs::create_dir_all(dir)?;
    let layout = Database::resolve_layout(
        dir,
        &LayoutOptions {
            metadata_file: Some(metadata_file.to_string()),
            ..LayoutOptions::default()
        },
    )?;

    let held = hold(dir, &layout, lock)?;
    let staged = dir.join(RESTORE_NEW);
    let outcome = install(dir, &layout, entries, &staged, &old, max_image_bytes, stage);
    let _ = fs::remove_dir_all(&staged);
    drop(held);
    if created && outcome.is_err() {
        // NOTE: a directory this call made and nothing was installed into must not outlive the refusal.
        let _ = fs::remove_dir_all(layout.runtime());
        let _ = fs::remove_dir(dir);
    }
    outcome
}

fn install(
    root: &Path,
    layout: &Layout,
    entries: &[ManifestEntry],
    staged: &Path,
    old: &Path,
    max_image_bytes: Option<u64>,
    stage: impl FnOnce(&Path) -> Result<u64, DatabaseError>,
) -> Result<RestoreResult, DatabaseError> {
    if staged.exists() {
        fs::remove_dir_all(staged)?;
    }
    let total = stage(staged)?;
    let report = verify(
        staged,
        &VerifyOptions {
            relations: false,
            max_image_bytes,
            ..VerifyOptions::default()
        },
    );
    if let Some(problem) = report.problems.iter().find(|p| p.code == "image-oversize") {
        return Err(refuse(format!(
            "the backup holds an image over the size limit ({}) — nothing was changed",
            problem.message
        )));
    }
    if let Some(problem) = report
        .problems
        .iter()
        .find(|p| p.severity == super::verify::Severity::Error)
    {
        return Err(refuse(format!(
            "the backup does not verify ({}: {}) — nothing was changed",
            problem.code, problem.message
        )));
    }

    // The files of the database now there: what its catalog accounts for (when it can be read), plus
    // whatever the backup is about to overwrite.
    let mut replaced: BTreeSet<PathBuf> = entries.iter().map(|e| e.relative.clone()).collect();
    if let Ok(catalog) = read_catalog_of(layout) {
        if let Ok(current) = owned_files(root, layout, &catalog, true, &[]) {
            replaced.extend(current);
        }
    }
    replaced.retain(|relative| root.join(relative).is_file());

    fs::create_dir_all(old)?;
    let mut moved_out: Vec<PathBuf> = Vec::new();
    let mut moved_in: Vec<PathBuf> = Vec::new();
    let swap = (|| -> io::Result<()> {
        for relative in &replaced {
            move_file(&root.join(relative), &old.join(relative))?;
            moved_out.push(relative.clone());
        }
        for entry in entries {
            move_file(&staged.join(&entry.relative), &root.join(&entry.relative))?;
            moved_in.push(entry.relative.clone());
        }
        Ok(())
    })();
    match swap {
        Ok(()) => {
            fs::remove_dir_all(old)?;
            Ok(RestoreResult {
                dir: root.to_path_buf(),
                files: entries.len(),
                bytes: total,
                replaced: replaced.len(),
            })
        }
        Err(error) => {
            let rollback = (|| -> io::Result<()> {
                for relative in &moved_in {
                    fs::remove_file(root.join(relative))?;
                }
                for relative in &moved_out {
                    move_file(&old.join(relative), &root.join(relative))?;
                }
                Ok(())
            })();
            match rollback {
                Ok(()) => {
                    let _ = fs::remove_dir_all(old);
                    Err(DatabaseError::Io(error))
                }
                Err(rollback_error) => Err(refuse(format!(
                    "restore failed ({error}) and putting the previous files back failed too ({rollback_error}) — the previous database is in \"{}\"",
                    old.display()
                ))),
            }
        }
    }
}

fn move_file(from: &Path, to: &Path) -> io::Result<()> {
    if let Some(parent) = to.parent() {
        fs::create_dir_all(parent)?;
    }
    fs::rename(from, to)
}