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};
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 {
pub lock: LockMode,
pub indexes: bool,
}
impl Default for BackupOptions {
fn default() -> Self {
BackupOptions {
lock: LockMode::default(),
indexes: false,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BackupResult {
pub dest: PathBuf,
pub files: usize,
pub bytes: u64,
}
#[derive(Debug, Clone, Default)]
pub struct RestoreOptions {
pub lock: LockMode,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RestoreResult {
pub dir: PathBuf,
pub files: usize,
pub bytes: u64,
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())
}
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)
}
pub(super) fn owned_files(
root: &Path,
layout: &Layout,
catalog: &Catalog,
indexes: bool,
) -> Result<Vec<PathBuf>, DatabaseError> {
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 {
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())
}
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)
}
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 files = owned_files(root, layout, catalog, options.indexes)?;
fs::create_dir_all(partial)?;
let mut listed = Vec::new();
let mut total = 0u64;
for relative in &files {
let bytes = 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))
}
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, |staged| {
let mut total = 0u64;
for entry in &entries {
total += copy_synced(&backup.join(&entry.relative), &staged.join(&entry.relative))?;
}
Ok(total)
})
}
pub(super) fn restore_staged(
dir: &Path,
metadata_file: &str,
entries: &[ManifestEntry],
lock: &LockMode,
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()
)));
}
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, stage);
let _ = fs::remove_dir_all(&staged);
drop(held);
outcome
}
fn install(
root: &Path,
layout: &Layout,
entries: &[ManifestEntry],
staged: &Path,
old: &Path,
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,
..VerifyOptions::default()
},
);
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
)));
}
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)
}