defter 0.5.3

//! Backups of the library database to object storage, and restoring them.
//!
//! A run takes a consistent snapshot of each library whose version changed
//! since the last backup (`VACUUM INTO`, which reads in one transaction and
//! never blocks the writer under WAL), checks it, gzips it and uploads it under
//! `backup/libraries/<name>/`. A manifest under `backup/manifests/` then names
//! the snapshot of every library at that point, reusing unchanged ones.
//! Manifests older than the retention window are removed, the newest always
//! kept, and with them the snapshots no manifest names any more.
//!
//! Attachment files are not copied: they are in object storage already. While
//! backups run, a deleted file is moved to `deleted/<date>/` rather than removed
//! (see `Storage::remove`), and purged after the same window, so a restored
//! library finds its files.
//!
//! Zotero clients refuse a library whose version goes down. A restore therefore
//! stamps every version in the restored library above the highest it can have
//! served, so clients take it as newer and catch up.

use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use aws_lc_rs::digest;
use rusqlite::{Connection, OpenFlags};
use serde::{Deserialize, Serialize};

use crate::storage::{FileScope, Storage};

const MANIFESTS: &str = "backup/manifests/";
const SNAPSHOTS: &str = "backup/libraries/";
const DELETED: &str = "deleted/";

/// How far a restore puts versions above the highest known: room for writes
/// that reached clients but no backup.
const RESTORE_GAP: i64 = 1000;

/// A snapshot no manifest names is left this long, in case a run that wrote it
/// is still about to write its manifest.
const ORPHAN_GRACE_SECS: u64 = 24 * 3600;

/// A database to back up: a library's, or the control database.
#[derive(Debug, Clone)]
pub struct Source {
    /// Its name in backups: `u<user id>`, `g<group id>` or `control`.
    pub name: String,
    pub path: PathBuf,
    /// Where a library's attachment files are; `None` for the control
    /// database, which has neither files nor versions clients have seen.
    pub files: Option<FileScope>,
}

/// What there is to back up, asked for at every run, as libraries come and go.
pub type Sources = Arc<dyn Fn() -> Result<Vec<Source>, String> + Send + Sync>;

/// A library's snapshot, as a manifest names it.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Entry {
    pub name: String,
    /// The library version the snapshot holds.
    pub version: i64,
    /// Where it is, under the backup storage.
    pub key: String,
    pub bytes: u64,
    /// Of the gzipped file.
    pub sha256: String,
    /// The schema version (`PRAGMA user_version`).
    pub schema: i64,
    /// A digest of what the library keeps outside versioned data (its `meta`
    /// table: profile settings), which changes without a new version.
    #[serde(default)]
    pub state: String,
}

/// Every library at one point in time.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Manifest {
    pub created: String,
    pub defter: String,
    pub entries: Vec<Entry>,
}

/// What a run did, for the dashboard.
#[derive(Debug, Clone, Serialize)]
pub struct Report {
    /// When it finished, in seconds since the epoch.
    pub at: u64,
    /// The manifest it wrote, or `None` when nothing had changed.
    pub manifest: Option<String>,
    pub snapshots: usize,
    pub bytes: u64,
    pub error: Option<String>,
    /// The backup succeeded, but removing old ones did not.
    pub warning: Option<String>,
}

pub struct Backup {
    /// Where backups go.
    pub store: Arc<Storage>,
    /// Where attachment files are, whose deleted copies a run purges.
    pub files: Arc<Storage>,
    sources: Sources,
    /// For snapshots on their way up or down.
    pub tmp: PathBuf,
    pub keep_days: u64,
    running: tokio::sync::Mutex<()>,
    last: Mutex<Option<Report>>,
}

fn now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |d| d.as_secs())
}

/// Days since 1970-01-01 as a civil date (Howard Hinnant's algorithm).
fn civil(days: i64) -> (i64, u32, u32) {
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let day = (doy - (153 * mp + 2) / 5 + 1) as u32;
    let month = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
    (yoe + era * 400 + i64::from(month <= 2), month, day)
}

/// `20261102T060000Z`: sorts like the time it names.
pub fn timestamp(secs: u64) -> String {
    let (y, m, d) = civil((secs / 86_400) as i64);
    let s = secs % 86_400;
    format!(
        "{y:04}{m:02}{d:02}T{:02}{:02}{:02}Z",
        s / 3600,
        s / 60 % 60,
        s % 60
    )
}

/// `2026-11-02 06:00 UTC`, for people.
/// `secs` as Zotero's API writes times: `2026-11-02T06:00:00Z`.
pub fn iso_time(secs: u64) -> String {
    let (y, m, d) = civil((secs / 86_400) as i64);
    let s = secs % 86_400;
    format!(
        "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z",
        s / 3600,
        s / 60 % 60,
        s % 60
    )
}

pub fn utc_text(secs: u64) -> String {
    let (y, m, d) = civil((secs / 86_400) as i64);
    let s = secs % 86_400;
    format!(
        "{y:04}-{m:02}-{d:02} {:02}:{:02} UTC",
        s / 3600,
        s / 60 % 60
    )
}

/// `2026-11-02`, the folder deleted files wait in.
pub fn date(secs: u64) -> String {
    let (y, m, d) = civil((secs / 86_400) as i64);
    format!("{y:04}-{m:02}-{d:02}")
}

/// Seconds since the epoch of a `YYYY-MM-DD` date, at its start.
fn date_secs(text: &str) -> Option<u64> {
    let mut parts = text.split('-').map(|p| p.parse::<i64>().ok());
    let (y, m, d) = (parts.next()??, parts.next()??, parts.next()??);
    if parts.next().is_some() || !(1..=12).contains(&m) || !(1..=31).contains(&d) {
        return None;
    }
    // Days from civil, the inverse of `civil`.
    let y = if m <= 2 { y - 1 } else { y };
    let era = y.div_euclid(400);
    let yoe = y.rem_euclid(400);
    let doy = (153 * (if m > 2 { m - 3 } else { m + 9 }) + 2) / 5 + d - 1;
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
    let days = era * 146_097 + doe - 719_468;
    u64::try_from(days * 86_400).ok()
}

/// The library version of a library's database; for the control database, how
/// many changes it has seen.
fn library_version(conn: &Connection) -> rusqlite::Result<i64> {
    let library: bool = conn.query_row(
        "SELECT count(*) > 0 FROM sqlite_master WHERE type = 'table' AND name = 'library'",
        [],
        |r| r.get(0),
    )?;
    if library {
        conn.query_row("SELECT version FROM library WHERE id = 1", [], |r| r.get(0))
    } else {
        conn.query_row("SELECT n FROM revision", [], |r| r.get(0))
    }
}

fn read_only(path: &Path) -> Result<Connection, String> {
    let conn = Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
    )
    .map_err(|e| format!("opening {}: {e}", path.display()))?;
    conn.busy_timeout(std::time::Duration::from_secs(30))
        .map_err(|e| e.to_string())?;
    Ok(conn)
}

/// A digest of the library's `meta` table, but its cache of Zotero's schema.
fn state(conn: &Connection) -> rusqlite::Result<String> {
    let mut stmt =
        conn.prepare("SELECT key, value FROM meta WHERE key NOT LIKE 'schema%' ORDER BY key")?;
    let mut digest = digest::Context::new(&digest::SHA256);
    let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
    for row in rows {
        let (key, value) = row?;
        for part in [key.as_bytes(), b"\0", value.as_bytes(), b"\0"] {
            digest.update(part);
        }
    }
    Ok(hex(digest.finish().as_ref()))
}

/// Removes files when dropped, unless kept: what a step leaves behind when it
/// fails partway.
struct Leftovers(Vec<PathBuf>);

impl Leftovers {
    fn keep(mut self) {
        self.0.clear();
    }
}

impl Drop for Leftovers {
    fn drop(&mut self) {
        for path in &self.0 {
            let _ = std::fs::remove_file(path);
        }
    }
}

/// A check of the database at `conn`, which must answer `ok`.
fn integrity(conn: &Connection) -> Result<(), String> {
    let result: String = conn
        .query_row("PRAGMA integrity_check", [], |r| r.get(0))
        .map_err(|e| e.to_string())?;
    if result == "ok" {
        Ok(())
    } else {
        Err(format!("integrity check: {result}"))
    }
}

/// Hashes what passes through to a writer, and counts it.
struct Hashing<W> {
    inner: W,
    digest: digest::Context,
    bytes: u64,
}

impl<W: Write> Write for Hashing<W> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = self.inner.write(buf)?;
        self.digest.update(&buf[..n]);
        self.bytes += n as u64;
        Ok(n)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

fn hex(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

/// A gzipped, checked snapshot on disk.
struct Snapshot {
    path: PathBuf,
    version: i64,
    schema: i64,
    state: String,
    sha256: String,
    bytes: u64,
}

/// Snapshot the database at `src` into `tmp`, as `<name>-<ts>.sqlite3.gz`.
fn snapshot(src: &Path, tmp: &Path, name: &str, ts: &str) -> Result<Snapshot, String> {
    let raw = tmp.join(format!("{name}-{ts}.sqlite3"));
    let path = tmp.join(format!("{name}-{ts}.sqlite3.gz"));
    let _ = std::fs::remove_file(&raw);
    // A failure leaves nothing behind; on success only the gzipped file stays.
    let leftovers = Leftovers(vec![raw.clone(), path.clone()]);
    {
        let conn = read_only(src)?;
        conn.execute("VACUUM INTO ?1", [raw.to_string_lossy()])
            .map_err(|e| format!("snapshot of {}: {e}", src.display()))?;
    }
    let (version, schema, state) = {
        let copy = read_only(&raw)?;
        integrity(&copy)?;
        let version = library_version(&copy).map_err(|e| e.to_string())?;
        let schema: i64 = copy
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .map_err(|e| e.to_string())?;
        (version, schema, state(&copy).map_err(|e| e.to_string())?)
    };
    let out = std::fs::File::create(&path).map_err(|e| e.to_string())?;
    let mut gz = flate2::write::GzEncoder::new(
        Hashing {
            inner: out,
            digest: digest::Context::new(&digest::SHA256),
            bytes: 0,
        },
        flate2::Compression::default(),
    );
    let mut input = std::fs::File::open(&raw).map_err(|e| e.to_string())?;
    std::io::copy(&mut input, &mut gz).map_err(|e| e.to_string())?;
    let hashing = gz.finish().map_err(|e| e.to_string())?;
    hashing.inner.sync_all().map_err(|e| e.to_string())?;
    let _ = std::fs::remove_file(&raw);
    leftovers.keep();
    Ok(Snapshot {
        path,
        version,
        schema,
        state,
        sha256: hex(hashing.digest.finish().as_ref()),
        bytes: hashing.bytes,
    })
}

/// The timestamp a manifest key is named for.
fn manifest_ts(key: &str) -> Option<&str> {
    key.strip_prefix(MANIFESTS)?.strip_suffix(".json")
}

/// Seconds since the epoch of a `timestamp`.
fn ts_secs(ts: &str) -> Option<u64> {
    let (day, time) = ts.split_once('T')?;
    if day.len() != 8 || time.len() != 7 {
        return None;
    }
    let date = format!("{}-{}-{}", &day[..4], &day[4..6], &day[6..]);
    let h: u64 = time[..2].parse().ok()?;
    let m: u64 = time[2..4].parse().ok()?;
    let s: u64 = time[4..6].parse().ok()?;
    Some(date_secs(&date)? + h * 3600 + m * 60 + s)
}

/// All manifests, oldest first.
async fn manifest_keys(store: &Storage) -> Result<Vec<String>, String> {
    let mut keys: Vec<String> = store
        .list(MANIFESTS)
        .await
        .map_err(|e| e.to_string())?
        .into_iter()
        .map(|o| o.key)
        .filter(|k| manifest_ts(k).is_some())
        .collect();
    keys.sort();
    Ok(keys)
}

async fn manifest(store: &Storage, key: &str) -> Result<Manifest, String> {
    let bytes = store.get_bytes(key).await.map_err(|e| e.to_string())?;
    serde_json::from_slice(&bytes).map_err(|e| format!("{key}: {e}"))
}

impl Backup {
    pub fn new(
        store: Arc<Storage>,
        files: Arc<Storage>,
        sources: Sources,
        tmp: PathBuf,
        keep_days: u64,
    ) -> Self {
        Backup {
            store,
            files,
            sources,
            tmp,
            keep_days,
            running: tokio::sync::Mutex::new(()),
            last: Mutex::new(None),
        }
    }

    /// What there is to back up now.
    pub fn sources(&self) -> Result<Vec<Source>, String> {
        (self.sources)()
    }

    /// What the last run did.
    pub fn last(&self) -> Option<Report> {
        self.last.lock().unwrap_or_else(|e| e.into_inner()).clone()
    }

    /// Back up now. A run already going makes this one fail rather than wait.
    pub async fn run(&self) -> Report {
        self.run_at(now()).await
    }

    async fn run_at(&self, at: u64) -> Report {
        let report = match self.running.try_lock() {
            Ok(_running) => match self.back_up(at).await {
                Ok(report) => report,
                Err(error) => Report {
                    at: now(),
                    manifest: None,
                    snapshots: 0,
                    bytes: 0,
                    error: Some(error),
                    warning: None,
                },
            },
            Err(_) => Report {
                at: now(),
                manifest: None,
                snapshots: 0,
                bytes: 0,
                error: Some("a backup is already running".to_string()),
                warning: None,
            },
        };
        match &report.error {
            None => tracing::info!(
                manifest = report
                    .manifest
                    .as_deref()
                    .unwrap_or("none, nothing changed"),
                snapshots = report.snapshots,
                bytes = report.bytes,
                "backup"
            ),
            Some(error) => tracing::error!(%error, "backup failed"),
        }
        *self.last.lock().unwrap_or_else(|e| e.into_inner()) = Some(report.clone());
        report
    }

    async fn back_up(&self, at: u64) -> Result<Report, String> {
        std::fs::create_dir_all(&self.tmp)
            .map_err(|e| format!("creating {}: {e}", self.tmp.display()))?;
        let ts = timestamp(at);
        let keys = manifest_keys(&self.store).await?;
        let previous = match keys.last() {
            Some(key) => Some(manifest(&self.store, key).await?),
            None => None,
        };
        let mut entries = Vec::new();
        let (mut snapshots, mut bytes) = (0, 0);
        let mut missing = Vec::new();
        for source in &self.sources()? {
            let before = previous
                .as_ref()
                .and_then(|m| m.entries.iter().find(|e| e.name == source.name));
            if !source.path.exists() {
                // A library nobody synced to yet has no file. One that went
                // missing keeps its last snapshot, and the run says so.
                if let Some(entry) = before {
                    entries.push(entry.clone());
                    missing.push(source.path.display().to_string());
                }
                continue;
            }
            let path = source.path.clone();
            let (live, live_state) = tokio::task::spawn_blocking(move || {
                let conn = read_only(&path)?;
                let version = library_version(&conn).map_err(|e| e.to_string())?;
                Ok::<_, String>((version, state(&conn).map_err(|e| e.to_string())?))
            })
            .await
            .map_err(|e| e.to_string())??;
            let unchanged = before.filter(|e| e.version == live && e.state == live_state);
            if let Some(entry) = unchanged {
                entries.push(entry.clone());
                continue;
            }
            let (path, tmp, name, ts2) = (
                source.path.clone(),
                self.tmp.clone(),
                source.name.clone(),
                ts.clone(),
            );
            let snap = tokio::task::spawn_blocking(move || snapshot(&path, &tmp, &name, &ts2))
                .await
                .map_err(|e| e.to_string())??;
            let key = format!(
                "{SNAPSHOTS}{}/v{}-{ts}.sqlite3.gz",
                source.name, snap.version
            );
            let uploaded = self.store.put_file(&key, &snap.path).await;
            let _ = std::fs::remove_file(&snap.path);
            uploaded.map_err(|e| format!("uploading {key}: {e}"))?;
            snapshots += 1;
            bytes += snap.bytes;
            entries.push(Entry {
                name: source.name.clone(),
                version: snap.version,
                key,
                bytes: snap.bytes,
                sha256: snap.sha256,
                schema: snap.schema,
                state: snap.state,
            });
        }
        let manifest = if snapshots == 0 && previous.is_some() {
            None
        } else {
            let key = format!("{MANIFESTS}{ts}.json");
            let body = serde_json::to_vec_pretty(&Manifest {
                created: ts.clone(),
                defter: env!("CARGO_PKG_VERSION").to_string(),
                entries,
            })
            .map_err(|e| e.to_string())?;
            self.store
                .put_bytes(&key, body)
                .await
                .map_err(|e| format!("uploading {key}: {e}"))?;
            Some(key)
        };
        let mut warnings = Vec::new();
        if !missing.is_empty() {
            let error = format!("missing, kept as last backed up: {}", missing.join(", "));
            tracing::warn!(%error, "backing up");
            warnings.push(error);
        }
        if let Err(error) = self.prune(at).await {
            tracing::warn!(%error, "removing old backups");
            warnings.push(error);
        }
        let warning = (!warnings.is_empty()).then(|| warnings.join("; "));
        Ok(Report {
            at: now(),
            manifest,
            snapshots,
            bytes,
            error: None,
            warning,
        })
    }

    /// Remove what is past the retention window: manifests but the newest,
    /// snapshots no manifest names, and deleted attachment files.
    async fn prune(&self, at: u64) -> Result<(), String> {
        let cutoff = at.saturating_sub(self.keep_days * 86_400);
        let keys = manifest_keys(&self.store).await?;
        let mut kept = Vec::new();
        for (i, key) in keys.iter().enumerate() {
            let newest = i + 1 == keys.len();
            let old = manifest_ts(key)
                .and_then(ts_secs)
                .is_some_and(|t| t < cutoff);
            if old && !newest {
                self.store.delete(key).await.map_err(|e| e.to_string())?;
            } else {
                kept.push(key.clone());
            }
        }
        let mut named = std::collections::HashSet::new();
        for key in &kept {
            for entry in manifest(&self.store, key).await?.entries {
                named.insert(entry.key);
            }
        }
        for object in self
            .store
            .list(SNAPSHOTS)
            .await
            .map_err(|e| e.to_string())?
        {
            let orphan = !named.contains(&object.key) && object.modified + ORPHAN_GRACE_SECS < at;
            if orphan {
                self.store
                    .delete(&object.key)
                    .await
                    .map_err(|e| e.to_string())?;
            }
        }
        for object in self.files.list(DELETED).await.map_err(|e| e.to_string())? {
            let day = object.key[DELETED.len()..].split('/').next().unwrap_or("");
            if date_secs(day).is_some_and(|t| t + 86_400 <= cutoff) {
                self.files
                    .delete(&object.key)
                    .await
                    .map_err(|e| e.to_string())?;
            }
        }
        Ok(())
    }
}

/// What `defter backup --list` shows: each manifest and what it holds.
pub async fn list(store: &Storage) -> Result<Vec<Manifest>, String> {
    let mut manifests = Vec::new();
    for key in manifest_keys(store).await? {
        manifests.push(manifest(store, &key).await?);
    }
    Ok(manifests)
}

/// What a restore did, or would do.
#[derive(Debug, Clone, PartialEq)]
pub struct Restored {
    pub manifest: String,
    pub snapshot_version: i64,
    /// What every version in the library becomes.
    pub version: i64,
    /// Where the database it replaced was kept.
    pub previous: Option<PathBuf>,
    /// Attachment files brought back from `deleted/`, and those still missing.
    pub files_restored: usize,
    pub files_missing: usize,
}

/// The highest library version among the live database and copies earlier
/// restores kept of it, `None` if none can be read.
fn highest_live_version(path: &Path) -> Option<i64> {
    let readable = |p: &Path| {
        read_only(p)
            .and_then(|c| library_version(&c).map_err(|e| e.to_string()))
            .ok()
    };
    let mut highest = readable(path);
    let kept = format!("{}.before-restore-", path.file_name()?.to_string_lossy());
    if let Ok(entries) = std::fs::read_dir(path.parent()?) {
        for entry in entries.flatten() {
            let name = entry.file_name().to_string_lossy().to_string();
            if name.starts_with(&kept) && !name.ends_with("-wal") && !name.ends_with("-shm") {
                highest = highest.max(readable(&entry.path()));
            }
        }
    }
    highest
}

/// Restore `source` from the manifest `wanted` (a timestamp, or `None` for the
/// newest). Only with `apply` is anything changed; the server must not be
/// running (see [`lock`]).
///
/// A library's versions are stamped above the live database's, which clients
/// may have seen beyond any backup, above `min_version`, and above `served`, the
/// highest version the control database recorded serving. If neither the live
/// database nor a copy an earlier restore kept can be read, `min_version` must
/// say what clients may have seen, unless `served` is newer than every backup.
/// The control database comes back as it was, its change count moved on past
/// every backup's.
#[allow(
    clippy::too_many_arguments,
    reason = "each is a separate input the restore needs"
)]
pub async fn restore(
    store: &Storage,
    files: &Storage,
    source: &Source,
    wanted: Option<&str>,
    tmp: &Path,
    apply: bool,
    min_version: Option<i64>,
    served: Option<i64>,
) -> Result<Restored, String> {
    let keys = manifest_keys(store).await?;
    let key = match wanted {
        None => keys.last().cloned(),
        Some(ts) => keys.iter().find(|k| manifest_ts(k) == Some(ts)).cloned(),
    }
    .ok_or_else(|| match wanted {
        None => "no backup found".to_string(),
        Some(ts) => format!("no backup from {ts}; see `defter backup --list`"),
    })?;
    let chosen = manifest(store, &key).await?;
    let entry = chosen
        .entries
        .iter()
        .find(|e| e.name == source.name)
        .cloned()
        .ok_or_else(|| format!("{key} has no library {}", source.name))?;
    // The highest version any backup recorded, which clients may have seen.
    let mut highest = entry.version;
    for other in &keys {
        for e in manifest(store, other).await?.entries {
            if e.name == source.name {
                highest = highest.max(e.version);
            }
        }
    }

    let live_path = source.path.clone();
    let live = tokio::task::spawn_blocking(move || highest_live_version(&live_path))
        .await
        .map_err(|e| e.to_string())?;
    let library = source.files.is_some();
    let served = served.unwrap_or(0);
    let floor = match (live, min_version) {
        _ if !library => 0,
        (Some(live), min) => live.max(min.unwrap_or(0)).max(served),
        (None, Some(min)) => min.max(served),
        // Recorded after the newest backup: it covers what clients saw since.
        (None, None) if served > highest => served,
        (None, None) => {
            return Err(format!(
                "cannot read the library at {} to know what versions clients have seen. Is \
                 this the server's data directory (DEFTER_DATA_DIR)? If the database is \
                 really gone, pass --min-version with a version at least as high as any \
                 client's, such as the Last-Modified-Version in the server's last log lines",
                source.path.display()
            ));
        }
    };

    std::fs::create_dir_all(tmp).map_err(|e| format!("creating {}: {e}", tmp.display()))?;
    if let Some(dir) = source.path.parent() {
        std::fs::create_dir_all(dir).map_err(|e| format!("creating {}: {e}", dir.display()))?;
    }
    let gz = tmp.join(format!("restore-{}.sqlite3.gz", source.name));
    let restoring = source.path.with_extension("sqlite3.restoring");
    let leftovers = Leftovers(vec![gz.clone(), restoring.clone()]);
    store
        .get_to_file(&entry.key, &gz)
        .await
        .map_err(|e| format!("downloading {}: {e}", entry.key))?;
    let (gz2, restoring2, entry2) = (gz.clone(), restoring.clone(), entry.clone());
    let version = tokio::task::spawn_blocking(move || -> Result<i64, String> {
        // The download must be what was uploaded.
        let mut digest = digest::Context::new(&digest::SHA256);
        let mut file = std::fs::File::open(&gz2).map_err(|e| e.to_string())?;
        let mut buffer = vec![0u8; 1 << 20];
        loop {
            let n = std::io::Read::read(&mut file, &mut buffer).map_err(|e| e.to_string())?;
            if n == 0 {
                break;
            }
            digest.update(&buffer[..n]);
        }
        if hex(digest.finish().as_ref()) != entry2.sha256 {
            return Err(format!("{} does not match its checksum", entry2.key));
        }
        let _ = std::fs::remove_file(&restoring2);
        let mut out = std::fs::File::create(&restoring2).map_err(|e| e.to_string())?;
        let mut gunzip =
            flate2::read::GzDecoder::new(std::fs::File::open(&gz2).map_err(|e| e.to_string())?);
        std::io::copy(&mut gunzip, &mut out).map_err(|e| e.to_string())?;
        out.sync_all().map_err(|e| e.to_string())?;
        let _ = std::fs::remove_file(&gz2);

        let mut conn = Connection::open(&restoring2).map_err(|e| e.to_string())?;
        integrity(&conn)?;
        if !library {
            // On past every backup's count, so the next backup cannot take it
            // for one it already has; and groups' versions past any a client
            // may have seen since, or it would not see their changes.
            // (The groups first: changing them counts as a change.)
            conn.execute("UPDATE grp SET version = version + ?1", [RESTORE_GAP])
                .map_err(|e| e.to_string())?;
            let count = highest + 1;
            conn.execute("UPDATE revision SET n = ?1", [count])
                .map_err(|e| e.to_string())?;
            return Ok(count);
        }
        let version = highest.max(floor) + RESTORE_GAP;
        stamp(&mut conn, version).map_err(|e| e.to_string())?;
        Ok(version)
    })
    .await
    .map_err(|e| e.to_string())??;

    let mut restored = Restored {
        manifest: key,
        snapshot_version: entry.version,
        version,
        previous: None,
        files_restored: 0,
        files_missing: 0,
    };
    if !apply {
        return Ok(restored);
    }

    // Keep what is replaced, with its write-ahead log, which must never meet
    // the restored file: SQLite would replay it onto it.
    let kept = source
        .path
        .with_extension(format!("sqlite3.before-restore-{}", timestamp(now())));
    for suffix in ["", "-wal", "-shm"] {
        let from = PathBuf::from(format!("{}{suffix}", source.path.display()));
        if from.exists() {
            let to = PathBuf::from(format!("{}{suffix}", kept.display()));
            std::fs::rename(&from, &to).map_err(|e| format!("keeping {}: {e}", from.display()))?;
            if suffix.is_empty() {
                restored.previous = Some(kept.clone());
            }
        }
    }
    std::fs::rename(&restoring, &source.path)
        .map_err(|e| format!("moving the restored library into place: {e}"))?;
    leftovers.keep();
    let Some(scope) = &source.files else {
        return Ok(restored);
    };

    // Bring back attachment files deleted since, from where they wait.
    let path = source.path.clone();
    let item_keys = tokio::task::spawn_blocking(move || -> Result<Vec<String>, String> {
        let conn = read_only(&path)?;
        let mut stmt = conn
            .prepare("SELECT item_key FROM file")
            .map_err(|e| e.to_string())?;
        stmt.query_map([], |r| r.get(0))
            .map_err(|e| e.to_string())?
            .collect::<Result<_, _>>()
            .map_err(|e| e.to_string())
    })
    .await
    .map_err(|e| e.to_string())??;
    // A file the restored library knows that was deleted or replaced since its
    // snapshot was set aside at its first change after it: the earliest copy set
    // aside from then on is the file as the snapshot knew it.
    let since = entry
        .key
        .rsplit_once('-')
        .and_then(|(_, rest)| rest.strip_suffix(".sqlite3.gz"))
        .unwrap_or("")
        .to_string();
    let mut aside: std::collections::HashMap<String, String> = Default::default();
    for object in files.list(DELETED).await.map_err(|e| e.to_string())? {
        // deleted/<date>/<time>-<n>/<the file's key>
        let mut parts = object.key[DELETED.len()..].splitn(3, '/');
        let (Some(_day), Some(at), Some(path)) = (parts.next(), parts.next(), parts.next()) else {
            continue;
        };
        if at.get(..since.len()).is_some_and(|t| t < since.as_str()) {
            continue;
        }
        let earlier = aside.get(path).is_none_or(|k| object.key < *k);
        if earlier {
            aside.insert(path.to_string(), object.key.clone());
        }
    }
    for item_key in item_keys {
        let path = scope.key(&item_key);
        let present = files.exists(&path).await.map_err(|e| e.to_string())?;
        match aside.get(&path) {
            Some(copy) => {
                if present {
                    // The newer file waits in turn, like any replaced one.
                    files.set_aside(&path).await.map_err(|e| e.to_string())?;
                }
                files.copy(copy, &path).await.map_err(|e| e.to_string())?;
                restored.files_restored += 1;
            }
            None if !present => restored.files_missing += 1,
            None => {}
        }
    }
    Ok(restored)
}

/// Set every version in a library to `version`, the objects' JSON included.
fn stamp(conn: &mut Connection, version: i64) -> rusqlite::Result<()> {
    let tx = conn.transaction()?;
    // A deleted tag used again before the snapshot was taken may still be in the
    // deletion log; stamped, it would make clients strip it from every item.
    tx.execute(
        "DELETE FROM deletion WHERE kind = 'tag' AND key IN (SELECT tag FROM item_tag)",
        [],
    )?;
    tx.execute("UPDATE library SET version = ?1 WHERE id = 1", [version])?;
    tx.execute(
        "UPDATE object SET version = ?1,
             data = CASE WHEN json_type(data, '$.version') IS NULL THEN data
                         ELSE json_set(data, '$.version', ?1) END",
        [version],
    )?;
    for table in ["setting", "deletion", "file", "fulltext"] {
        tx.execute(&format!("UPDATE {table} SET version = ?1"), [version])?;
    }
    tx.commit()
}

/// Hold the data directory: the server does while it runs, and a restore needs
/// it to itself. The lock lasts as long as the returned file is open.
pub fn lock(data_dir: &Path) -> Result<std::fs::File, String> {
    use std::os::fd::AsRawFd;
    std::fs::create_dir_all(data_dir)
        .map_err(|e| format!("creating {}: {e}", data_dir.display()))?;
    let path = data_dir.join("defter.lock");
    let file = std::fs::OpenOptions::new()
        .create(true)
        .truncate(false)
        .write(true)
        .open(&path)
        .map_err(|e| format!("opening {}: {e}", path.display()))?;
    // SAFETY: flock on a file descriptor this function owns.
    let locked = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } == 0;
    if locked {
        Ok(file)
    } else {
        Err(format!(
            "{} is in use: is defter running? Stop it first",
            data_dir.display()
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::Db;
    use crate::storage::Backend;
    use serde_json::json;
    use std::time::Duration;

    #[test]
    fn writes_times_that_sort() {
        assert_eq!(timestamp(0), "19700101T000000Z");
        assert_eq!(timestamp(1_793_599_200), "20261102T060000Z");
        assert_eq!(utc_text(1_793_599_200), "2026-11-02 06:00 UTC");
        assert_eq!(iso_time(1_793_599_205), "2026-11-02T06:00:05Z");
        assert_eq!(date(1_793_599_200), "2026-11-02");
        assert_eq!(ts_secs("20261102T060000Z"), Some(1_793_599_200));
        assert_eq!(date_secs("2026-11-02"), Some(1_793_577_600));
        assert_eq!(
            date_secs("2024-02-29").map(date),
            Some("2024-02-29".to_string())
        );
        for bad in ["2026-13-01", "2026-11", "x", ""] {
            assert_eq!(date_secs(bad), None, "{bad}");
        }
        assert_eq!(ts_secs("2026"), None);
    }

    struct Setup {
        _dir: tempfile::TempDir,
        db: Arc<Db>,
        backup: Backup,
        path: PathBuf,
        source: Source,
    }

    fn setup(keep_days: u64) -> Setup {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("defter.sqlite3");
        let db = Db::open(&path).unwrap();
        let store =
            Arc::new(Storage::new(&Backend::Memory, Duration::from_secs(60), true).unwrap());
        let files =
            Arc::new(Storage::new(&Backend::Memory, Duration::from_secs(60), true).unwrap());
        let source = Source {
            name: "u1".to_string(),
            path: path.clone(),
            files: Some(FileScope::Legacy),
        };
        let sources = vec![source.clone()];
        let backup = Backup::new(
            store,
            files,
            Arc::new(move || Ok(sources.clone())),
            dir.path().join("tmp"),
            keep_days,
        );
        Setup {
            _dir: dir,
            db,
            backup,
            path,
            source,
        }
    }

    async fn post(db: &Arc<Db>, items: serde_json::Value) {
        let items: Vec<serde_json::Value> = serde_json::from_value(items).unwrap();
        db.write(move |c| {
            let version = crate::store::current_version(c)?;
            crate::store::write_batch(c, crate::store::Kind::Item, items, Some(version), None)?;
            Ok(())
        })
        .await
        .unwrap();
    }

    const T0: u64 = 1_793_599_200;
    const DAY: u64 = 86_400;

    #[tokio::test]
    async fn backs_up_only_what_changed() {
        let s = setup(30);
        post(
            &s.db,
            json!([{"key": "AAAAAAAA", "itemType": "book", "title": "One"}]),
        )
        .await;
        let first = s.backup.run_at(T0).await;
        assert_eq!(first.error, None);
        assert_eq!(first.snapshots, 1);
        assert_eq!(
            first.manifest.as_deref(),
            Some("backup/manifests/20261102T060000Z.json")
        );

        // Nothing changed: no snapshot and no manifest.
        let second = s.backup.run_at(T0 + 3600).await;
        assert_eq!((second.snapshots, second.manifest), (0, None));

        post(
            &s.db,
            json!([{"key": "BBBBBBBB", "itemType": "book", "title": "Two"}]),
        )
        .await;
        let third = s.backup.run_at(T0 + 7200).await;
        assert_eq!(third.snapshots, 1);
        let manifests = list(&s.backup.store).await.unwrap();
        assert_eq!(manifests.len(), 2);
        assert_eq!(manifests[1].entries[0].version, 2);
        assert!(
            manifests[1].entries[0]
                .key
                .starts_with("backup/libraries/u1/v2-")
        );
        assert_eq!(s.backup.last().unwrap().snapshots, 1);
    }

    #[tokio::test]
    async fn keeps_backups_for_the_retention_window() {
        let s = setup(7);
        // Objects get the real time as they are written, so the runs start now.
        let t0 = now();
        for day in 0..10 {
            post(
                &s.db,
                json!([{"key": format!("AAAAAAA{}", (b'A' + day as u8) as char),
                                "itemType": "book", "title": "x"}]),
            )
            .await;
            assert_eq!(s.backup.run_at(t0 + day * DAY).await.error, None);
        }
        let manifests = list(&s.backup.store).await.unwrap();
        // Days 2 to 9 are within seven days of day 9.
        assert_eq!(
            manifests.len(),
            8,
            "{:?}",
            manifests.iter().map(|m| &m.created).collect::<Vec<_>>()
        );
        // The snapshots only the removed manifests named go too, past their grace.
        assert_eq!(s.backup.store.list(SNAPSHOTS).await.unwrap().len(), 8);

        // The newest manifest stays, however old.
        let alone = setup(1);
        post(
            &alone.db,
            json!([{"key": "AAAAAAAA", "itemType": "book", "title": "x"}]),
        )
        .await;
        alone.backup.run_at(T0).await;
        alone.backup.run_at(T0 + 100 * DAY).await;
        assert_eq!(list(&alone.backup.store).await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn purges_deleted_files_after_the_window() {
        let s = setup(7);
        post(
            &s.db,
            json!([{"key": "AAAAAAAA", "itemType": "book", "title": "x"}]),
        )
        .await;
        let old = format!("deleted/{}/files/OLDOLDOL", date(T0 - 9 * DAY));
        let recent = format!("deleted/{}/files/NEWNEWNE", date(T0 - DAY));
        for key in [&old, &recent] {
            s.backup
                .files
                .put_bytes(key, b"bytes".to_vec())
                .await
                .unwrap();
        }
        s.backup.run_at(T0).await;
        assert!(!s.backup.files.exists(&old).await.unwrap());
        assert!(s.backup.files.exists(&recent).await.unwrap());
    }

    #[tokio::test]
    async fn restores_above_every_version_served_and_brings_files_back() {
        let s = setup(30);
        post(
            &s.db,
            json!([{"key": "AAAAAAAA", "itemType": "book", "title": "Kept"}]),
        )
        .await;
        s.db.write(|c| {
            crate::store::write_settings(c, json!({"tagColors": {"value": [1]}}), Some(1))?;
            Ok(())
        })
        .await
        .unwrap();
        let backed_up = s.backup.run_at(T0).await;
        let wanted = manifest_ts(backed_up.manifest.as_deref().unwrap())
            .unwrap()
            .to_string();
        // A file that is later deleted waits under deleted/.
        s.db.write(|c| {
            c.execute(
                "INSERT INTO file (item_key, md5, filename, filesize, mtime, version)
                 VALUES ('AAAAAAAA', 'm', 'a.pdf', 5, 0, 3)",
                [],
            )?;
            c.execute("UPDATE library SET version = 3", [])?;
            Ok(())
        })
        .await
        .unwrap();
        let second = s.backup.run_at(T0 + 60).await;
        assert_eq!(second.snapshots, 1);
        s.backup
            .files
            .put_bytes(
                &format!(
                    "deleted/{}/{}-1/files/AAAAAAAA",
                    date(T0),
                    timestamp(T0 + 120)
                ),
                b"bytes".to_vec(),
            )
            .await
            .unwrap();
        // Later writes that no backup holds.
        post(
            &s.db,
            json!([{"key": "BBBBBBBB", "itemType": "book", "title": "Lost"}]),
        )
        .await;
        drop(s.db);

        let tmp = s.path.parent().unwrap().join("tmp");
        // Without applying, nothing changes.
        let plan = restore(
            &s.backup.store,
            &s.backup.files,
            &s.source,
            None,
            &tmp,
            false,
            None,
            None,
        )
        .await
        .unwrap();
        assert_eq!(plan.snapshot_version, 3);
        assert_eq!(plan.version, 4 + RESTORE_GAP, "above the live library's 4");
        assert!(!s.path.with_extension("sqlite3.restoring").exists());
        assert!(!s.backup.files.exists("files/AAAAAAAA").await.unwrap());

        let done = restore(
            &s.backup.store,
            &s.backup.files,
            &s.source,
            None,
            &tmp,
            true,
            None,
            None,
        )
        .await
        .unwrap();
        assert_eq!(done.files_restored, 1);
        assert_eq!(done.files_missing, 0);
        assert!(s.backup.files.exists("files/AAAAAAAA").await.unwrap());
        assert!(done.previous.as_ref().unwrap().exists());
        let conn = read_only(&s.path).unwrap();
        assert_eq!(library_version(&conn).unwrap(), 1004);
        let (version, data_version): (i64, i64) = conn
            .query_row(
                "SELECT version, json_extract(data, '$.version') FROM object WHERE key = 'AAAAAAAA'",
                [],
                |r| Ok((r.get(0)?, r.get(1)?)),
            )
            .unwrap();
        assert_eq!((version, data_version), (1004, 1004));
        let lost: i64 = conn
            .query_row(
                "SELECT count(*) FROM object WHERE key = 'BBBBBBBB'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(lost, 0);
        for table in ["setting", "file"] {
            let v: i64 = conn
                .query_row(&format!("SELECT version FROM {table}"), [], |r| r.get(0))
                .unwrap();
            assert_eq!(v, 1004, "{table}");
        }

        // An older backup comes back above the newer one too.
        let older = restore(
            &s.backup.store,
            &s.backup.files,
            &s.source,
            Some(&wanted),
            &tmp,
            true,
            None,
            None,
        )
        .await
        .unwrap();
        assert_eq!(older.snapshot_version, 2);
        assert_eq!(older.version, 1004 + RESTORE_GAP);
        assert!(
            restore(
                &s.backup.store,
                &s.backup.files,
                &s.source,
                Some("19990101T000000Z"),
                &tmp,
                false,
                None,
                None,
            )
            .await
            .is_err()
        );
    }

    #[tokio::test]
    async fn backs_up_profile_settings_that_make_no_version() {
        let s = setup(30);
        assert_eq!(s.backup.run_at(T0).await.snapshots, 1);
        s.db.write(|c| {
            c.execute(
                "INSERT INTO meta (key, value) VALUES ('profile_hidden', '1')",
                [],
            )?;
            Ok(())
        })
        .await
        .unwrap();
        assert_eq!(s.backup.run_at(T0 + 60).await.snapshots, 1);
        assert_eq!(s.backup.run_at(T0 + 120).await.snapshots, 0);
    }

    #[tokio::test]
    async fn restores_the_file_a_snapshot_knew_when_it_was_replaced() {
        let s = setup(30);
        post(
            &s.db,
            json!([{"key": "AAAAAAAA", "itemType": "attachment", "linkMode": "imported_file"}]),
        )
        .await;
        s.db.write(|c| {
            c.execute(
                "INSERT INTO file (item_key, md5, filename, filesize, mtime, version)
                 VALUES ('AAAAAAAA', 'm', 'a.pdf', 3, 0, 1)",
                [],
            )?;
            Ok(())
        })
        .await
        .unwrap();
        s.backup.files.keep_deleted(true);
        s.backup
            .files
            .put_bytes("files/AAAAAAAA", b"old".to_vec())
            .await
            .unwrap();
        // Back up as of a minute ago, so what follows counts as after it.
        assert_eq!(s.backup.run_at(now() - 60).await.error, None);
        for (token, bytes) in [("t1", b"new" as &'static [u8]), ("t2", b"newer")] {
            s.backup
                .files
                .receive(
                    token,
                    futures::stream::iter([Ok::<_, String>(axum::body::Bytes::from_static(bytes))]),
                    1024,
                )
                .await
                .unwrap();
            s.backup
                .files
                .commit(token, &FileScope::Legacy, "AAAAAAAA")
                .await
                .unwrap();
        }
        drop(s.db);
        let tmp = s.path.parent().unwrap().join("tmp");
        let done = restore(
            &s.backup.store,
            &s.backup.files,
            &s.source,
            None,
            &tmp,
            true,
            None,
            None,
        )
        .await
        .unwrap();
        assert_eq!((done.files_restored, done.files_missing), (1, 0));
        assert_eq!(
            s.backup.files.get_bytes("files/AAAAAAAA").await.unwrap(),
            b"old"
        );
        // The newest file is set aside in turn, not lost.
        let mut aside = Vec::new();
        for object in s.backup.files.list(DELETED).await.unwrap() {
            aside.push(s.backup.files.get_bytes(&object.key).await.unwrap());
        }
        assert!(aside.contains(&b"newer".to_vec()), "{aside:?}");
    }

    #[tokio::test]
    async fn restore_needs_to_know_what_clients_have_seen() {
        let s = setup(30);
        post(
            &s.db,
            json!([{"key": "AAAAAAAA", "itemType": "book", "title": "x"}]),
        )
        .await;
        assert_eq!(s.backup.run_at(T0).await.error, None);
        drop(s.db);
        let tmp = s.path.parent().unwrap().join("tmp");
        let run = |min| {
            restore(
                &s.backup.store,
                &s.backup.files,
                &s.source,
                None,
                &tmp,
                true,
                min,
                None,
            )
        };
        let first = run(None).await.unwrap();
        assert_eq!(first.version, 1 + RESTORE_GAP);
        // A restore interrupted after setting the library aside still counts it.
        std::fs::rename(&s.path, s.path.with_extension("sqlite3.before-restore-x")).unwrap();
        let second = run(None).await.unwrap();
        assert_eq!(second.version, 1 + 2 * RESTORE_GAP);
        // With nothing left to read, it takes being told.
        for entry in std::fs::read_dir(s.path.parent().unwrap()).unwrap() {
            let path = entry.unwrap().path();
            if path.to_string_lossy().contains("sqlite3") {
                std::fs::remove_file(path).unwrap();
            }
        }
        let refused = run(None).await.unwrap_err();
        assert!(refused.contains("--min-version"), "{refused}");
        assert!(!s.path.exists());
        // What the server recorded serving stands in, if newer than every backup.
        let with_served = |served| {
            restore(
                &s.backup.store,
                &s.backup.files,
                &s.source,
                None,
                &tmp,
                false,
                None,
                Some(served),
            )
        };
        assert!(with_served(1).await.is_err(), "no newer than the backup");
        assert_eq!(with_served(7000).await.unwrap().version, 7000 + RESTORE_GAP);
        assert_eq!(run(Some(5000)).await.unwrap().version, 5000 + RESTORE_GAP);
    }

    #[tokio::test]
    async fn restores_the_control_database_and_counts_on() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("control.sqlite3");
        let control = crate::control::Control::open(&path).unwrap();
        let source = Source {
            name: "control".to_string(),
            path: path.clone(),
            files: None,
        };
        let sources = vec![source.clone()];
        let store =
            Arc::new(Storage::new(&Backend::Memory, Duration::from_secs(60), true).unwrap());
        let files =
            Arc::new(Storage::new(&Backend::Memory, Duration::from_secs(60), true).unwrap());
        let backup = Backup::new(
            store,
            files,
            Arc::new(move || Ok(sources.clone())),
            dir.path().join("tmp"),
            30,
        );
        // A group, which clients sync at its version.
        let owner = crate::registry::owner(9, "nine", false);
        control
            .db
            .write(move |c| crate::control::insert_user(c, &owner))
            .await
            .unwrap();
        let group = control.create_group(9, "G".into()).await.unwrap();
        let first = backup.run_at(T0).await;
        let wanted = manifest_ts(first.manifest.as_deref().unwrap())
            .unwrap()
            .to_string();
        for id in [1, 2] {
            let user = crate::registry::owner(id, &format!("u{id}"), false);
            control
                .db
                .write(move |c| crate::control::insert_user(c, &user))
                .await
                .unwrap();
        }
        assert_eq!(backup.run_at(T0 + 60).await.snapshots, 1);
        let newest = list(&backup.store).await.unwrap()[1].entries[0].version;
        drop(control);

        // Into a data directory that lost it.
        std::fs::remove_file(&path).unwrap();
        let tmp = dir.path().join("tmp");
        let done = restore(
            &backup.store,
            &backup.files,
            &source,
            Some(&wanted),
            &tmp,
            true,
            None,
            None,
        )
        .await
        .unwrap();
        assert_eq!(done.version, newest + 1);
        // Groups' versions move on past any a client may have seen.
        let restored_group: i64 = read_only(&path)
            .unwrap()
            .query_row("SELECT coalesce(max(version), -1) FROM grp", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!(restored_group, group.version + RESTORE_GAP);
        let conn = read_only(&path).unwrap();
        let users: i64 = conn
            .query_row("SELECT count(*) FROM user", [], |r| r.get(0))
            .unwrap();
        assert_eq!(users, 1, "as it was then");
        // So the next backup takes it, though the count it had matches a newer one.
        assert_eq!(library_version(&conn).unwrap(), newest + 1);
        drop(conn);
        assert_eq!(backup.run_at(T0 + 120).await.snapshots, 1);
    }

    #[tokio::test]
    async fn moves_a_stray_write_ahead_log_aside() {
        let s = setup(30);
        assert_eq!(s.backup.run_at(T0).await.error, None);
        drop(s.db);
        let wal = PathBuf::from(format!("{}-wal", s.path.display()));
        std::fs::remove_file(&s.path).unwrap();
        std::fs::write(&wal, b"stale").unwrap();
        let tmp = s.path.parent().unwrap().join("tmp");
        let done = restore(
            &s.backup.store,
            &s.backup.files,
            &s.source,
            None,
            &tmp,
            true,
            Some(10),
            None,
        )
        .await
        .unwrap();
        assert!(done.previous.is_none());
        assert!(!wal.exists() || std::fs::read(&wal).unwrap() != b"stale");
        integrity(&read_only(&s.path).unwrap()).unwrap();
    }

    #[test]
    fn stamping_drops_deletions_of_tags_in_use() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("l.sqlite3");
        drop(Db::open(&path).unwrap());
        let mut conn = Connection::open(&path).unwrap();
        conn.execute_batch(
            "INSERT INTO object (kind, key, version, data)
                 VALUES ('item', 'AAAAAAAA', 1, '{\"key\":\"AAAAAAAA\",\"version\":1}');
             INSERT INTO item_tag (item_key, tag) VALUES ('AAAAAAAA', 'kept');
             INSERT INTO deletion (kind, key, version) VALUES ('tag', 'kept', 1), ('tag', 'gone', 1);",
        )
        .unwrap();
        stamp(&mut conn, 1001).unwrap();
        let tags: Vec<String> = conn
            .prepare("SELECT key FROM deletion WHERE kind = 'tag'")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(tags, ["gone"]);
    }

    #[test]
    fn a_held_directory_cannot_be_locked_again() {
        let dir = tempfile::tempdir().unwrap();
        let held = lock(dir.path()).unwrap();
        assert!(lock(dir.path()).is_err());
        drop(held);
        assert!(lock(dir.path()).is_ok());
    }
}