1use crate::{apply_key_first, DbError, DbOptions, Result};
11use rusqlite::backup::{Backup, StepResult};
12use rusqlite::Connection;
13use std::path::{Path, PathBuf};
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::time::{Duration, Instant};
16
17static COUNTER: AtomicU64 = AtomicU64::new(0);
18
19pub trait BackupProgress {
20 fn step(&mut self, remaining: i32) -> bool;
22}
23
24pub fn online_backup_to(
27 options: &DbOptions,
28 source: &Connection,
29 target: &Path,
30 mut progress: Option<&mut dyn BackupProgress>,
31) -> Result<()> {
32 let key = options.key_pragma_value()?;
33 let temp = temp_path(target)?;
34 let result = (|| -> Result<()> {
35 let mut dst = Connection::open(&temp)?;
36 dst.busy_timeout(options.busy_timeout)?;
37 apply_key_first(&dst, key.as_ref().map(|k| k.as_str()))?;
38 {
39 let backup = Backup::new(source, &mut dst)?;
40 loop {
41 let step = backup.step(512)?;
42 if let Some(sink) = progress.as_deref_mut() {
43 if !sink.step(backup.progress().remaining) {
44 return Err(DbError::Backup("cancelled".into()));
45 }
46 }
47 match step {
48 StepResult::Done => break,
49 StepResult::More => {}
50 _ => std::thread::sleep(Duration::from_millis(25)),
52 }
53 }
54 }
55 let _ = dst.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(()));
57 dst.query_row("PRAGMA journal_mode=DELETE", [], |row| {
58 row.get::<_, String>(0)
59 })?;
60 let report: String = dst.query_row("PRAGMA quick_check", [], |row| row.get(0))?;
61 if !report.trim().eq_ignore_ascii_case("ok") {
62 return Err(DbError::Backup(format!(
63 "quick_check on the backup said: {report}"
64 )));
65 }
66 drop(dst);
67 rightkit_fs::restrict_owner_only(&temp)?;
68 std::fs::OpenOptions::new()
70 .write(true)
71 .open(&temp)
72 .and_then(|f| f.sync_all())
73 .map_err(|e| DbError::Backup(format!("fsync backup: {e}")))?;
74 std::fs::rename(&temp, target)
75 .map_err(|e| DbError::Backup(format!("publish backup: {e}")))?;
76 if let Some(parent) = target.parent().filter(|p| !p.as_os_str().is_empty()) {
77 rightkit_fs::atomic::sync_dir(parent);
78 }
79 Ok(())
80 })();
81 if result.is_err() {
82 remove_with_sidecars(&temp);
83 }
84 result
85}
86
87pub fn restore_from(
92 options: &DbOptions,
93 destination: &mut Connection,
94 backup_path: &Path,
95) -> Result<()> {
96 let source = validated_backup(options, backup_path)?;
97 restore_validated(options, &source, destination)
98}
99
100impl DbOptions {
101 pub fn restore_from(&self, backup_path: impl AsRef<Path>) -> Result<()> {
105 let source = validated_backup(self, backup_path.as_ref())?;
106 let mut destination = self.open_existing_writer()?;
107 restore_validated(self, &source, &mut destination)
108 }
109}
110
111fn validated_backup(options: &DbOptions, path: &Path) -> Result<Connection> {
112 let source = options.open_existing_at(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?;
113 source.execute_batch("BEGIN;")?;
115 let messages = {
116 let mut statement = source.prepare("PRAGMA integrity_check")?;
117 let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
118 rows.collect::<rusqlite::Result<Vec<_>>>()?
119 };
120 if messages.is_empty()
121 || messages
122 .iter()
123 .any(|message| !message.trim().eq_ignore_ascii_case("ok"))
124 {
125 return Err(DbError::Integrity(messages.join("; ")));
126 }
127 Ok(source)
128}
129
130fn restore_validated(
131 options: &DbOptions,
132 source: &Connection,
133 destination: &mut Connection,
134) -> Result<()> {
135 let backup = Backup::new(source, destination)?;
136 let started = Instant::now();
137 loop {
138 match backup.step(512)? {
139 StepResult::Done => return Ok(()),
140 StepResult::More => {}
141 StepResult::Busy | StepResult::Locked => {
142 if started.elapsed() >= options.busy_timeout {
143 return Err(DbError::Backup("restore busy timeout expired".into()));
144 }
145 std::thread::sleep(Duration::from_millis(25));
146 }
147 _ => return Err(DbError::Backup("unexpected restore step result".into())),
148 }
149 }
150}
151
152fn temp_path(target: &Path) -> Result<PathBuf> {
153 let name = target
154 .file_name()
155 .ok_or(DbError::InvalidDatabasePath)?
156 .to_string_lossy();
157 let n = COUNTER.fetch_add(1, Ordering::Relaxed);
158 Ok(target.with_file_name(format!(
159 ".{name}{}{}-{n}",
160 rightkit_fs::atomic::TEMP_MARKER,
161 std::process::id()
162 )))
163}
164
165fn remove_with_sidecars(path: &Path) {
166 let _ = std::fs::remove_file(path);
167 for suffix in ["-wal", "-shm", "-journal"] {
168 let mut name = path.as_os_str().to_os_string();
169 name.push(suffix);
170 let _ = std::fs::remove_file(PathBuf::from(name));
171 }
172}