appcore_sync_sqlite/

backup.rs

// =============================================================================
//        #######
//     ###       ###     F: backup.rs
//    ##   ## ##   ##    P: AppCore-Runtime
//         ## ##
//                       C: 2026/08/26 00:00:00 by dnettoRaw
//    ##   ## ##   ##    U: 2026/08/26 00:00:00 by dnettoRaw
//      ###########      S: 2.0.0
// =============================================================================

//! Defines bounded backup contracts and behavior for this crate.

use crate::store::{configure_memory, normalize_path};
use crate::{
    SqliteSyncError, SqliteSyncResult, SqliteSyncStore, SQLITE_SYNC_SCHEMA_V1,
    SQLITE_SYNC_SCHEMA_V2, SQLITE_SYNC_SCHEMA_V3, SQLITE_SYNC_SCHEMA_V4, SQLITE_SYNC_SCHEMA_V5,
};
use rusqlite::backup::Backup;
use rusqlite::{Connection, OpenFlags, TransactionBehavior};
use sha2::{Digest, Sha256};
use std::fs::{self, File, OpenOptions};
use std::io::Read;
use std::path::{Path, PathBuf};
use std::time::Duration;

/// Evidence returned after an integrity-checked online backup.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SqliteBackupReport {
    /// Internal schema version copied to the backup.
    pub schema_version: u32,
    /// Final backup file size in bytes.
    pub bytes: u64,
}

/// Evidence returned after an integrity-checked backup of both attached databases.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SqlitePairedBackupReport {
    /// Backup evidence for complete local state.
    pub local: SqliteBackupReport,
    /// Backup evidence for the pending-only sync queue.
    pub queue: SqliteBackupReport,
    /// Number of unconfirmed operations captured in the same snapshot.
    pub queue_entries: u64,
    /// Combined size of the two SQLite files.
    pub total_bytes: u64,
}

impl SqliteSyncStore {
    /// Creates an integrity-checked online backup and atomically publishes it.
    pub fn online_backup(
        &self,
        destination: impl AsRef<Path>,
    ) -> SqliteSyncResult<SqliteBackupReport> {
        if self.config().separate_outbox_path().is_some() {
            return Err(SqliteSyncError::InvalidConfiguration(
                "separate outbox requires a paired backup",
            ));
        }
        let destination = normalize_backup_destination(destination.as_ref())?;
        let temporary = reserve_temporary(&destination)?;
        let result = self.copy_backup(&temporary).and_then(|report| {
            sync_file(&temporary)?;
            publish_temporary(&temporary, &destination)?;
            Ok(report)
        });
        if result.is_err() {
            let _ = fs::remove_file(&temporary);
        }
        result
    }

    /// Publishes a consistent bundle containing the local provider database and the
    /// pending-only queue database. The destination directory must not exist.
    pub fn paired_online_backup(
        &self,
        destination: impl AsRef<Path>,
    ) -> SqliteSyncResult<SqlitePairedBackupReport> {
        if self.config().separate_outbox_path().is_none() {
            return Err(SqliteSyncError::InvalidConfiguration(
                "paired backup requires a separate outbox",
            ));
        }
        let destination = normalize_backup_destination(destination.as_ref())?;
        let temporary = reserve_temporary_directory(&destination)?;
        let local_name = safe_manifest_name(
            self.config()
                .path
                .file_name()
                .and_then(|name| name.to_str()),
        )?;
        let queue_name = safe_manifest_name(
            self.config()
                .separate_outbox_path()
                .and_then(Path::file_name)
                .and_then(|name| name.to_str()),
        )?;
        let result = self.with_connection(|connection| {
            let transaction = connection
                .transaction_with_behavior(TransactionBehavior::Exclusive)
                .map_err(SqliteSyncError::database)?;
            // Establish a read snapshot on both attached files before either copy.
            let _: i64 = transaction
                .query_row("SELECT count(*) FROM main.appcore_object_head", [], |row| row.get(0))
                .map_err(SqliteSyncError::database)?;
            let queue_entries: u64 = transaction
                .query_row("SELECT count(*) FROM sync_queue.appcore_sync_outbox", [], |row| row.get(0))
                .map_err(SqliteSyncError::database)?;
            let local_path = temporary.join("local.sqlite3");
            let queue_path = temporary.join("sync-queue.sqlite3");
            fs::copy(self.config().path(), &local_path)
                .map_err(|_| SqliteSyncError::DatabaseOperation)?;
            fs::copy(
                self.config().separate_outbox_path().ok_or(
                    SqliteSyncError::InvalidConfiguration("separate outbox required"),
                )?,
                &queue_path,
            )
            .map_err(|_| SqliteSyncError::DatabaseOperation)?;
            let local = validate_pair_database(&local_path, false)?;
            let queue = validate_pair_database(&queue_path, true)?;
            drop(transaction);
            let manifest = format!(
                "appcore-sync-pair-v1\nlocal_file={local_name}\nqueue_file={queue_name}\nlocal_schema={}\nlocal_bytes={}\nlocal_sha256={}\nqueue_schema={}\nqueue_bytes={}\nqueue_sha256={}\nqueue_entries={}\n",
                local.schema_version, local.bytes, file_sha256(&local_path)?, queue.schema_version, queue.bytes, file_sha256(&queue_path)?, queue_entries
            );
            let manifest_path = temporary.join("manifest.txt");
            fs::write(&manifest_path, manifest).map_err(|_| SqliteSyncError::DatabaseOperation)?;
            sync_file(&local_path)?;
            sync_file(&queue_path)?;
            sync_file(&manifest_path)?;
            sync_parent(&local_path)?;
            let total_bytes = local.bytes.saturating_add(queue.bytes);
            Ok(SqlitePairedBackupReport { local, queue, queue_entries, total_bytes })
        });
        let result = result.and_then(|report| {
            publish_directory(&temporary, &destination)?;
            Ok(report)
        });
        if result.is_err() {
            let _ = fs::remove_dir_all(&temporary);
        }
        result
    }

    /// Restores a validated paired bundle into a new directory and publishes it atomically.
    pub fn restore_paired_backup_to_new(
        bundle: impl AsRef<Path>,
        destination_directory: impl AsRef<Path>,
    ) -> SqliteSyncResult<SqlitePairedBackupReport> {
        let bundle = normalize_bundle_directory(bundle.as_ref())?;
        let report = validate_paired_bundle(&bundle)?;
        let destination = normalize_backup_destination(destination_directory.as_ref())?;
        let temporary = reserve_temporary_directory(&destination)?;
        let result = (|| {
            let manifest = fs::read_to_string(bundle.join("manifest.txt"))
                .map_err(|_| SqliteSyncError::IntegrityFailed)?;
            let (local_name, queue_name) = manifest_names(&manifest)?;
            for (source, name) in [
                ("local.sqlite3", local_name),
                ("sync-queue.sqlite3", queue_name),
            ] {
                fs::copy(bundle.join(source), temporary.join(&name))
                    .map_err(|_| SqliteSyncError::DatabaseOperation)?;
                sync_file(&temporary.join(name))?;
            }
            fs::copy(bundle.join("manifest.txt"), temporary.join("manifest.txt"))
                .map_err(|_| SqliteSyncError::DatabaseOperation)?;
            sync_file(&temporary.join("manifest.txt"))?;
            validate_paired_bundle(&temporary)?;
            sync_parent(&temporary.join("local.sqlite3"))?;
            publish_directory(&temporary, &destination)?;
            Ok(report)
        })();
        if result.is_err() {
            let _ = fs::remove_dir_all(&temporary);
        }
        result
    }

    fn copy_backup(&self, temporary: &Path) -> SqliteSyncResult<SqliteBackupReport> {
        self.with_connection(|source| {
            copy_connection(source, temporary, self.config().backup_pages_per_step)
        })
    }

    /// Restores a verified backup into a new, previously absent database path.
    pub fn restore_backup_to_new(
        backup_path: impl AsRef<Path>,
        destination: impl AsRef<Path>,
    ) -> SqliteSyncResult<SqliteBackupReport> {
        let original_backup_path = backup_path.as_ref();
        let backup_path = normalize_path(original_backup_path)?;
        if !backup_path.is_file() {
            return Err(SqliteSyncError::UnsafePath);
        }
        let destination = normalize_backup_destination(destination.as_ref())?;
        let source = Connection::open_with_flags(
            &backup_path,
            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_FULL_MUTEX,
        )
        .map_err(SqliteSyncError::database)?;
        configure_memory(&source)?;
        validate_backup_connection(&source)?;
        let temporary = reserve_temporary(&destination)?;
        let result = copy_connection(&source, &temporary, 128).and_then(|report| {
            sync_file(&temporary)?;
            publish_temporary(&temporary, &destination)?;
            Ok(report)
        });
        if result.is_err() {
            let _ = fs::remove_file(&temporary);
        }
        result
    }
}

fn copy_connection(
    source: &Connection,
    temporary: &Path,
    pages_per_step: i32,
) -> SqliteSyncResult<SqliteBackupReport> {
    let flags = OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_FULL_MUTEX;
    let mut destination =
        Connection::open_with_flags(temporary, flags).map_err(SqliteSyncError::database)?;
    configure_memory(&destination)?;
    let backup = Backup::new(source, &mut destination).map_err(SqliteSyncError::database)?;
    backup
        .run_to_completion(pages_per_step, Duration::from_millis(2), None)
        .map_err(SqliteSyncError::database)?;
    drop(backup);
    let schema_version = validate_backup_connection(&destination)?;
    drop(destination);
    let bytes = fs::metadata(temporary)
        .map_err(|_| SqliteSyncError::DatabaseOperation)?
        .len();
    Ok(SqliteBackupReport {
        schema_version,
        bytes,
    })
}

fn reserve_temporary_directory(destination: &Path) -> SqliteSyncResult<PathBuf> {
    let parent = destination.parent().unwrap_or_else(|| Path::new("."));
    let name = destination
        .file_name()
        .and_then(|value| value.to_str())
        .ok_or(SqliteSyncError::UnsafePath)?;
    for attempt in 0..16u8 {
        let candidate = parent.join(format!(".{name}.{}-{attempt}.tmp", std::process::id()));
        match fs::create_dir(&candidate) {
            Ok(()) => return Ok(candidate),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(_) => return Err(SqliteSyncError::DatabaseOperation),
        }
    }
    Err(SqliteSyncError::CapacityExceeded("backup temporary"))
}

fn publish_directory(temporary: &Path, destination: &Path) -> SqliteSyncResult<()> {
    fs::rename(temporary, destination).map_err(|_| SqliteSyncError::DatabaseOperation)?;
    if sync_parent(destination).is_err() {
        let _ = fs::rename(destination, temporary);
        return Err(SqliteSyncError::DatabaseOperation);
    }
    Ok(())
}

fn validate_paired_bundle(bundle: &Path) -> SqliteSyncResult<SqlitePairedBackupReport> {
    let manifest_path = bundle.join("manifest.txt");
    let manifest =
        fs::read_to_string(&manifest_path).map_err(|_| SqliteSyncError::IntegrityFailed)?;
    let (local_name, queue_name) = manifest_names(&manifest)?;
    let local_path = if bundle.join("local.sqlite3").is_file() {
        bundle.join("local.sqlite3")
    } else {
        bundle.join(&local_name)
    };
    let queue_path = if bundle.join("sync-queue.sqlite3").is_file() {
        bundle.join("sync-queue.sqlite3")
    } else {
        bundle.join(&queue_name)
    };
    let local = validate_pair_database(&local_path, false)?;
    let queue = validate_pair_database(&queue_path, true)?;
    let local_connection = Connection::open_with_flags(
        &local_path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_FULL_MUTEX,
    )
    .map_err(SqliteSyncError::database)?;
    let recorded_queue_name: String = local_connection
        .query_row(
            "SELECT outbox_file FROM appcore_sync_storage_mode WHERE singleton=1",
            [],
            |row| row.get(0),
        )
        .map_err(SqliteSyncError::database)?;
    if recorded_queue_name != queue_name {
        return Err(SqliteSyncError::IntegrityFailed);
    }
    let connection = Connection::open_with_flags(
        &queue_path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_FULL_MUTEX,
    )
    .map_err(SqliteSyncError::database)?;
    let queue_entries: u64 = connection
        .query_row("SELECT count(*) FROM appcore_sync_outbox", [], |row| {
            row.get(0)
        })
        .map_err(SqliteSyncError::database)?;
    let expected = format!(
        "appcore-sync-pair-v1\nlocal_file={local_name}\nqueue_file={queue_name}\nlocal_schema={}\nlocal_bytes={}\nlocal_sha256={}\nqueue_schema={}\nqueue_bytes={}\nqueue_sha256={}\nqueue_entries={}\n",
        local.schema_version, local.bytes, file_sha256(&local_path)?, queue.schema_version, queue.bytes, file_sha256(&queue_path)?, queue_entries
    );
    if manifest != expected {
        return Err(SqliteSyncError::IntegrityFailed);
    }
    Ok(SqlitePairedBackupReport {
        total_bytes: local.bytes.saturating_add(queue.bytes),
        local,
        queue,
        queue_entries,
    })
}

fn manifest_names(manifest: &str) -> SqliteSyncResult<(String, String)> {
    let mut lines = manifest.lines();
    if lines.next() != Some("appcore-sync-pair-v1") {
        return Err(SqliteSyncError::UpdateRequired);
    }
    let mut local = None;
    let mut queue = None;
    for line in lines {
        if let Some(value) = line.strip_prefix("local_file=") {
            local = Some(safe_manifest_name(Some(value))?);
        }
        if let Some(value) = line.strip_prefix("queue_file=") {
            queue = Some(safe_manifest_name(Some(value))?);
        }
    }
    Ok((
        local.ok_or(SqliteSyncError::IntegrityFailed)?,
        queue.ok_or(SqliteSyncError::IntegrityFailed)?,
    ))
}

fn safe_manifest_name(value: Option<&str>) -> SqliteSyncResult<String> {
    let value = value.ok_or(SqliteSyncError::UnsafePath)?;
    if value.is_empty()
        || value == "."
        || value == ".."
        || value.contains('/')
        || value.contains('\\')
        || value.bytes().any(|byte| byte.is_ascii_control())
    {
        return Err(SqliteSyncError::UnsafePath);
    }
    Ok(value.to_owned())
}

fn validate_pair_database(path: &Path, queue: bool) -> SqliteSyncResult<SqliteBackupReport> {
    if !path.is_file() {
        return Err(SqliteSyncError::UnsafePath);
    }
    let connection = Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_FULL_MUTEX,
    )
    .map_err(SqliteSyncError::database)?;
    let schema_version: u32 = connection
        .pragma_query_value(None, "user_version", |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    let application_id: i64 = connection
        .pragma_query_value(None, "application_id", |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    let expected_id = if queue { 0x4150_4351 } else { 0x4150_4353 };
    if application_id != expected_id {
        return Err(SqliteSyncError::UpdateRequired);
    }
    if queue {
        let integrity: String = connection
            .query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
            .map_err(SqliteSyncError::database)?;
        if integrity != "ok" {
            return Err(SqliteSyncError::IntegrityFailed);
        }
    } else if validate_backup_connection(&connection).is_err()
        && !validate_local_backup_connection(&connection)?
    {
        return Err(SqliteSyncError::IntegrityFailed);
    }
    if queue {
        let has_queue: bool = connection
            .query_row("SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type='table' AND name='appcore_sync_outbox')", [], |row| row.get(0))
            .map_err(SqliteSyncError::database)?;
        if !has_queue || schema_version != 1 {
            return Err(SqliteSyncError::UpdateRequired);
        }
    }
    let bytes = fs::metadata(path)
        .map_err(|_| SqliteSyncError::DatabaseOperation)?
        .len();
    Ok(SqliteBackupReport {
        schema_version,
        bytes,
    })
}

fn validate_local_backup_connection(connection: &Connection) -> SqliteSyncResult<bool> {
    let version: u32 = connection
        .pragma_query_value(None, "user_version", |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    if !(SQLITE_SYNC_SCHEMA_V1..=SQLITE_SYNC_SCHEMA_V5).contains(&version) {
        return Ok(false);
    }
    let integrity: String = connection
        .query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    let mode: String = connection
        .query_row(
            "SELECT outbox_mode FROM appcore_sync_storage_mode WHERE singleton=1",
            [],
            |row| row.get(0),
        )
        .map_err(SqliteSyncError::database)?;
    Ok(integrity == "ok" && mode == "separate")
}

fn file_sha256(path: &Path) -> SqliteSyncResult<String> {
    let mut file = File::open(path).map_err(|_| SqliteSyncError::DatabaseOperation)?;
    let mut hasher = Sha256::new();
    let mut buffer = [0u8; 64 * 1024];
    loop {
        let read = file
            .read(&mut buffer)
            .map_err(|_| SqliteSyncError::DatabaseOperation)?;
        if read == 0 {
            break;
        }
        hasher.update(&buffer[..read]);
    }
    Ok(hasher
        .finalize()
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect())
}

fn normalize_backup_destination(path: &Path) -> SqliteSyncResult<PathBuf> {
    if path.as_os_str().is_empty() {
        return Err(SqliteSyncError::UnsafePath);
    }
    let normalized = normalize_path(path)?;
    if normalized.exists() {
        return Err(SqliteSyncError::UnsafePath);
    }
    Ok(normalized)
}

fn normalize_bundle_directory(path: &Path) -> SqliteSyncResult<PathBuf> {
    let metadata = fs::symlink_metadata(path).map_err(|_| SqliteSyncError::UnsafePath)?;
    if !metadata.is_dir() || metadata.file_type().is_symlink() {
        return Err(SqliteSyncError::UnsafePath);
    }
    fs::canonicalize(path).map_err(|_| SqliteSyncError::UnsafePath)
}

fn reserve_temporary(destination: &Path) -> SqliteSyncResult<PathBuf> {
    let parent = destination.parent().unwrap_or_else(|| Path::new("."));
    let name = destination
        .file_name()
        .and_then(|value| value.to_str())
        .ok_or(SqliteSyncError::UnsafePath)?;
    for attempt in 0..16u8 {
        let candidate = parent.join(format!(".{name}.{}-{attempt}.tmp", std::process::id()));
        match OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&candidate)
        {
            Ok(file) => {
                drop(file);
                return Ok(candidate);
            }
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(_) => return Err(SqliteSyncError::DatabaseOperation),
        }
    }
    Err(SqliteSyncError::CapacityExceeded("backup temporary"))
}

fn publish_temporary(temporary: &Path, destination: &Path) -> SqliteSyncResult<()> {
    fs::hard_link(temporary, destination).map_err(|_| SqliteSyncError::DatabaseOperation)?;
    if fs::remove_file(temporary).is_err() {
        let _ = fs::remove_file(destination);
        return Err(SqliteSyncError::DatabaseOperation);
    }
    if sync_parent(destination).is_err() {
        let _ = fs::remove_file(destination);
        let _ = sync_parent(destination);
        return Err(SqliteSyncError::DatabaseOperation);
    }
    Ok(())
}

fn validate_backup_connection(connection: &Connection) -> SqliteSyncResult<u32> {
    let version: u32 = connection
        .pragma_query_value(None, "user_version", |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    if version >= SQLITE_SYNC_SCHEMA_V5 {
        let mode: String = connection
            .query_row(
                "SELECT outbox_mode FROM appcore_sync_storage_mode WHERE singleton=1",
                [],
                |row| row.get(0),
            )
            .map_err(SqliteSyncError::database)?;
        if mode != "local" {
            return Err(SqliteSyncError::UpdateRequired);
        }
    }
    let application_id: i64 = connection
        .pragma_query_value(None, "application_id", |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    if application_id != 0x4150_4353 {
        return Err(SqliteSyncError::UpdateRequired);
    }
    let integrity: String = connection
        .query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    if matches!(
        version,
        SQLITE_SYNC_SCHEMA_V1
            | SQLITE_SYNC_SCHEMA_V2
            | SQLITE_SYNC_SCHEMA_V3
            | SQLITE_SYNC_SCHEMA_V4
            | SQLITE_SYNC_SCHEMA_V5
    ) && integrity == "ok"
    {
        Ok(version)
    } else if !matches!(
        version,
        SQLITE_SYNC_SCHEMA_V1
            | SQLITE_SYNC_SCHEMA_V2
            | SQLITE_SYNC_SCHEMA_V3
            | SQLITE_SYNC_SCHEMA_V4
            | SQLITE_SYNC_SCHEMA_V5
    ) {
        Err(SqliteSyncError::UpdateRequired)
    } else {
        Err(SqliteSyncError::IntegrityFailed)
    }
}

fn sync_file(path: &Path) -> SqliteSyncResult<()> {
    let mut options = OpenOptions::new();
    options.read(true);
    #[cfg(windows)]
    options.write(true);
    options
        .open(path)
        .and_then(|file| file.sync_all())
        .map_err(|_| SqliteSyncError::DatabaseOperation)
}

#[cfg(unix)]
fn sync_parent(path: &Path) -> SqliteSyncResult<()> {
    let parent = path.parent().unwrap_or_else(|| Path::new("."));
    File::open(parent)
        .and_then(|directory| directory.sync_all())
        .map_err(|_| SqliteSyncError::DatabaseOperation)
}

#[cfg(windows)]
fn sync_parent(path: &Path) -> SqliteSyncResult<()> {
    use std::os::windows::fs::OpenOptionsExt;
    let parent = path.parent().ok_or(SqliteSyncError::UnsafePath)?;
    OpenOptions::new()
        .read(true)
        .write(true)
        .custom_flags(windows_sys::Win32::Storage::FileSystem::FILE_FLAG_BACKUP_SEMANTICS)
        .open(parent)
        .and_then(|directory| directory.sync_all())
        .map_err(|_| SqliteSyncError::DatabaseOperation)
}
#[cfg(not(any(unix, windows)))]
fn sync_parent(_path: &Path) -> SqliteSyncResult<()> {
    Err(SqliteSyncError::DatabaseOperation)
}