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;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SqliteBackupReport {
pub schema_version: u32,
pub bytes: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SqlitePairedBackupReport {
pub local: SqliteBackupReport,
pub queue: SqliteBackupReport,
pub queue_entries: u64,
pub total_bytes: u64,
}
impl SqliteSyncStore {
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
}
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)?;
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
}
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)
})
}
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()));
let mut builder = fs::DirBuilder::new();
#[cfg(unix)]
{
use std::os::unix::fs::DirBuilderExt;
builder.mode(0o700);
}
match builder.create(&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)
}