use std::fs;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use web_time::{SystemTime, UNIX_EPOCH};
use lora_wal::WalError;
use super::format::{extract_archive, ExtractedArchive};
use super::format::{read_archive_snapshot, ContainerSnapshot};
use super::platform::sync_dir;
static ARCHIVE_TMP_COUNTER: AtomicU64 = AtomicU64::new(0);
pub(super) fn make_work_dir(archive_path: &Path) -> PathBuf {
archive_path.with_extension("loradb.wal")
}
pub(super) fn prepare_work_dir(
archive_path: &Path,
work_dir: &Path,
max_archive_bytes: u64,
) -> Result<Option<ContainerSnapshot>, WalError> {
if has_wal_files(work_dir)? {
return if archive_path.exists() {
Ok(read_archive_snapshot(archive_path).ok().flatten())
} else {
Ok(None)
};
}
if work_dir.exists() {
fs::remove_dir_all(work_dir)?;
}
if archive_path.exists() {
let existing_len = fs::metadata(archive_path)?.len();
if existing_len > max_archive_bytes {
return Err(WalError::Malformed(format!(
"database container {} is {} bytes, above configured limit {}",
archive_path.display(),
existing_len,
max_archive_bytes
)));
}
extract_archive_into_work_dir(archive_path, work_dir)
} else {
fs::create_dir_all(work_dir)?;
Ok(None)
}
}
fn extract_archive_into_work_dir(
archive_path: &Path,
work_dir: &Path,
) -> Result<Option<ContainerSnapshot>, WalError> {
let tmp_dir = make_extract_tmp_path(work_dir);
let result = (|| {
fs::create_dir_all(&tmp_dir)?;
let ExtractedArchive { snapshot, saw_wal } = extract_archive(archive_path, &tmp_dir)?;
sync_dir(&tmp_dir)?;
if saw_wal {
fs::rename(&tmp_dir, work_dir)?;
} else {
fs::remove_dir_all(&tmp_dir)?;
fs::create_dir_all(work_dir)?;
}
if let Some(parent) = work_dir.parent() {
sync_dir(parent)?;
}
Ok(snapshot)
})();
if result.is_err() {
let _ = fs::remove_dir_all(&tmp_dir);
let _ = fs::remove_dir_all(work_dir);
}
result
}
pub(super) fn make_archive_tmp_path(archive_path: &Path) -> PathBuf {
let archive_name = archive_path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("database.loradb");
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let sequence = ARCHIVE_TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
archive_path.with_file_name(format!(
"{}.{}.{}.{}.tmp",
sanitize_for_temp(archive_name),
std::process::id(),
nanos,
sequence
))
}
fn make_extract_tmp_path(work_dir: &Path) -> PathBuf {
let dir_name = work_dir
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("database.loradb.wal");
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let sequence = ARCHIVE_TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
work_dir.with_file_name(format!(
"{}.extract.{}.{}.{}",
sanitize_for_temp(dir_name),
std::process::id(),
nanos,
sequence
))
}
pub(super) fn cleanup_stale_temp_paths(archive_path: &Path) -> Result<(), WalError> {
let parent = archive_path.parent().unwrap_or_else(|| Path::new("."));
if !parent.exists() {
return Ok(());
}
let archive_name = archive_path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("database.loradb");
let archive_tmp_prefix = format!("{}.", sanitize_for_temp(archive_name));
let extract_tmp_prefix = format!(
"{}.extract.",
sanitize_for_temp(
make_work_dir(archive_path)
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("database.loradb.wal")
)
);
for entry in fs::read_dir(parent)? {
let entry = entry?;
let file_name = entry.file_name();
let Some(file_name) = file_name.to_str() else {
continue;
};
let is_archive_tmp = is_generated_archive_tmp_name(file_name, &archive_tmp_prefix);
let is_extract_tmp = is_generated_extract_tmp_name(file_name, &extract_tmp_prefix);
if !is_archive_tmp && !is_extract_tmp {
continue;
}
let path = entry.path();
if path.is_dir() {
fs::remove_dir_all(path)?;
} else {
fs::remove_file(path)?;
}
}
Ok(())
}
fn sanitize_for_temp(value: &str) -> String {
value
.bytes()
.map(|b| {
if b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'.') {
b as char
} else {
'_'
}
})
.collect()
}
fn is_generated_archive_tmp_name(file_name: &str, prefix: &str) -> bool {
let Some(rest) = file_name.strip_prefix(prefix) else {
return false;
};
let Some(rest) = rest.strip_suffix(".tmp") else {
return false;
};
let mut parts = rest.split('.');
matches!(
(parts.next(), parts.next(), parts.next(), parts.next()),
(Some(pid), Some(nanos), Some(sequence), None)
if is_ascii_digits(pid) && is_ascii_digits(nanos) && is_ascii_digits(sequence)
)
}
fn is_generated_extract_tmp_name(file_name: &str, prefix: &str) -> bool {
let Some(rest) = file_name.strip_prefix(prefix) else {
return false;
};
let mut parts = rest.split('.');
matches!(
(parts.next(), parts.next(), parts.next(), parts.next()),
(Some(pid), Some(nanos), Some(sequence), None)
if is_ascii_digits(pid) && is_ascii_digits(nanos) && is_ascii_digits(sequence)
)
}
fn is_ascii_digits(value: &str) -> bool {
!value.is_empty() && value.bytes().all(|b| b.is_ascii_digit())
}
pub(super) fn sorted_wal_files(wal_dir: &Path) -> Result<Vec<PathBuf>, WalError> {
let mut entries = Vec::new();
for entry in fs::read_dir(wal_dir)? {
let path = entry?.path();
if path.extension().and_then(|s| s.to_str()) == Some("wal") {
entries.push(path);
}
}
entries.sort();
Ok(entries)
}
fn has_wal_files(wal_dir: &Path) -> Result<bool, WalError> {
if !wal_dir.exists() {
return Ok(false);
}
Ok(sorted_wal_files(wal_dir)?.into_iter().next().is_some())
}