use anyhow::{bail, ensure, Context, Result};
use rusqlite::{
backup::Backup,
types::{Type, ValueRef},
Connection,
};
use std::{
fs::{self, OpenOptions},
io::{ErrorKind, Read},
path::{Path, PathBuf},
process,
sync::atomic::{AtomicU64, Ordering},
time::{Duration, SystemTime, UNIX_EPOCH},
};
use super::run::run_post_migration_hook;
use super::state::{applied_versions, has_migration_table};
use super::transition::backfill_to_baseline;
use super::types::{DryRunResult, Migration, MIGRATIONS, OLD_BASELINE_VERSION};
pub(crate) fn dry_run_pending(real_conn: &Connection) -> Result<DryRunResult> {
let raw_current_version: i64 = real_conn
.query_row("PRAGMA user_version", [], |row| row.get(0))
.unwrap_or(0);
let applied = infer_applied_versions(real_conn, raw_current_version)?;
let current_version = logical_current_version(raw_current_version, &applied);
let migration_version = applied_migration_version(&applied);
let invariant_errors = super::validate_schema_invariants(real_conn)?;
if !invariant_errors.is_empty() {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: applied_pending_count(&applied),
error: Some(format!("schema drift: {}", invariant_errors.join("; "))),
});
}
let temp_path = match DryRunTempPath::create() {
Ok(temp_path) => temp_path,
Err(error) => {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: applied_pending_count(&applied),
error: Some(format!("database clone: {}", error)),
});
}
};
let mut test_conn = match Connection::open(temp_path.path()) {
Ok(conn) => conn,
Err(error) => {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: applied_pending_count(&applied),
error: Some(format!("database clone: {}", error)),
});
}
};
let clone_key = match clone_cipher_key_for_source(real_conn) {
Ok(key) => key,
Err(error) => {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: applied_pending_count(&applied),
error: Some(format!("database clone: {}", error)),
});
}
};
if let Some(key) = clone_key {
if let Err(error) = crate::db::configure_cipher(&test_conn, Some(&key)) {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: applied_pending_count(&applied),
error: Some(format!("database clone: {}", error)),
});
}
}
if let Err(error) = clone_database(real_conn, &mut test_conn) {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: applied_pending_count(&applied),
error: Some(format!("database clone: {}", error)),
});
}
if raw_current_version >= OLD_BASELINE_VERSION || has_migration_table(real_conn) {
if let Err(error) = backfill_to_baseline(&test_conn) {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: applied_pending_count(&applied),
error: Some(format!("baseline backfill: {}", error)),
});
}
}
let pending: Vec<&Migration> = MIGRATIONS
.iter()
.filter(|migration| !applied.contains(&migration.version))
.collect();
if pending.is_empty() {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: 0,
error: None,
});
}
for migration in &pending {
if let Err(error) = test_conn.execute_batch(migration.sql) {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: pending.len(),
error: Some(format!(
"v{:03}_{}: {}",
migration.version, migration.name, error
)),
});
}
if let Err(error) = run_post_migration_hook(&test_conn, migration.version, migration.name) {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: pending.len(),
error: Some(format!(
"v{:03}_{} post-migration hook: {}",
migration.version, migration.name, error
)),
});
}
}
if let Err(error) = backfill_to_baseline(&test_conn) {
return Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: pending.len(),
error: Some(format!("baseline backfill: {}", error)),
});
}
Ok(DryRunResult {
migration_version,
sqlite_user_version: raw_current_version,
current_version,
pending_count: pending.len(),
error: None,
})
}
fn applied_pending_count(applied: &[i64]) -> usize {
MIGRATIONS
.iter()
.filter(|migration| !applied.contains(&migration.version))
.count()
}
fn applied_migration_version(applied: &[i64]) -> i64 {
applied.iter().copied().max().unwrap_or(0)
}
fn infer_applied_versions(conn: &Connection, current_version: i64) -> Result<Vec<i64>> {
if has_migration_table(conn) {
return applied_versions(conn);
}
if current_version >= OLD_BASELINE_VERSION {
return Ok(vec![1]);
}
Ok(Vec::new())
}
fn logical_current_version(raw_current_version: i64, applied: &[i64]) -> i64 {
let Some(latest_applied) = applied.iter().max() else {
return raw_current_version;
};
raw_current_version.max(OLD_BASELINE_VERSION - 1 + latest_applied)
}
fn clone_database(src: &Connection, dst: &mut Connection) -> Result<()> {
let page_size = query_page_size(src)?;
ensure!(page_size > 0, "source database page_size must be positive");
dst.execute_batch(&format!("PRAGMA page_size = {page_size}"))?;
let backup = Backup::new(src, dst)?;
backup.run_to_completion(100, Duration::from_millis(1), None)?;
Ok(())
}
fn clone_cipher_key_for_source(src: &Connection) -> Result<Option<crate::db::CipherKey>> {
if source_database_looks_encrypted(src)? {
return crate::db::load_cipher_key();
}
Ok(None)
}
fn source_database_looks_encrypted(conn: &Connection) -> Result<bool> {
let mut stmt = conn.prepare("PRAGMA database_list")?;
let rows = stmt.query_map([], |row| {
Ok((row.get::<_, String>(1)?, row.get::<_, String>(2)?))
})?;
for row in rows {
let (name, file) = row?;
if name != "main" {
continue;
}
let file = file.trim();
if file.is_empty() {
return Ok(false);
}
return sqlite_file_looks_encrypted(Path::new(file));
}
Ok(false)
}
fn sqlite_file_looks_encrypted(path: &Path) -> Result<bool> {
let mut file = match fs::File::open(path) {
Ok(file) => file,
Err(error) if error.kind() == ErrorKind::NotFound => return Ok(false),
Err(error) => {
return Err(error)
.with_context(|| format!("inspect source database {}", path.display()));
}
};
let mut header = [0_u8; 16];
let read = file
.read(&mut header)
.with_context(|| format!("read source database header {}", path.display()))?;
if read < header.len() {
return Ok(false);
}
Ok(&header != b"SQLite format 3\0")
}
fn query_page_size(conn: &Connection) -> Result<i64> {
conn.query_row("PRAGMA page_size", [], |row| match row.get_ref(0)? {
ValueRef::Integer(value) => Ok(value),
ValueRef::Text(bytes) => {
let text = std::str::from_utf8(bytes).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(0, Type::Text, Box::new(error))
})?;
text.parse::<i64>().map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(0, Type::Text, Box::new(error))
})
}
other => Err(rusqlite::Error::InvalidColumnType(
0,
"page_size".to_string(),
other.data_type(),
)),
})
.map_err(Into::into)
}
struct DryRunTempPath {
path: PathBuf,
}
impl DryRunTempPath {
fn create() -> Result<Self> {
static COUNTER: AtomicU64 = AtomicU64::new(0);
let temp_dir = std::env::temp_dir();
for _ in 0..32 {
let counter = COUNTER.fetch_add(1, Ordering::Relaxed);
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.context("system time before unix epoch")?
.as_nanos();
let path = temp_dir.join(format!("remem-dry-run-{}-{nonce}-{counter}", process::id()));
let mut options = OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600);
}
match options.open(&path) {
Ok(_) => return Ok(Self { path }),
Err(error) if error.kind() == ErrorKind::AlreadyExists => continue,
Err(error) => {
return Err(error)
.with_context(|| format!("create dry-run database {}", path.display()));
}
}
}
bail!("create unique dry-run database path")
}
fn path(&self) -> &Path {
&self.path
}
}
impl Drop for DryRunTempPath {
fn drop(&mut self) {
cleanup_sqlite_files(&self.path);
}
}
fn cleanup_sqlite_files(path: &Path) {
remove_sqlite_file(path);
remove_sqlite_file(&sqlite_sidecar_path(path, "-wal"));
remove_sqlite_file(&sqlite_sidecar_path(path, "-shm"));
remove_sqlite_file(&sqlite_sidecar_path(path, "-journal"));
}
fn remove_sqlite_file(path: &Path) {
match fs::remove_file(path) {
Ok(()) => {}
Err(error) if error.kind() == ErrorKind::NotFound => {}
Err(error) => eprintln!(
"failed to remove dry-run database file {}: {}",
path.display(),
error
),
}
}
fn sqlite_sidecar_path(path: &Path, suffix: &str) -> PathBuf {
let mut sidecar = path.as_os_str().to_os_string();
sidecar.push(suffix);
PathBuf::from(sidecar)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::db::test_support::{cleanup_temp_db_files, unique_temp_db_path};
#[cfg(unix)]
#[test]
fn dry_run_temp_path_uses_owner_only_permissions() -> Result<()> {
use std::os::unix::fs::PermissionsExt;
let temp_path = DryRunTempPath::create()?;
let mode = fs::metadata(temp_path.path())?.permissions().mode() & 0o777;
assert_eq!(mode, 0o600);
Ok(())
}
#[test]
fn dry_run_temp_path_uses_on_disk_database_and_cleans_up() -> Result<()> {
let source_path = unique_temp_db_path("dry-run-source");
let dry_run_path;
{
let source = Connection::open(&source_path)?;
source.execute_batch(
"PRAGMA page_size = 8192;
CREATE TABLE items (id INTEGER PRIMARY KEY);
INSERT INTO items DEFAULT VALUES;",
)?;
let temp_path = DryRunTempPath::create()?;
dry_run_path = temp_path.path().to_path_buf();
{
let mut dst = Connection::open(temp_path.path())?;
clone_database(&source, &mut dst)?;
let count: i64 =
dst.query_row("SELECT COUNT(*) FROM items", [], |row| row.get(0))?;
assert_eq!(count, 1);
let page_size: i64 = dst.query_row("PRAGMA page_size", [], |row| row.get(0))?;
assert_eq!(page_size, 8192);
}
assert!(
dry_run_path.exists(),
"dry-run clone should use a temporary on-disk database file"
);
}
assert!(
!dry_run_path.exists(),
"temporary dry-run database should be removed after use"
);
assert!(!sqlite_sidecar_path(&dry_run_path, "-wal").exists());
assert!(!sqlite_sidecar_path(&dry_run_path, "-shm").exists());
cleanup_temp_db_files(&source_path);
Ok(())
}
#[test]
fn plaintext_source_database_does_not_request_cipher_clone() -> Result<()> {
let source_path = unique_temp_db_path("dry-run-plaintext-source");
{
let source = Connection::open(&source_path)?;
source.execute_batch(
"CREATE TABLE items (id INTEGER PRIMARY KEY);
INSERT INTO items DEFAULT VALUES;",
)?;
assert!(!source_database_looks_encrypted(&source)?);
assert_eq!(clone_cipher_key_for_source(&source)?, None);
}
cleanup_temp_db_files(&source_path);
Ok(())
}
}