//! Undo/redo snapshot ring (ADR-0006 Amendment 1).
//!
//! A *snapshot* is a whole-project copy taken before a mutation:
//! the database (via SQLite's online backup API) plus the
//! authoritative text (`project.yaml` + `data/*.csv`, copied as
//! files). Undo restores all three directly, re-establishing a
//! consistent `(db, yaml, csv)` triple (ADR-0015). Storing both the
//! database and its text projection means undo is a direct restore —
//! no rebuild step.
//!
//! On-disk layout under `<project>/.snapshots/`:
//!
//! ```text
//! index.yaml   — the ordered undo ring + redo stack (source of truth)
//! <id>/        — one payload per snapshot: playground.db, project.yaml,
//!                data/*.csv
//! .staging/    — a snapshot being assembled (only one at a time; the
//!                db worker is single-threaded)
//! ```
//!
//! **Payload semantics.** Each undo entry's payload is the state that
//! existed *before* its command ran; each redo entry's payload is the
//! state *after* its command ran (the state undo moved away from).
//! Undo pops the newest undo entry, snapshots the current state onto
//! the redo stack, then restores the popped payload. Redo is the
//! mirror. New work (a fresh `finalize`) clears the redo stack
//! (ADR-0006 Amendment 1: redo discarded on new work).
//!
//! **Worker-only.** Every method that touches the database takes the
//! live `&Connection` / `&mut Connection` owned by the db worker, so
//! all live-database access stays on the worker thread (the `db.rs`
//! invariant). `backup` / `restore` go through that connection and
//! snapshot *files*; this module never opens the live database file
//! directly.

use std::fs;
use std::io::Write as _;
use std::path::{Path, PathBuf};

use rusqlite::{Connection, MAIN_DB};
use serde::{Deserialize, Serialize};

use crate::persistence::utc_iso8601_now;
use crate::project::{DATA_DIR, PLAYGROUND_DB, PROJECT_YAML};

/// Default undo ring depth (ADR-0006 Amendment 1: N = 50, raised
/// from the original ADR's N = 10 because single-step undo counts
/// commands). A single tunable constant.
pub const DEFAULT_RING_CAPACITY: usize = 50;

/// Directory under the project that holds the ring.
///
/// Public so the `.gitignore` template, the export exclusion, and
/// the temp-project cleanup allowlist all reference the one
/// canonical name (ADR-0006 Amendment 1).
pub const SNAPSHOTS_DIR: &str = ".snapshots";
/// In-flight snapshot directory (one at a time).
const STAGING_DIR: &str = ".staging";
/// The ring index file.
const INDEX_FILE: &str = "index.yaml";

#[derive(Debug)]
pub enum SnapshotError {
    Io {
        op: &'static str,
        path: PathBuf,
        source: std::io::Error,
    },
    Db(rusqlite::Error),
    Index {
        message: String,
    },
}

impl std::fmt::Display for SnapshotError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Io { op, path, source } => {
                write!(f, "snapshot {op} failed for {}: {source}", path.display())
            }
            Self::Db(e) => write!(f, "snapshot database operation failed: {e}"),
            Self::Index { message } => write!(f, "snapshot index error: {message}"),
        }
    }
}

impl std::error::Error for SnapshotError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io { source, .. } => Some(source),
            Self::Db(e) => Some(e),
            Self::Index { .. } => None,
        }
    }
}

impl From<rusqlite::Error> for SnapshotError {
    fn from(e: rusqlite::Error) -> Self {
        Self::Db(e)
    }
}

type Result<T> = std::result::Result<T, SnapshotError>;

/// A staged-but-not-committed snapshot. Held by the worker between
/// `stage` and `finalize`/`discard`; for a batch command the worker
/// holds one of these across the whole batch.
#[derive(Debug, Clone)]
pub struct Staged {
    command: String,
    timestamp: String,
}

/// What `peek_*` / `undo` / `redo` report back to the caller — enough
/// to build the confirmation prompt ("Undo `<command>` (run <when>)?").
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SnapshotMeta {
    pub id: u64,
    pub timestamp: String,
    pub command: String,
}

#[derive(Clone, Serialize, Deserialize)]
struct Entry {
    id: u64,
    timestamp: String,
    command: String,
}

impl Entry {
    fn meta(&self) -> SnapshotMeta {
        SnapshotMeta {
            id: self.id,
            timestamp: self.timestamp.clone(),
            command: self.command.clone(),
        }
    }
}

#[derive(Default, Serialize, Deserialize)]
struct Index {
    /// Monotonic id allocator; never reuses an id.
    next_id: u64,
    /// Undo ring; the *last* element is the most recent (top).
    undo: Vec<Entry>,
    /// Redo stack; the *last* element is the most recently undone.
    redo: Vec<Entry>,
}

/// The snapshot ring for one project. Cheap to construct; the worker
/// holds one for the project's lifetime (or `None` when `--no-undo`).
#[derive(Debug, Clone)]
pub struct SnapshotStore {
    project_dir: PathBuf,
    root: PathBuf,
    cap: usize,
}

impl SnapshotStore {
    #[must_use]
    pub fn new(project_dir: &Path, cap: usize) -> Self {
        Self {
            project_dir: project_dir.to_path_buf(),
            root: project_dir.join(SNAPSHOTS_DIR),
            cap,
        }
    }

    // ---- public API ----------------------------------------------------

    /// Stage a snapshot of the *current* committed state (database +
    /// text) into `.staging/`. Call before a mutation's transaction
    /// opens; `finalize` commits it into the ring after the db commit,
    /// or `discard` drops it if the mutation rolled back.
    pub fn stage(&self, live: &Connection, command: &str) -> Result<Staged> {
        let staging = self.staging_dir();
        remove_dir_all_if_exists(&staging)?;
        tracing::debug!(command, dir = %staging.display(), "staging undo snapshot");
        self.snapshot_into(live, &staging)?;
        Ok(Staged {
            command: command.to_string(),
            timestamp: utc_iso8601_now(),
        })
    }

    /// Commit a staged snapshot into the undo ring: rename `.staging/`
    /// to a fresh `<id>/`, push the entry, **clear the redo stack**
    /// (new work) and evict the oldest beyond the cap. Returns the new
    /// undo entry's metadata.
    pub fn finalize(&self, staged: Staged) -> Result<SnapshotMeta> {
        let mut index = self.load_index()?;
        let id = index.next_id;
        index.next_id += 1;

        let dest = self.payload_dir(id);
        rename(&self.staging_dir(), &dest)?;

        let entry = Entry {
            id,
            timestamp: staged.timestamp,
            command: staged.command,
        };
        let meta = entry.meta();
        index.undo.push(entry);

        // New work invalidates redo (ADR-0006 Amendment 1).
        for r in std::mem::take(&mut index.redo) {
            remove_dir_all_if_exists(&self.payload_dir(r.id))?;
        }

        // Evict oldest beyond the cap.
        while index.undo.len() > self.cap {
            let old = index.undo.remove(0);
            tracing::debug!(id = old.id, "evicting oldest undo snapshot (ring full)");
            remove_dir_all_if_exists(&self.payload_dir(old.id))?;
        }

        self.save_index(&index)?;
        tracing::info!(id = meta.id, command = %meta.command, "undo snapshot recorded");
        Ok(meta)
    }

    /// Drop a staged snapshot (the mutation rolled back, so there is
    /// nothing to undo).
    pub fn discard(&self, _staged: Staged) -> Result<()> {
        tracing::debug!("discarding staged undo snapshot (op did not commit)");
        remove_dir_all_if_exists(&self.staging_dir())
    }

    /// Metadata of the snapshot `undo` would restore, without
    /// restoring it. `None` when the ring is empty.
    pub fn peek_undo(&self) -> Result<Option<SnapshotMeta>> {
        Ok(self.load_index()?.undo.last().map(Entry::meta))
    }

    /// Metadata of the snapshot `redo` would restore. `None` when the
    /// redo stack is empty.
    pub fn peek_redo(&self) -> Result<Option<SnapshotMeta>> {
        Ok(self.load_index()?.redo.last().map(Entry::meta))
    }

    /// Restore the most recent snapshot: snapshot the current state
    /// onto the redo stack (so redo is possible), then restore the
    /// popped payload (text first, database last — ADR-0015 §6
    /// commit-db-last). Returns the metadata of the command that was
    /// undone, or `None` if there is nothing to undo.
    pub fn undo(&self, live: &mut Connection) -> Result<Option<SnapshotMeta>> {
        let mut index = self.load_index()?;
        let Some(entry) = index.undo.pop() else {
            return Ok(None);
        };
        // Snapshot current ("after entry.command") onto redo.
        let redo_id = index.next_id;
        index.next_id += 1;
        self.snapshot_into(live, &self.payload_dir(redo_id))?;
        index.redo.push(Entry {
            id: redo_id,
            timestamp: utc_iso8601_now(),
            command: entry.command.clone(),
        });
        // Restore the popped payload, then persist the index, then
        // delete the consumed payload (last so a failure there leaves
        // the index consistent).
        self.restore_from(live, &self.payload_dir(entry.id))?;
        self.save_index(&index)?;
        remove_dir_all_if_exists(&self.payload_dir(entry.id))?;
        tracing::info!(command = %entry.command, "undo applied");
        Ok(Some(entry.meta()))
    }

    /// Restore the most recently undone snapshot: snapshot the current
    /// state onto the undo ring, then restore the redo payload.
    /// Returns the metadata of the command that was re-applied, or
    /// `None` if there is nothing to redo. Does **not** clear redo or
    /// evict — only new work does that.
    pub fn redo(&self, live: &mut Connection) -> Result<Option<SnapshotMeta>> {
        let mut index = self.load_index()?;
        let Some(entry) = index.redo.pop() else {
            return Ok(None);
        };
        // Snapshot current ("before entry.command") back onto undo.
        let undo_id = index.next_id;
        index.next_id += 1;
        self.snapshot_into(live, &self.payload_dir(undo_id))?;
        index.undo.push(Entry {
            id: undo_id,
            timestamp: utc_iso8601_now(),
            command: entry.command.clone(),
        });
        self.restore_from(live, &self.payload_dir(entry.id))?;
        self.save_index(&index)?;
        remove_dir_all_if_exists(&self.payload_dir(entry.id))?;
        tracing::info!(command = %entry.command, "redo applied");
        Ok(Some(entry.meta()))
    }

    /// Clear the redo stack (deleting its payloads) without touching
    /// the undo ring. Called when new work commits but its snapshot
    /// could not be staged: `finalize` (which normally clears redo)
    /// never runs, so the redo entries are left stale — a later
    /// `redo` would restore an outdated state and silently discard
    /// the new work. Clearing redo here closes that data-loss hole
    /// (ADR-0006 Amendment 1; the snapshot-failure policy is
    /// non-fatal, so this keeps it *safe*).
    pub fn clear_redo(&self) -> Result<()> {
        let mut index = self.load_index()?;
        if index.redo.is_empty() {
            return Ok(());
        }
        for r in std::mem::take(&mut index.redo) {
            remove_dir_all_if_exists(&self.payload_dir(r.id))?;
        }
        self.save_index(&index)?;
        tracing::debug!("redo stack cleared (new work committed without a snapshot)");
        Ok(())
    }

    /// Remove crash leftovers on project open: the `.staging/` dir and
    /// any payload dir not referenced by the index.
    pub fn cleanup(&self) -> Result<()> {
        remove_dir_all_if_exists(&self.staging_dir())?;
        if !self.root.exists() {
            return Ok(());
        }
        let index = self.load_index()?;
        let referenced: std::collections::HashSet<u64> = index
            .undo
            .iter()
            .chain(index.redo.iter())
            .map(|e| e.id)
            .collect();
        for entry in read_dir(&self.root)? {
            let entry = entry.map_err(|e| io_err("read_dir entry", &self.root, e))?;
            let name = entry.file_name();
            let Some(name) = name.to_str() else { continue };
            if let Ok(id) = name.parse::<u64>()
                && !referenced.contains(&id)
            {
                tracing::debug!(id, "removing orphan snapshot payload");
                remove_dir_all_if_exists(&self.payload_dir(id))?;
            }
        }
        Ok(())
    }

    // ---- internals -----------------------------------------------------

    fn staging_dir(&self) -> PathBuf {
        self.root.join(STAGING_DIR)
    }

    fn payload_dir(&self, id: u64) -> PathBuf {
        self.root.join(id.to_string())
    }

    fn index_path(&self) -> PathBuf {
        self.root.join(INDEX_FILE)
    }

    /// Capture the current state (database + text) into `dir`.
    fn snapshot_into(&self, live: &Connection, dir: &Path) -> Result<()> {
        create_dir_all(dir)?;
        live.backup(MAIN_DB, dir.join(PLAYGROUND_DB), None)?;
        copy_if_exists(
            &self.project_dir.join(PROJECT_YAML),
            &dir.join(PROJECT_YAML),
        )?;
        self.copy_data_dir(&self.project_dir.join(DATA_DIR), &dir.join(DATA_DIR))?;
        Ok(())
    }

    /// Restore the project state from a payload `dir`: text first, then
    /// the database (commit-db-last, ADR-0015 §6).
    fn restore_from(&self, live: &mut Connection, dir: &Path) -> Result<()> {
        // Text: project.yaml (atomic replace) + a wholesale data/ swap.
        let yaml_src = dir.join(PROJECT_YAML);
        if yaml_src.exists() {
            atomic_replace(&yaml_src, &self.project_dir.join(PROJECT_YAML))?;
        }
        self.replace_data_dir(&dir.join(DATA_DIR), &self.project_dir.join(DATA_DIR))?;
        // Database last.
        live.restore(MAIN_DB, dir.join(PLAYGROUND_DB), NO_PROGRESS)?;
        Ok(())
    }

    /// Copy a flat `data/` directory (`*.csv`, no subdirs per ADR-0015)
    /// into `dst`. Tolerates a missing source (empty project).
    fn copy_data_dir(&self, src: &Path, dst: &Path) -> Result<()> {
        if !src.exists() {
            return Ok(());
        }
        create_dir_all(dst)?;
        for entry in read_dir(src)? {
            let entry = entry.map_err(|e| io_err("read_dir entry", src, e))?;
            let path = entry.path();
            if path.is_file() {
                let name = entry.file_name();
                copy_if_exists(&path, &dst.join(name))?;
            }
        }
        Ok(())
    }

    /// Make the project's `data/` match the snapshot's exactly: clear
    /// the live data dir, then copy the snapshot's files in.
    fn replace_data_dir(&self, src: &Path, dst: &Path) -> Result<()> {
        remove_dir_all_if_exists(dst)?;
        self.copy_data_dir(src, dst)
    }

    fn load_index(&self) -> Result<Index> {
        let path = self.index_path();
        let mut index = if path.exists() {
            let body = fs::read_to_string(&path).map_err(|e| io_err("read index", &path, e))?;
            serde_norway::from_str(&body).map_err(|e| SnapshotError::Index {
                message: e.to_string(),
            })?
        } else {
            Index::default()
        };
        // Reconcile next_id against any payload dirs present, so a crash
        // that left an orphan can never cause an id to be reused.
        if self.root.exists() {
            let mut max_seen = index.next_id;
            for entry in read_dir(&self.root)? {
                let entry = entry.map_err(|e| io_err("read_dir entry", &self.root, e))?;
                if let Some(name) = entry.file_name().to_str()
                    && let Ok(id) = name.parse::<u64>()
                {
                    max_seen = max_seen.max(id + 1);
                }
            }
            index.next_id = max_seen;
        }
        Ok(index)
    }

    fn save_index(&self, index: &Index) -> Result<()> {
        create_dir_all(&self.root)?;
        let body = serde_norway::to_string(index).map_err(|e| SnapshotError::Index {
            message: e.to_string(),
        })?;
        let path = self.index_path();
        let tmp = path.with_extension("yaml.tmp");
        write_fsync(&tmp, body.as_bytes())?;
        rename(&tmp, &path)?;
        Ok(())
    }
}

/// Restore's progress callback type, fixed to `None`.
const NO_PROGRESS: Option<fn(rusqlite::backup::Progress)> = None;

// ---- small fs helpers (own error context) -----------------------------

fn io_err(op: &'static str, path: &Path, source: std::io::Error) -> SnapshotError {
    SnapshotError::Io {
        op,
        path: path.to_path_buf(),
        source,
    }
}

fn create_dir_all(dir: &Path) -> Result<()> {
    fs::create_dir_all(dir).map_err(|e| io_err("create_dir_all", dir, e))
}

fn read_dir(dir: &Path) -> Result<fs::ReadDir> {
    fs::read_dir(dir).map_err(|e| io_err("read_dir", dir, e))
}

fn rename(from: &Path, to: &Path) -> Result<()> {
    fs::rename(from, to).map_err(|e| io_err("rename", to, e))
}

fn remove_dir_all_if_exists(dir: &Path) -> Result<()> {
    if dir.exists() {
        fs::remove_dir_all(dir).map_err(|e| io_err("remove_dir_all", dir, e))?;
    }
    Ok(())
}

fn copy_if_exists(src: &Path, dst: &Path) -> Result<()> {
    if src.exists() {
        if let Some(parent) = dst.parent() {
            create_dir_all(parent)?;
        }
        fs::copy(src, dst).map_err(|e| io_err("copy", src, e))?;
    }
    Ok(())
}

/// Replace `dst` with `src` atomically via copy-to-temp + rename.
fn atomic_replace(src: &Path, dst: &Path) -> Result<()> {
    let body = fs::read(src).map_err(|e| io_err("read", src, e))?;
    let tmp = dst.with_extension("restore.tmp");
    write_fsync(&tmp, &body)?;
    rename(&tmp, dst)
}

fn write_fsync(path: &Path, bytes: &[u8]) -> Result<()> {
    if let Some(parent) = path.parent() {
        create_dir_all(parent)?;
    }
    let mut f = fs::File::create(path).map_err(|e| io_err("create", path, e))?;
    f.write_all(bytes).map_err(|e| io_err("write", path, e))?;
    f.sync_all().map_err(|e| io_err("fsync", path, e))?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusqlite::Connection;
    use tempfile::TempDir;

    /// A test project: a temp dir with a file-backed db connection and
    /// helpers to read/write the text targets the store snapshots.
    struct Fixture {
        _dir: TempDir,
        path: PathBuf,
        conn: Connection,
    }

    impl Fixture {
        fn new() -> Self {
            let dir = TempDir::new().unwrap();
            let path = dir.path().to_path_buf();
            let conn = Connection::open(path.join(PLAYGROUND_DB)).unwrap();
            conn.execute_batch("CREATE TABLE t (n INTEGER);").unwrap();
            Self {
                _dir: dir,
                path,
                conn,
            }
        }

        fn store(&self) -> SnapshotStore {
            SnapshotStore::new(&self.path, DEFAULT_RING_CAPACITY)
        }

        fn store_cap(&self, cap: usize) -> SnapshotStore {
            SnapshotStore::new(&self.path, cap)
        }

        fn insert(&self, n: i64) {
            self.conn
                .execute("INSERT INTO t (n) VALUES (?1)", [n])
                .unwrap();
        }

        fn rows(&self) -> Vec<i64> {
            let mut stmt = self.conn.prepare("SELECT n FROM t ORDER BY n").unwrap();
            stmt.query_map([], |r| r.get::<_, i64>(0))
                .unwrap()
                .map(std::result::Result::unwrap)
                .collect()
        }

        fn write_yaml(&self, body: &str) {
            fs::write(self.path.join(PROJECT_YAML), body).unwrap();
        }

        fn read_yaml(&self) -> String {
            fs::read_to_string(self.path.join(PROJECT_YAML)).unwrap()
        }

        fn write_csv(&self, table: &str, body: &str) {
            let data = self.path.join(DATA_DIR);
            fs::create_dir_all(&data).unwrap();
            fs::write(data.join(format!("{table}.csv")), body).unwrap();
        }

        fn csv_exists(&self, table: &str) -> bool {
            self.path
                .join(DATA_DIR)
                .join(format!("{table}.csv"))
                .exists()
        }
    }

    fn stage_finalize(store: &SnapshotStore, conn: &Connection, command: &str) {
        let staged = store.stage(conn, command).unwrap();
        store.finalize(staged).unwrap();
    }

    #[test]
    fn stage_then_finalize_records_undo_entry() {
        let fx = Fixture::new();
        let store = fx.store();
        assert!(store.peek_undo().unwrap().is_none());

        stage_finalize(&store, &fx.conn, "delete from t where n=1");

        let meta = store.peek_undo().unwrap().expect("an undo entry");
        assert_eq!(meta.command, "delete from t where n=1");
        assert!(!meta.timestamp.is_empty());
    }

    #[test]
    fn discard_leaves_no_undo_entry() {
        let fx = Fixture::new();
        let store = fx.store();
        let staged = store.stage(&fx.conn, "drop table t").unwrap();
        store.discard(staged).unwrap();
        assert!(store.peek_undo().unwrap().is_none());
        assert!(!store.staging_dir().exists());
    }

    #[test]
    fn undo_restores_database_state() {
        let fx = Fixture::new();
        let store = fx.store();
        fx.insert(1);
        // Snapshot the pre-mutation state, then mutate.
        let staged = store.stage(&fx.conn, "insert into t values (2)").unwrap();
        store.finalize(staged).unwrap();
        fx.insert(2);
        assert_eq!(fx.rows(), vec![1, 2]);

        let mut conn = fx.conn;
        let undone = store.undo(&mut conn).unwrap().expect("something to undo");
        assert_eq!(undone.command, "insert into t values (2)");
        // Re-read through a fresh connection to be sure it is the file
        // that changed, not just connection cache.
        let check = Connection::open(fx.path.join(PLAYGROUND_DB)).unwrap();
        let n: Vec<i64> = check
            .prepare("SELECT n FROM t ORDER BY n")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .map(std::result::Result::unwrap)
            .collect();
        assert_eq!(n, vec![1], "row 2 should be undone");
    }

    #[test]
    fn redo_reapplies_after_undo() {
        let fx = Fixture::new();
        let store = fx.store();
        fx.insert(1);
        stage_finalize(&store, &fx.conn, "insert into t values (2)");
        fx.insert(2);

        let mut conn = fx.conn;
        store.undo(&mut conn).unwrap();
        // restore the connection's view by reopening through the fixture
        // (the in-place restore mutated the same file/connection).
        let after_undo: Vec<i64> = conn
            .prepare("SELECT n FROM t ORDER BY n")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .map(std::result::Result::unwrap)
            .collect();
        assert_eq!(after_undo, vec![1]);

        let redone = store.redo(&mut conn).unwrap().expect("something to redo");
        assert_eq!(redone.command, "insert into t values (2)");
        let after_redo: Vec<i64> = conn
            .prepare("SELECT n FROM t ORDER BY n")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .map(std::result::Result::unwrap)
            .collect();
        assert_eq!(after_redo, vec![1, 2], "redo should re-apply the insert");
    }

    #[test]
    fn undo_restores_text_files() {
        let mut fx = Fixture::new();
        let store = fx.store();
        fx.write_yaml("version: 1\nA\n");
        fx.write_csv("t", "n\n1\n");

        stage_finalize(&store, &fx.conn, "update t set n=9 --all-rows");
        fx.write_yaml("version: 1\nB\n");
        fx.write_csv("t", "n\n9\n");

        store.undo(&mut fx.conn).unwrap();
        assert_eq!(fx.read_yaml(), "version: 1\nA\n");
        assert_eq!(
            fs::read_to_string(fx.path.join(DATA_DIR).join("t.csv")).unwrap(),
            "n\n1\n"
        );
    }

    #[test]
    fn undo_removes_csv_for_table_created_after_snapshot() {
        let mut fx = Fixture::new();
        let store = fx.store();
        fx.write_csv("t", "n\n1\n");
        // Snapshot, then a later command "creates" another table's CSV.
        stage_finalize(&store, &fx.conn, "create table u with pk id(serial)");
        fx.write_csv("u", "id\n1\n");
        assert!(fx.csv_exists("u"));

        store.undo(&mut fx.conn).unwrap();
        assert!(fx.csv_exists("t"), "t existed at the snapshot");
        assert!(
            !fx.csv_exists("u"),
            "u was created after the snapshot and must be removed on undo"
        );
    }

    #[test]
    fn new_work_clears_redo() {
        let fx = Fixture::new();
        let store = fx.store();
        stage_finalize(&store, &fx.conn, "delete from t --all-rows");

        let mut conn = fx.conn;
        store.undo(&mut conn).unwrap();
        assert!(store.peek_redo().unwrap().is_some(), "redo available");

        // New work: stage + finalize again.
        let staged = store.stage(&conn, "drop table t").unwrap();
        store.finalize(staged).unwrap();
        assert!(
            store.peek_redo().unwrap().is_none(),
            "new work discards the redo stack"
        );
    }

    #[test]
    fn eviction_keeps_only_cap_entries() {
        let fx = Fixture::new();
        let store = fx.store_cap(2);
        stage_finalize(&store, &fx.conn, "cmd 1");
        stage_finalize(&store, &fx.conn, "cmd 2");
        stage_finalize(&store, &fx.conn, "cmd 3");

        // Top is cmd 3; only 2 payloads survive.
        let meta = store.peek_undo().unwrap().unwrap();
        assert_eq!(meta.command, "cmd 3");
        let payload_dirs = fs::read_dir(&store.root)
            .unwrap()
            .filter_map(std::result::Result::ok)
            .filter(|e| e.file_name().to_str().is_some_and(|n| n.parse::<u64>().is_ok()))
            .count();
        assert_eq!(payload_dirs, 2, "ring capped at 2 payloads");
    }

    #[test]
    fn clear_redo_drops_redo_without_touching_undo() {
        let fx = Fixture::new();
        let store = fx.store();
        // Two entries so one survives in the undo ring after undo.
        stage_finalize(&store, &fx.conn, "cmd 1");
        stage_finalize(&store, &fx.conn, "cmd 2");
        let mut conn = fx.conn;
        store.undo(&mut conn).unwrap(); // pops cmd 2, redo gets it
        assert!(store.peek_redo().unwrap().is_some(), "redo available");
        assert_eq!(store.peek_undo().unwrap().unwrap().command, "cmd 1");

        store.clear_redo().unwrap();
        assert!(store.peek_redo().unwrap().is_none(), "redo cleared");
        // The remaining undo entry is untouched.
        assert_eq!(
            store.peek_undo().unwrap().unwrap().command,
            "cmd 1",
            "undo ring intact"
        );
        // The redo payload is gone; only the surviving undo payload remains.
        let payload_dirs = fs::read_dir(&store.root)
            .unwrap()
            .filter_map(std::result::Result::ok)
            .filter(|e| e.file_name().to_str().is_some_and(|n| n.parse::<u64>().is_ok()))
            .count();
        assert_eq!(payload_dirs, 1, "only the surviving undo payload remains");
    }

    #[test]
    fn empty_stacks_return_none() {
        let fx = Fixture::new();
        let store = fx.store();
        let mut conn = fx.conn;
        assert!(store.undo(&mut conn).unwrap().is_none());
        assert!(store.redo(&mut conn).unwrap().is_none());
        assert!(store.peek_undo().unwrap().is_none());
        assert!(store.peek_redo().unwrap().is_none());
    }

    #[test]
    fn peek_does_not_consume() {
        let fx = Fixture::new();
        let store = fx.store();
        stage_finalize(&store, &fx.conn, "delete from t --all-rows");
        let a = store.peek_undo().unwrap();
        let b = store.peek_undo().unwrap();
        assert_eq!(a, b);
        assert!(a.is_some());
    }

    #[test]
    fn cleanup_removes_orphan_payloads_and_staging() {
        let fx = Fixture::new();
        let store = fx.store();
        stage_finalize(&store, &fx.conn, "keep me");

        // Fabricate an orphan payload dir and a leftover staging dir.
        fs::create_dir_all(store.payload_dir(999)).unwrap();
        fs::create_dir_all(store.staging_dir()).unwrap();

        store.cleanup().unwrap();
        assert!(!store.payload_dir(999).exists(), "orphan removed");
        assert!(!store.staging_dir().exists(), "staging removed");
        assert!(
            store.peek_undo().unwrap().is_some(),
            "referenced snapshot retained"
        );
    }

    #[test]
    fn next_id_is_not_reused_after_orphan() {
        let fx = Fixture::new();
        let store = fx.store();
        // An orphan dir with a high id must push next_id past it.
        fs::create_dir_all(store.payload_dir(41)).unwrap();
        stage_finalize(&store, &fx.conn, "cmd");
        let meta = store.peek_undo().unwrap().unwrap();
        assert!(meta.id >= 42, "id allocated above the orphan, got {}", meta.id);
    }
}