recall-server 0.4.9

//! `recall-server admin`, driven as the owner drives it: the real binary,
//! against a real database file, sometimes with a real server serving that
//! same file.
//!
//! These edit the only copy of someone's memory, so most of what is here is
//! about what must *not* happen: a change nobody confirmed, a change with no
//! backup behind it, a partial change, a change to a key that was only
//! nearly named.

use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::path::{Path, PathBuf};
use std::process::{Command, Output, Stdio};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use axum::body::Body;
use axum::http::{Request, StatusCode};
use recall_server::audit::leaf;
use recall_server::merge::Status;
use recall_server::store::{Queued, Settled, Settlement};
use recall_server::{Config, Server, Store};
use recall_wire::{MergeResult, MergeSide, ResultRequest};
use rusqlite::Connection;
use serde_json::{json, Value};
use tempfile::TempDir;
use time::OffsetDateTime;
use tower::ServiceExt;

const OLD: &str = "local:-Users-me-thing";
const T0: &str = "2026-09-20T10:00:00.000Z";
const T1: &str = "2026-09-21T10:00:00.000Z";
const TOKEN: &str = "admin-test-token";

/// Every column of a row: key, path, content, source, updated_at, deleted.
type DumpRow = (String, String, String, Option<String>, String, i64);

/// Every row, so "nothing changed" is checked on every column rather than
/// on a count.
type Dump = Vec<DumpRow>;

struct Fixture {
    dir: TempDir,
}

impl Fixture {
    /// Three keys: the one a machine with no git remote wrote (a file, one
    /// with no source, a tombstone), the key it should have used, and an
    /// unrelated project that must never be touched.
    fn new() -> Self {
        let dir = tempfile::tempdir().unwrap();
        // The server creates the schema; the rows go in by hand, since
        // their exact timestamps are what the listings below are checked
        // against, and the store stamps what it writes with the time.
        drop(Store::open(dir.path().join("recall.db")).unwrap());
        let conn = Connection::open(dir.path().join("recall.db")).unwrap();
        conn.execute_batch(&format!(
            "INSERT INTO memory_files VALUES ('{OLD}', 'MEMORY.md', '- [notes](notes.md)\n', 'laptop', '{T0}', 0);
             INSERT INTO memory_files VALUES ('{OLD}', 'notes.md', 'a fact\n', NULL, '{T1}', 0);
             INSERT INTO memory_files VALUES ('{OLD}', 'old.md', 'kept after delete\n', 'laptop', '{T1}', 1);
             INSERT INTO memory_files VALUES ('me/thing', 'MEMORY.md', '- other\n', 'cloud', '{T1}', 0);
             INSERT INTO memory_files VALUES ('acme/app', 'unrelated.md', 'leave me\n', 'cloud', '{T0}', 0);"
        ))
        .unwrap();
        Self { dir }
    }

    fn db(&self) -> PathBuf {
        self.dir.path().join("recall.db")
    }

    fn backups(&self) -> PathBuf {
        self.dir.path().join("backups")
    }

    fn command(&self, args: &[&str]) -> Command {
        let mut cmd = Command::new(env!("CARGO_BIN_EXE_recall-server"));
        cmd.arg("admin")
            .args(args)
            // No RECALL_TOKEN on purpose: an admin command needs none.
            .env_clear()
            .env("RECALL_DB_PATH", self.db())
            .env("RECALL_BACKUP_DIR", self.backups())
            // Every change waits out the server's merge window after it
            // commits. A millisecond here leaves the fixed margin, about a
            // second, instead of the 45 a real server's default costs; the
            // race test sets its own.
            .env("RECALL_MERGE_TIMEOUT_MS", "1");
        cmd
    }

    fn admin(&self, args: &[&str]) -> Output {
        self.command(args)
            .stdin(Stdio::null())
            .output()
            .expect("recall-server runs")
    }

    /// Runs with `typed` on stdin, as if typed at the confirmation prompt.
    fn admin_typing(&self, args: &[&str], typed: &str) -> Output {
        let mut child = self
            .command(args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();
        child
            .stdin
            .take()
            .unwrap()
            .write_all(typed.as_bytes())
            .unwrap();
        child.wait_with_output().unwrap()
    }

    fn dump(&self) -> Dump {
        dump(&self.db())
    }

    /// The backups admin commands took, oldest first.
    fn snapshots(&self) -> Vec<PathBuf> {
        let Ok(entries) = std::fs::read_dir(self.backups().join("admin")) else {
            return Vec::new();
        };
        let mut found: Vec<PathBuf> = entries.map(|e| e.unwrap().path()).collect();
        found.sort();
        found
    }

    fn sql(&self, sql: &str) {
        Connection::open(self.db())
            .unwrap()
            .execute_batch(sql)
            .unwrap();
    }
}

fn dump(db: &Path) -> Dump {
    let conn = Connection::open(db).unwrap();
    let mut stmt = conn
        .prepare(
            "SELECT project_key, file_path, content, source_env, updated_at, deleted
             FROM memory_files ORDER BY project_key, file_path",
        )
        .unwrap();
    let rows = stmt
        .query_map([], |r| {
            Ok((
                r.get(0)?,
                r.get(1)?,
                r.get(2)?,
                r.get(3)?,
                r.get(4)?,
                r.get(5)?,
            ))
        })
        .unwrap();
    rows.map(Result::unwrap).collect()
}

fn keys(dump: &Dump) -> Vec<&str> {
    let mut keys: Vec<&str> = dump.iter().map(|r| r.0.as_str()).collect();
    keys.dedup();
    keys
}

fn under<'a>(dump: &'a Dump, key: &str) -> Vec<&'a DumpRow> {
    dump.iter().filter(|r| r.0 == key).collect()
}

fn text(out: &Output) -> (String, String) {
    (
        String::from_utf8_lossy(&out.stdout).into_owned(),
        String::from_utf8_lossy(&out.stderr).into_owned(),
    )
}

#[track_caller]
fn assert_exit(out: &Output, code: i32) {
    let (stdout, stderr) = text(out);
    assert_eq!(
        out.status.code(),
        Some(code),
        "stdout:\n{stdout}\nstderr:\n{stderr}"
    );
}

// ---------------------------------------------------------------- list

#[test]
fn list_shows_every_key_with_its_counts_and_needs_no_token() {
    let fx = Fixture::new();
    let out = fx.admin(&["list"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    let line = |key: &str| {
        stdout
            .lines()
            .find(|l| l.ends_with(&format!("  {key}")))
            .unwrap_or_else(|| panic!("no line for {key} in:\n{stdout}"))
            .split_whitespace()
            .collect::<Vec<_>>()
    };
    // FILES, TOMBSTONES, LAST UPDATE, KEY
    assert_eq!(line(OLD), vec!["2", "1", T1, OLD]);
    assert_eq!(line("me/thing"), vec!["1", "0", T1, "me/thing"]);
    assert_eq!(line("acme/app"), vec!["1", "0", T0, "acme/app"]);
    assert!(stdout.contains("3 project key(s), 4 file(s), 1 tombstone(s)."));
}

#[test]
fn list_with_a_key_shows_its_files() {
    let fx = Fixture::new();
    let out = fx.admin(&["list", OLD]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    assert!(stdout.contains("2 file(s), 1 tombstone(s)"), "{stdout}");
    for path in ["MEMORY.md", "notes.md", "old.md"] {
        assert!(
            stdout.lines().any(|l| l.ends_with(path)),
            "{path}: {stdout}"
        );
    }
    assert!(
        stdout
            .lines()
            .any(|l| l.starts_with("tombstone") && l.ends_with("old.md")),
        "{stdout}"
    );
}

/// A mistyped path must not quietly become an empty database, which would
/// list as "no projects" to someone already worried about their memory.
#[test]
fn a_database_that_is_not_there_is_an_error_and_is_not_created() {
    let fx = Fixture::new();
    let missing = fx.dir.path().join("typo.db");
    let out = fx
        .command(&["list"])
        .env("RECALL_DB_PATH", &missing)
        .stdin(Stdio::null())
        .output()
        .unwrap();
    assert_exit(&out, 1);
    assert!(text(&out).1.contains("no database"), "{:?}", text(&out));
    assert!(!missing.exists());
}

/// The way to see what a backup holds before restoring from it.
#[test]
fn list_backup_shows_what_a_backup_holds() {
    let fx = Fixture::new();
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    fx.sql(&format!(
        "DELETE FROM memory_files WHERE project_key = '{OLD}'"
    ));
    let before = std::fs::read(&snapshot).unwrap();

    let out = fx.admin(&["list", "--backup", snapshot.to_str().unwrap()]);
    assert_exit(&out, 0);
    assert!(text(&out).0.contains(OLD), "{:?}", text(&out));
    let out = fx.admin(&["list", "--backup", snapshot.to_str().unwrap(), OLD]);
    assert_exit(&out, 0);
    assert!(text(&out).0.contains("notes.md"), "{:?}", text(&out));

    assert_eq!(
        std::fs::read(&snapshot).unwrap(),
        before,
        "a backup is read, never written"
    );
}

// ---------------------------------------------------------------- rename

#[test]
fn rename_moves_every_row_after_taking_a_backup() {
    let fx = Fixture::new();
    let before = fx.dump();
    let out = fx.admin(&["rename", OLD, "me/new", "--yes"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);

    let after = fx.dump();
    assert!(under(&after, OLD).is_empty());
    let moved: Vec<_> = under(&after, "me/new")
        .into_iter()
        .map(|r| (&r.1, &r.2, &r.3, &r.4, r.5))
        .collect();
    let original: Vec<_> = under(&before, OLD)
        .into_iter()
        .map(|r| (&r.1, &r.2, &r.3, &r.4, r.5))
        .collect();
    assert_eq!(
        moved, original,
        "every column of every row survives the move"
    );
    assert_eq!(under(&after, "acme/app"), under(&before, "acme/app"));

    // The backup is printed, and it is the state *before* the change: the
    // only way it can be is if it was taken first.
    let snapshots = fx.snapshots();
    assert_eq!(snapshots.len(), 1, "{stdout}");
    let name = snapshots[0].file_name().unwrap().to_str().unwrap();
    assert!(
        stdout.contains(name),
        "the backup's path is printed: {stdout}"
    );
    assert_eq!(dump(&snapshots[0]), before);
    assert!(
        stdout.contains(&format!(
            "Confirmed with --yes for project key to rename \"{OLD}\", and key to rename it to \
             \"me/new\"."
        )),
        "{stdout}"
    );
    assert!(stdout.contains("Checked: the change held."), "{stdout}");
}

/// A typo in the target is the costlier one: it strands the project under
/// a key no machine asks for. So the target is typed back too.
#[test]
fn a_rename_goes_ahead_only_when_both_keys_are_typed_back() {
    let fx = Fixture::new();
    let before = fx.dump();
    for typed in [
        format!("{OLD}\n"),
        format!("{OLD}\nme/nwe\n"),
        format!("{OLD}\nme/new \n"),
        "me/new\nme/new\n".to_string(),
    ] {
        let out = fx.admin_typing(&["rename", OLD, "me/new"], &typed);
        assert_exit(&out, 1);
        let (stdout, _) = text(&out);
        assert_eq!(
            stdout.contains("Then type the key to rename it to, \"me/new\""),
            typed.starts_with(OLD),
            "asked for the target once the source was right: {stdout}"
        );
    }
    let out = fx.admin_typing(&["rename", OLD, "me/new"], &format!("{OLD}\nme/nwe\n"));
    assert!(
        text(&out).1.contains("\"me/nwe\" is not \"me/new\""),
        "{:?}",
        text(&out)
    );
    assert_eq!(fx.dump(), before);
    assert!(fx.snapshots().is_empty(), "no backup until it is confirmed");

    let out = fx.admin_typing(&["rename", OLD, "me/new"], &format!("{OLD}\nme/new\n"));
    assert_exit(&out, 0);
    assert_eq!(under(&fx.dump(), "me/new").len(), 3);
}

/// The primary key is (project_key, file_path). Refused whether or not the
/// paths overlap, with `--yes`, and before anything is written, a backup
/// included.
#[test]
fn a_rename_onto_a_key_that_holds_rows_is_refused_and_changes_nothing() {
    let fx = Fixture::new();
    let before = fx.dump();
    // Overlapping: both have MEMORY.md.
    let out = fx.admin(&["rename", OLD, "me/thing", "--yes"]);
    assert_exit(&out, 1);
    let (_, stderr) = text(&out);
    assert!(stderr.contains("already holds 1 row(s)"), "{stderr}");
    assert!(stderr.contains("The database was not changed."), "{stderr}");
    // The advice is the safe fold, not "point the machine at it and let it
    // push", whose first pull overwrites that machine's files.
    assert!(stderr.contains("Do not fold"), "{stderr}");
    assert!(stderr.contains("memory directory aside first"), "{stderr}");
    assert!(stderr.contains(&format!("\"{OLD}.archived-20")), "{stderr}");
    assert!(!stderr.contains("then remove"), "{stderr}");
    // Disjoint: acme/app has only unrelated.md. Still refused.
    let out = fx.admin(&["rename", OLD, "acme/app", "--yes"]);
    assert_exit(&out, 1);

    assert_eq!(fx.dump(), before);
    assert!(
        fx.snapshots().is_empty(),
        "a refused change takes no backup"
    );
}

#[test]
fn a_rename_onto_a_key_no_client_could_use_is_refused() {
    let fx = Fixture::new();
    let before = fx.dump();
    for to in ["me/new ", "", "me/\tnew", OLD] {
        let out = fx.admin(&["rename", OLD, to, "--yes"]);
        assert_exit(&out, 1);
    }
    assert_eq!(fx.dump(), before);
}

// ---------------------------------------------------------------- remove

#[test]
fn remove_deletes_that_key_only_after_a_backup() {
    let fx = Fixture::new();
    let before = fx.dump();
    let out = fx.admin(&["remove", OLD, "--yes"]);
    assert_exit(&out, 0);

    let after = fx.dump();
    assert_eq!(keys(&after), vec!["acme/app", "me/thing"]);
    assert_eq!(under(&after, "me/thing"), under(&before, "me/thing"));
    let snapshots = fx.snapshots();
    assert_eq!(snapshots.len(), 1);
    assert_eq!(
        dump(&snapshots[0]),
        before,
        "the backup holds what was removed"
    );
}

/// Removing a key whose notes were meant to have been folded elsewhere is
/// the moment content is lost without anyone noticing. The plan says how
/// many files hold content no other key has, before the confirmation.
#[test]
fn remove_warns_before_confirming_about_content_no_other_key_has() {
    let fx = Fixture::new();
    // MEMORY.md and notes.md are nowhere else; old.md is a tombstone.
    let out = fx.admin_typing(&["remove", OLD], "no\n");
    assert_exit(&out, 1);
    let (stdout, _) = text(&out);
    let warning = stdout
        .find("Warning: 2 of these file(s) hold content that no live file under any other key")
        .unwrap_or_else(|| panic!("{stdout}"));
    let prompt = stdout.find("To go ahead").unwrap();
    assert!(warning < prompt, "warned before asking: {stdout}");
    assert!(stdout.contains(&format!("\"{OLD}.archived-20")), "{stdout}");

    // Once notes.md's content is under another key too, at any path, only
    // MEMORY.md is left to warn about.
    fx.sql(&format!(
        "INSERT INTO memory_files VALUES ('me/thing', 'folded/notes.md', 'a fact\n', 'x', '{T1}', 0)"
    ));
    let out = fx.admin(&["remove", OLD, "--dry-run"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    assert!(stdout.contains("Warning: 1 of these file(s)"), "{stdout}");

    // A key whose every file is somewhere else gets no warning at all.
    fx.sql(&format!(
        "INSERT INTO memory_files VALUES ('copy', 'notes.md', 'a fact\n', 'x', '{T1}', 0)"
    ));
    let out = fx.admin(&["remove", "copy", "--dry-run"]);
    assert_exit(&out, 0);
    assert!(!text(&out).0.contains("Warning"), "{:?}", text(&out));
}

// ---------------------------------------------------------------- restore

/// A remove and a restore from the backup it took are exact inverses:
/// content, source (NULL included), timestamp and tombstone flag.
#[test]
fn restore_puts_back_exactly_the_rows_a_backup_holds() {
    let fx = Fixture::new();
    let before = fx.dump();
    assert_exit(&fx.admin(&["remove", OLD, "--yes"]), 0);
    let snapshot = fx.snapshots().pop().unwrap();

    let out = fx.admin(&["restore", snapshot.to_str().unwrap(), OLD, "--yes"]);
    assert_exit(&out, 0);
    assert_eq!(fx.dump(), before);
    assert_eq!(
        fx.snapshots().len(),
        2,
        "the restore took its own backup first"
    );
}

#[test]
fn restore_does_not_overwrite_without_the_flag_and_says_what_would_change() {
    let fx = Fixture::new();
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    // Since the backup: one file edited, one added. The edit collides with
    // the backup; the addition is not in it.
    fx.sql(&format!(
        "UPDATE memory_files SET content = 'edited since', updated_at = '2026-09-23T00:00:00.000Z'
             WHERE project_key = '{OLD}' AND file_path = 'notes.md';
         INSERT INTO memory_files VALUES ('{OLD}', 'later.md', 'new', 'laptop', '2026-09-23T00:00:00.000Z', 0);
         DELETE FROM memory_files WHERE project_key = '{OLD}' AND file_path = 'MEMORY.md';"
    ));
    let before = fx.dump();
    let snap = snapshot.to_str().unwrap();

    let out = fx.admin(&["restore", snap, OLD, "--yes"]);
    assert_exit(&out, 1);
    let (stdout, stderr) = text(&out);
    assert!(stdout.contains("1 added and 1 overwritten"), "{stdout}");
    let overwrite = stdout
        .lines()
        .find(|l| l.trim_start().starts_with("overwrite"))
        .unwrap_or_else(|| panic!("{stdout}"));
    assert!(overwrite.contains("notes.md"), "{overwrite}");
    assert!(overwrite.contains("content 12 -> 7 bytes"), "{overwrite}");
    assert!(
        overwrite.contains(&format!("updated 2026-09-23T00:00:00.000Z -> {T1}")),
        "{overwrite}"
    );
    assert!(stdout.contains("untouched  later.md"), "{stdout}");
    assert!(stderr.contains("--overwrite"), "{stderr}");
    assert_eq!(fx.dump(), before, "nothing written, not even the addition");
    assert!(fx.snapshots().is_empty());

    let out = fx.admin(&["restore", snap, OLD, "--yes", "--overwrite"]);
    assert_exit(&out, 0);
    let after = fx.dump();
    let notes = under(&after, OLD)
        .into_iter()
        .find(|r| r.1 == "notes.md")
        .unwrap();
    assert_eq!(notes.2, "a fact\n");
    assert!(
        under(&after, OLD).iter().any(|r| r.1 == "later.md"),
        "a restore never deletes what the backup does not have"
    );
    assert!(under(&after, OLD).iter().any(|r| r.1 == "MEMORY.md"));
}

/// A backup's tombstone for a file that is live now is an old deletion,
/// and putting it back deletes the file on every machine at its next pull.
/// `--overwrite` alone does not do that; `--restore-deletions` must be
/// asked for as well.
#[test]
fn restore_turns_a_live_file_into_a_tombstone_only_with_restore_deletions() {
    let fx = Fixture::new();
    // The backup has old.md as a tombstone and notes.md as "a fact".
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    let snap = snapshot.to_str().unwrap();
    // Since: old.md written again, on purpose, and notes.md edited.
    fx.sql(&format!(
        "UPDATE memory_files SET content = 'written again', deleted = 0,
             updated_at = '2026-09-23T00:00:00.000Z'
             WHERE project_key = '{OLD}' AND file_path = 'old.md';
         UPDATE memory_files SET content = 'edited since', updated_at = '2026-09-23T00:00:00.000Z'
             WHERE project_key = '{OLD}' AND file_path = 'notes.md';"
    ));
    let old_md = |fx: &Fixture| {
        under(&fx.dump(), OLD)
            .into_iter()
            .find(|r| r.1 == "old.md")
            .cloned()
            .unwrap()
    };

    let out = fx.admin(&["restore", snap, OLD, "--yes", "--overwrite"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    let skipped = stdout
        .lines()
        .find(|l| l.trim_start().starts_with("skipped"))
        .unwrap_or_else(|| panic!("{stdout}"));
    assert!(skipped.contains("old.md"), "{skipped}");
    assert!(skipped.contains("--restore-deletions"), "{skipped}");
    assert!(
        stdout.contains("1 live file(s) the backup has as deleted were left as they are"),
        "{stdout}"
    );
    assert!(!stdout.contains("tombstones, which deletes"), "{stdout}");
    let kept = old_md(&fx);
    assert_eq!((kept.2.as_str(), kept.5), ("written again", 0));
    let notes = under(&fx.dump(), OLD)
        .into_iter()
        .find(|r| r.1 == "notes.md")
        .cloned()
        .unwrap();
    assert_eq!(notes.2, "a fact\n", "the ordinary overwrite still happened");

    let out = fx.admin(&[
        "restore",
        snap,
        OLD,
        "--yes",
        "--overwrite",
        "--restore-deletions",
    ]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    let delete = stdout
        .lines()
        .find(|l| l.trim_start().starts_with("delete "))
        .unwrap_or_else(|| panic!("{stdout}"));
    assert!(delete.contains("old.md"), "{delete}");
    assert!(delete.contains("file -> tombstone"), "{delete}");
    assert!(
        stdout
            .contains("1 live file(s) turned into tombstones, which deletes them on every machine"),
        "{stdout}"
    );
    assert_eq!(old_md(&fx).5, 1, "a tombstone now, as asked");
}

#[test]
fn restore_refuses_the_live_database_and_a_key_the_backup_lacks() {
    let fx = Fixture::new();
    let before = fx.dump();
    let db = fx.db();
    let out = fx.admin(&["restore", db.to_str().unwrap(), OLD, "--yes"]);
    assert_exit(&out, 1);
    assert!(text(&out).1.contains("is the live database itself"));

    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    let out = fx.admin(&["restore", snapshot.to_str().unwrap(), "local", "--yes"]);
    assert_exit(&out, 1);
    assert!(text(&out).1.contains("The backup holds no project key"));
    assert_eq!(fx.dump(), before);
    assert!(fx.snapshots().is_empty());
}

// ---------------------------------------------------------------- every change

#[test]
fn a_dry_run_changes_nothing_and_takes_no_backup() {
    let fx = Fixture::new();
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    fx.sql(&format!(
        "DELETE FROM memory_files WHERE project_key = '{OLD}'"
    ));
    let bytes = std::fs::read(fx.db()).unwrap();
    let snap = snapshot.to_str().unwrap();

    for (args, lists) in [
        (vec!["remove", "me/thing", "--dry-run"], "MEMORY.md"),
        (
            vec!["rename", "me/thing", "me/new", "--dry-run"],
            "MEMORY.md",
        ),
        (vec!["restore", snap, OLD, "--dry-run"], "notes.md"),
        // --yes changes nothing about a dry run.
        (
            vec!["remove", "acme/app", "--dry-run", "--yes"],
            "unrelated.md",
        ),
    ] {
        let out = fx.admin(&args);
        assert_exit(&out, 0);
        let (stdout, _) = text(&out);
        assert!(
            stdout.contains(lists),
            "{args:?} names what it would touch: {stdout}"
        );
        assert!(stdout.contains("Dry run: nothing was changed"), "{stdout}");
    }
    assert_eq!(
        std::fs::read(fx.db()).unwrap(),
        bytes,
        "not one byte of the file"
    );
    assert!(fx.snapshots().is_empty());
}

/// No globbing, no prefixes, no case folding, no SQL wildcards, no
/// trimming: `=` or nothing.
#[test]
fn keys_are_matched_exactly_and_never_by_pattern() {
    let fx = Fixture::new();
    let before = fx.dump();
    for near in [
        "local",
        "local:-Users-me-*",
        "local:-Users-me-thin_",
        "local:%",
        "%",
        "*",
        "LOCAL:-USERS-ME-THING",
        "local:-Users-me-thing ",
        " local:-Users-me-thing",
    ] {
        let out = fx.admin(&["remove", near, "--yes"]);
        assert_exit(&out, 1);
        let (_, stderr) = text(&out);
        assert!(stderr.contains("matched exactly"), "{near:?}: {stderr}");
        let out = fx.admin(&["rename", near, "me/new", "--yes"]);
        assert_exit(&out, 1);
    }
    assert_eq!(fx.dump(), before);
    assert!(fx.snapshots().is_empty());
}

/// Without `--yes`, the key has to be typed back exactly.
#[test]
fn a_change_goes_ahead_only_when_the_key_is_typed_back() {
    let fx = Fixture::new();
    let before = fx.dump();
    for typed in [
        "",
        "y\n",
        "yes\n",
        "local:-Users-me-thin\n",
        "LOCAL:-USERS-ME-THING\n",
    ] {
        let out = fx.admin_typing(&["remove", OLD], typed);
        assert_exit(&out, 1);
    }
    assert_eq!(fx.dump(), before);
    assert!(fx.snapshots().is_empty(), "no backup until it is confirmed");

    let out = fx.admin_typing(&["remove", OLD], &format!("{OLD}\n"));
    assert_exit(&out, 0);
    assert!(under(&fx.dump(), OLD).is_empty());
}

/// Every change is refused when there is nowhere to put its backup, or the
/// backup cannot be written: never "go ahead without one".
#[test]
fn no_backup_means_no_change() {
    let fx = Fixture::new();
    let before = fx.dump();

    let out = fx
        .command(&["remove", OLD, "--yes"])
        .env_remove("RECALL_BACKUP_DIR")
        .stdin(Stdio::null())
        .output()
        .unwrap();
    assert_exit(&out, 1);
    assert!(text(&out).1.contains("RECALL_BACKUP_DIR is not set"));

    let not_a_dir = fx.dir.path().join("a-file");
    std::fs::write(&not_a_dir, "").unwrap();
    for args in [
        vec!["remove", OLD, "--yes"],
        vec!["rename", OLD, "me/new", "--yes"],
    ] {
        let out = fx
            .command(&args)
            .env("RECALL_BACKUP_DIR", &not_a_dir)
            .stdin(Stdio::null())
            .output()
            .unwrap();
        assert_exit(&out, 1);
        assert!(
            text(&out).1.contains("taking the backup"),
            "{:?}",
            text(&out)
        );
    }
    assert_eq!(fx.dump(), before);
}

/// A backup that fails part way leaves nothing behind: a partial file named
/// like a good snapshot is worse than none, since it sorts among them and
/// does not open. A file size limit cuts `VACUUM INTO` short for real; the
/// shell ignores SIGXFSZ first so the write fails with EFBIG instead of the
/// signal killing the process.
#[cfg(target_os = "linux")]
#[test]
fn a_backup_cut_short_leaves_no_partial_file_and_changes_nothing() {
    let fx = Fixture::new();
    // Beside a running server, as it usually is. The command then maps the
    // server's WAL index (`recall.db-shm`, 32 KiB) rather than creating
    // one, which the file size limit below would refuse before the backup
    // is ever reached.
    let _server = Running::start(&fx.db());
    let before = fx.dump();
    let out = Command::new("/bin/sh")
        .arg("-c")
        // A few 512-byte blocks: less than one database page, so the very
        // first page the backup writes is cut short.
        .arg(r#"trap '' XFSZ; ulimit -f 4 && exec "$0" admin remove "$1" --yes"#)
        .arg(env!("CARGO_BIN_EXE_recall-server"))
        .arg(OLD)
        .env_clear()
        .env("RECALL_DB_PATH", fx.db())
        .env("RECALL_BACKUP_DIR", fx.backups())
        .stdin(Stdio::null())
        .output()
        .unwrap();
    assert_exit(&out, 1);
    assert!(
        text(&out).1.contains("taking the backup"),
        "{:?}",
        text(&out)
    );
    assert!(
        fx.backups().join("admin").is_dir(),
        "the backup was started, so this is the case that matters"
    );
    assert_eq!(fx.snapshots(), Vec::<PathBuf>::new(), "no partial file");
    assert_eq!(fx.dump(), before);
}

/// A push that lands while the owner is reading the plan: the backup taken
/// after the confirmation does not hold what was shown, so nothing changes,
/// and that backup is deleted rather than left to pile up, one per retry.
#[test]
fn a_backup_that_does_not_hold_what_was_shown_stops_the_change_and_is_deleted() {
    let fx = Fixture::new();
    let mut child = fx
        .command(&["remove", OLD])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut stdout = child.stdout.take().unwrap();
    let mut shown = Vec::new();
    let mut buf = [0u8; 4096];
    while !String::from_utf8_lossy(&shown).contains("without the quotes: ") {
        let n = stdout.read(&mut buf).unwrap();
        assert!(
            n > 0,
            "ended before asking: {}",
            String::from_utf8_lossy(&shown)
        );
        shown.extend_from_slice(&buf[..n]);
    }
    fx.sql(&format!(
        "INSERT INTO memory_files VALUES ('{OLD}', 'late.md', 'arrived', 'x', '{T1}', 0)"
    ));
    let before = fx.dump();
    child
        .stdin
        .take()
        .unwrap()
        .write_all(format!("{OLD}\n").as_bytes())
        .unwrap();
    let mut rest = String::new();
    stdout.read_to_string(&mut rest).unwrap();
    let out = child.wait_with_output().unwrap();

    assert_eq!(out.status.code(), Some(1), "{rest}");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("does not hold the rows shown above"),
        "{stderr}"
    );
    assert!(stderr.contains("The database was not changed."), "{stderr}");
    assert!(rest.contains("Deleted that backup again"), "{rest}");
    assert!(fx.snapshots().is_empty(), "{:?}", fx.snapshots());
    assert_eq!(fx.dump(), before);
}

/// `changes()` disagreeing with the plan rolls the whole change back. A
/// trigger that silently skips one row stands in for whatever could make
/// that happen in real life; it is exactly the case the check exists for,
/// since every other row *was* changed inside the transaction.
#[test]
fn a_changes_mismatch_rolls_back_every_command() {
    let fx = Fixture::new();
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    let snap = snapshot.to_str().unwrap();

    for (trigger, args) in [
        (
            "BEFORE UPDATE ON memory_files WHEN OLD.file_path = 'notes.md'",
            vec!["rename", OLD, "me/new", "--yes"],
        ),
        (
            "BEFORE DELETE ON memory_files WHEN OLD.file_path = 'notes.md'",
            vec!["remove", OLD, "--yes"],
        ),
        (
            "BEFORE INSERT ON memory_files WHEN NEW.file_path = 'notes.md'",
            vec!["restore", snap, OLD, "--yes"],
        ),
    ] {
        if args[0] == "restore" {
            // Something for the restore to add back.
            fx.sql(&format!(
                "DELETE FROM memory_files WHERE project_key = '{OLD}'"
            ));
        }
        let before = fx.dump();
        let logged = leaves(&fx.db()).len();
        fx.sql(&format!(
            "CREATE TRIGGER skip_one {trigger} BEGIN SELECT RAISE(IGNORE); END;"
        ));
        let out = fx.admin(&args);
        fx.sql("DROP TRIGGER skip_one");

        assert_exit(&out, 1);
        let (_, stderr) = text(&out);
        assert!(stderr.contains("were expected"), "{args:?}: {stderr}");
        assert!(stderr.contains("rolled back"), "{args:?}: {stderr}");
        assert_eq!(fx.dump(), before, "{args:?} left a partial change");
        assert_eq!(leaves(&fx.db()).len(), logged, "{args:?} left a leaf");
    }

    // The same for the jobs a change closes: one skipped is the whole
    // change rolled back.
    let job = queue_merge(&fx.db(), "me/thing", "MEMORY.md", "- other, edited\n");
    let before = (fx.dump(), jobs(&fx.db()), leaves(&fx.db()).len());
    fx.sql("CREATE TRIGGER skip_job BEFORE UPDATE ON jobs BEGIN SELECT RAISE(IGNORE); END;");
    let out = fx.admin(&["remove", "me/thing", "--yes"]);
    fx.sql("DROP TRIGGER skip_job");
    assert_exit(&out, 1);
    let (_, stderr) = text(&out);
    assert!(
        stderr.contains("closed 0 merge job(s) where 1 were expected"),
        "{stderr}"
    );
    assert_eq!((fx.dump(), jobs(&fx.db()), leaves(&fx.db()).len()), before);
    assert_eq!(job_state(&fx.db(), &job).0, "queued");
}

/// `docker exec` runs as root unless told otherwise; a change made as
/// anyone but the database's owner is refused before it starts. End to end
/// only where the test itself can hand the file to someone else, which is
/// as root; the decision itself is unit-tested in `admin.rs` everywhere.
#[cfg(target_os = "linux")]
#[test]
fn a_change_as_someone_other_than_the_owner_is_refused() {
    use std::os::unix::fs::MetadataExt;
    if std::fs::metadata("/proc/self").map(|m| m.uid()).ok() != Some(0) {
        return;
    }
    let fx = Fixture::new();
    let before = fx.dump();
    std::os::unix::fs::chown(fx.db(), Some(1000), Some(1000)).unwrap();
    let out = fx.admin(&["remove", OLD, "--yes"]);
    assert_exit(&out, 1);
    assert!(text(&out).1.contains("-u node"), "{:?}", text(&out));
    assert_eq!(fx.dump(), before);
    // Reading is fine as anyone.
    assert_exit(&fx.admin(&["list"]), 0);
}

// ---------------------------------------------------------------- merge jobs

/// A leaf these tests do not look at, for the store's audited writes.
fn some_leaf(seq: u64, at: &str) -> Vec<u8> {
    leaf::encode(
        seq,
        at,
        leaf::action::START,
        &leaf::Actor::Server,
        leaf::subject_start("test"),
        None,
    )
}

/// What a stale push of `content` to `key`/`path` does with a worker
/// approved: the content is stored at once, and a job queued to merge it
/// with the version it displaced. Answers the job's id.
fn queue_merge(db: &Path, key: &str, path: &str, content: &str) -> String {
    static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
    let id = format!(
        "job_{}",
        NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
    );
    let incoming = MergeSide {
        sha256: recall_wire::content_sha256(content),
        content: content.into(),
        source_env: "laptop".into(),
        updated_at: T1.into(),
    };
    let (queued, _) = Store::open(db)
        .unwrap()
        .write_and_queue_merge_audited(
            key,
            path,
            &incoming,
            &id,
            OffsetDateTime::now_utc(),
            |seq, at, _| some_leaf(seq, at),
        )
        .unwrap();
    assert_eq!(queued, Queued::Queued(id.clone()));
    id
}

/// A worker leasing the oldest job it may run, as `POST /v1/jobs/claim`
/// does: the job's id and the lease its result is posted under.
fn claim(db: &Path) -> Option<(String, String)> {
    let lease = format!("lse_{}", recall_server::now());
    let job = Store::open(db)
        .unwrap()
        .claim_job_audited(
            &["merge".to_string()],
            &lease,
            Duration::from_secs(600),
            OffsetDateTime::now_utc(),
            |seq, at, _| some_leaf(seq, at),
        )
        .unwrap()?;
    Some((job.id, lease))
}

/// The worker holding `lease` posting `merged` as job `id`'s result, as
/// `POST /v1/jobs/{id}/result` does.
fn post_result(db: &Path, id: &str, lease: &str, merged: &str) -> Settled {
    let settled = Store::open(db)
        .unwrap()
        .settle_job_audited(
            id,
            &ResultRequest {
                lease_id: lease.into(),
                merge: Some(MergeResult {
                    content: merged.into(),
                }),
                error: None,
                evaluate: None,
            },
            "worker",
            "job_follow_up",
            OffsetDateTime::now_utc(),
            |seq, at, _| some_leaf(seq, at),
        )
        .unwrap();
    match settled {
        Settlement::Recorded(s) => s,
        other => panic!("{other:?}"),
    }
}

/// Every job: id, state, key, path, error.
fn jobs(db: &Path) -> Vec<(String, String, String, String, Option<String>)> {
    let conn = Connection::open(db).unwrap();
    let mut stmt = conn
        .prepare("SELECT id, state, project_key, file_path, error FROM jobs ORDER BY id")
        .unwrap();
    let rows = stmt
        .query_map([], |r| {
            Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
        })
        .unwrap();
    rows.map(Result::unwrap).collect()
}

fn job_state(db: &Path, id: &str) -> (String, Option<String>) {
    let job = jobs(db).into_iter().find(|j| j.0 == id).unwrap();
    (job.1, job.4)
}

/// A rename closes the merge jobs open under the old key in its own
/// transaction, a leased one included, and says so: a worker's result
/// posted afterwards is answered as already recorded and writes nothing
/// under either key, not even onto a file a machine still on the old key
/// pushes after the rename, which a job left open would chase its result
/// onto.
#[test]
fn a_rename_closes_the_open_jobs_so_a_late_result_writes_nothing() {
    let fx = Fixture::new();
    let leased = queue_merge(&fx.db(), OLD, "notes.md", "a fact, edited\n");
    let (held, lease) = claim(&fx.db()).unwrap();
    assert_eq!(held, leased);
    let queued = queue_merge(&fx.db(), OLD, "MEMORY.md", "- edited\n");
    let elsewhere = queue_merge(&fx.db(), "acme/app", "unrelated.md", "edited\n");

    let out = fx.admin(&["rename", OLD, "me/new", "--dry-run"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    assert!(
        stdout.contains("Open merge jobs for these rows: 2, closed with the change"),
        "{stdout}"
    );
    assert_eq!(job_state(&fx.db(), &queued).0, "queued", "not by a dry run");

    let out = fx.admin(&["rename", OLD, "me/new", "--yes"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    for id in [&leased, &queued] {
        assert!(stdout.contains(&format!("  {id}  ")), "listed: {stdout}");
    }
    assert!(
        stdout.contains(&format!(
            "Closed 2 open merge job(s), so no result lands on these rows: {queued}, {leased}."
        )),
        "{stdout}"
    );
    for id in [&leased, &queued] {
        let (state, error) = job_state(&fx.db(), id);
        assert_eq!(state, "done");
        assert!(
            error.as_deref().unwrap().starts_with("renamed by admin"),
            "{error:?}"
        );
    }
    assert_eq!(job_state(&fx.db(), &elsewhere).0, "queued", "another key's");

    // A machine still on the old key pushes, then the worker answers.
    let late = Store::open(fx.db()).unwrap();
    late.upsert_audited(OLD, "notes.md", "pushed after\n", "laptop", some_leaf)
        .unwrap();
    let moved = under(&fx.dump(), "me/new")
        .into_iter()
        .cloned()
        .collect::<Vec<_>>();
    let settled = post_result(&fx.db(), &leased, &lease, "a fact, merged\n");
    assert!(!settled.applied && !settled.queued, "{settled:?}");
    assert_eq!(settled.response.state, "done");
    assert_eq!(settled.response.follow_up, None);

    let after = fx.dump();
    assert_eq!(under(&after, "me/new"), moved.iter().collect::<Vec<_>>());
    let old: Vec<_> = under(&after, OLD).into_iter().map(|r| &r.2).collect();
    assert_eq!(old, ["pushed after\n"]);
    assert!(
        !after.iter().any(|r| r.2.contains("merged")),
        "the result is nowhere"
    );
    // Nothing left for a worker but the other key's job.
    assert_eq!(claim(&fx.db()).map(|(id, _)| id), Some(elsewhere));
    assert_eq!(claim(&fx.db()), None);
}

/// A remove closes its key's open jobs as done, "removed by admin", and
/// drops their input as a delete does: a result posted after it cannot
/// bring the removed notes back into a file pushed later under that key.
#[test]
fn a_remove_closes_the_open_jobs_so_a_late_result_writes_nothing() {
    let fx = Fixture::new();
    let job = queue_merge(&fx.db(), OLD, "notes.md", "a fact, edited\n");
    let (_, lease) = claim(&fx.db()).unwrap();

    let out = fx.admin(&["remove", OLD, "--yes"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    assert!(
        stdout.contains("Open merge jobs for these rows: 1, closed with the change"),
        "{stdout}"
    );
    assert!(
        stdout.contains(&format!(
            "Closed 1 open merge job(s), so no result lands on these rows: {job}."
        )),
        "{stdout}"
    );
    let (state, error) = job_state(&fx.db(), &job);
    assert_eq!(state, "done");
    assert!(
        error.as_deref().unwrap().starts_with("removed by admin"),
        "{error:?}"
    );
    let payload: String = Connection::open(fx.db())
        .unwrap()
        .query_row("SELECT payload FROM jobs WHERE id = ?1", [&job], |r| {
            r.get(0)
        })
        .unwrap();
    assert_eq!(payload, "{}", "the removed notes are not kept in the job");

    Store::open(fx.db())
        .unwrap()
        .upsert_audited(OLD, "notes.md", "a new start\n", "laptop", some_leaf)
        .unwrap();
    let settled = post_result(&fx.db(), &job, &lease, "a fact, merged\n");
    assert!(!settled.applied && !settled.queued, "{settled:?}");
    assert_eq!(settled.response.follow_up, None);
    let old: Vec<_> = under(&fx.dump(), OLD)
        .into_iter()
        .map(|r| r.2.clone())
        .collect();
    assert_eq!(old, ["a new start\n"]);
    assert_eq!(claim(&fx.db()), None);
}

/// A restore closes the open jobs for each path it writes, and only those.
/// Left open, the job's result (a merge of the version the restore
/// replaced) would no longer match the restored file, and be queued again
/// against it as a follow-up, to land over the restore once merged.
#[test]
fn a_restore_closes_the_open_jobs_of_the_rows_it_writes() {
    let fx = Fixture::new();
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    let snap = snapshot.to_str().unwrap();
    // Since the backup: notes.md edited on a laptop, with a merge queued;
    // extra.md new, with a merge queued too, which the restore leaves be.
    let overwritten = queue_merge(&fx.db(), OLD, "notes.md", "a fact, edited\n");
    let (_, lease) = claim(&fx.db()).unwrap();
    Store::open(fx.db())
        .unwrap()
        .upsert_audited(OLD, "extra.md", "one\n", "laptop", some_leaf)
        .unwrap();
    let untouched = queue_merge(&fx.db(), OLD, "extra.md", "two\n");

    let out = fx.admin(&["restore", snap, OLD, "--overwrite", "--dry-run"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    assert!(
        stdout.contains("Open merge jobs for these rows: 1, closed with the change"),
        "{stdout}"
    );
    assert!(stdout.contains(&overwritten), "{stdout}");
    assert!(!stdout.contains(&untouched), "{stdout}");

    let out = fx.admin(&["restore", snap, OLD, "--overwrite", "--yes"]);
    assert_exit(&out, 0);
    let (stdout, _) = text(&out);
    assert!(
        stdout.contains(&format!(
            "Closed 1 open merge job(s), so no result lands on these rows: {overwritten}."
        )),
        "{stdout}"
    );
    let (state, error) = job_state(&fx.db(), &overwritten);
    assert_eq!(state, "done");
    assert!(
        error.as_deref().unwrap().starts_with("restored by admin"),
        "{error:?}"
    );
    assert_eq!(job_state(&fx.db(), &untouched).0, "queued");

    let settled = post_result(&fx.db(), &overwritten, &lease, "a fact, merged\n");
    assert!(!settled.applied && !settled.queued, "{settled:?}");
    assert_eq!(settled.response.follow_up, None);
    let notes = under(&fx.dump(), OLD)
        .into_iter()
        .find(|r| r.1 == "notes.md")
        .unwrap()
        .clone();
    assert_eq!(notes.2, "a fact\n", "the backup's version stands");
    assert_eq!(claim(&fx.db()).map(|(id, _)| id), Some(untouched));
}

// ---------------------------------------------------------------- the audit log

/// Every leaf in the log, parsed.
fn leaves(db: &Path) -> Vec<Value> {
    let store = Store::open(db).unwrap();
    let n = store.audit_checkpoint().0;
    store
        .audit_entries(0, n, usize::MAX)
        .unwrap()
        .iter()
        .map(|e| serde_json::from_slice(&e.leaf).unwrap())
        .collect()
}

/// Each change that commits appends exactly one leaf, the host's, naming
/// its keys, its counts, the jobs it closed and its backup by file name;
/// listing, a dry run, a refusal and a confirmation not given append
/// nothing.
#[test]
fn each_committed_change_appends_one_leaf_and_nothing_else_does() {
    let fx = Fixture::new();
    let job = queue_merge(&fx.db(), OLD, "notes.md", "a fact, edited\n");
    let n = leaves(&fx.db()).len();

    for (args, code) in [
        (vec!["list"], 0),
        (vec!["list", OLD], 0),
        (vec!["rename", OLD, "me/new", "--dry-run"], 0),
        (vec!["remove", OLD, "--dry-run"], 0),
        (vec!["rename", OLD, "me/thing", "--yes"], 1),
        (vec!["remove", "no/such/key", "--yes"], 1),
    ] {
        assert_exit(&fx.admin(&args), code);
    }
    assert_exit(&fx.admin_typing(&["remove", OLD], "not it\n"), 1);
    assert_eq!(leaves(&fx.db()).len(), n, "nothing appended yet");

    let out = fx.admin(&["rename", OLD, "me/new", "--yes"]);
    assert_exit(&out, 0);
    let all = leaves(&fx.db());
    assert_eq!(all.len(), n + 1);
    let backup = fx.snapshots()[0]
        .file_name()
        .unwrap()
        .to_str()
        .unwrap()
        .to_string();
    let rename = &all[n];
    assert_eq!(rename["seq"], json!(n));
    assert_eq!(rename["action"], "admin_rename");
    assert_eq!(rename["actor"], json!({"kind": "host"}));
    assert_eq!(rename["request"], Value::Null);
    assert_eq!(
        rename["subject"],
        json!({"from": OLD, "to": "me/new", "rows": 3, "jobs_closed": [job], "backup": backup})
    );
    assert!(
        text(&out).0.contains(&format!(
            "Recorded in the audit log as leaf {n} (admin_rename)"
        )),
        "{:?}",
        text(&out)
    );

    assert_exit(&fx.admin(&["remove", "me/new", "--yes"]), 0);
    let all = leaves(&fx.db());
    assert_eq!(all.len(), n + 2);
    assert_eq!(all[n + 1]["action"], "admin_remove");
    assert_eq!(all[n + 1]["subject"]["project_key"], "me/new");
    assert_eq!(all[n + 1]["subject"]["rows"], 3);
    assert_eq!(all[n + 1]["subject"]["jobs_closed"], json!([]));

    // The rename's backup holds OLD's rows: put them back.
    let source = fx.snapshots()[0].clone();
    assert_exit(
        &fx.admin(&["restore", source.to_str().unwrap(), OLD, "--yes"]),
        0,
    );
    let all = leaves(&fx.db());
    assert_eq!(all.len(), n + 3);
    let restore = &all[n + 2];
    assert_eq!(restore["action"], "admin_restore");
    assert_eq!(restore["subject"]["project_key"], OLD);
    assert_eq!(restore["subject"]["source"], json!(backup));
    assert_eq!(
        (
            &restore["subject"]["added"],
            &restore["subject"]["overwritten"],
            &restore["subject"]["deleted"]
        ),
        (&json!(3), &json!(0), &json!(0))
    );
    assert_ne!(restore["subject"]["backup"], json!(backup), "its own");
    // Nothing of any file is in them: no content, and no path.
    for leaf in &all[n..] {
        let text = leaf.to_string();
        for never in ["a fact", "notes.md", "MEMORY.md", "/backups"] {
            assert!(!text.contains(never), "{never:?} in {text}");
        }
    }
    // Nothing more to restore is nothing to record.
    assert_exit(
        &fx.admin(&["restore", source.to_str().unwrap(), OLD, "--yes"]),
        0,
    );
    assert_eq!(leaves(&fx.db()).len(), n + 3);
}

/// The leaf commits with the change or not at all. A log that refuses the
/// leaf leaves every row and job as it was, rather than a change the log
/// never heard of; a leaf appended after the commit, in a transaction of
/// its own, would leave the change made here.
#[test]
fn a_change_whose_leaf_cannot_be_appended_is_not_made() {
    let fx = Fixture::new();
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    let snap = snapshot.to_str().unwrap();
    let job = queue_merge(&fx.db(), OLD, "notes.md", "a fact, edited\n");
    fx.sql(
        "CREATE TRIGGER no_admin_leaves BEFORE INSERT ON audit_log
         WHEN CAST(NEW.leaf AS TEXT) LIKE '%\"action\":\"admin_%'
         BEGIN SELECT RAISE(ABORT, 'no admin leaves here'); END;",
    );
    let before = (fx.dump(), jobs(&fx.db()), leaves(&fx.db()).len());
    for args in [
        vec!["rename", OLD, "me/new", "--yes"],
        vec!["remove", OLD, "--yes"],
        vec!["restore", snap, OLD, "--overwrite", "--yes"],
    ] {
        let out = fx.admin(&args);
        assert_exit(&out, 1);
        let (_, stderr) = text(&out);
        assert!(
            stderr.contains("no admin leaves here"),
            "{args:?}: {stderr}"
        );
        assert!(stderr.contains("The database was not changed."), "{stderr}");
        assert_eq!(
            (fx.dump(), jobs(&fx.db()), leaves(&fx.db()).len()),
            before,
            "{args:?}"
        );
    }
    assert_eq!(job_state(&fx.db(), &job).0, "queued");
}

// ---------------------------------------------------------------- beside a server

/// A real server on a real socket, serving the fixture's database file.
struct Running {
    addr: SocketAddr,
    stop: Option<tokio::sync::oneshot::Sender<()>>,
    // Held, never read: the server runs on it, and stops when it drops.
    _rt: tokio::runtime::Runtime,
}

impl Running {
    fn start(db: &Path) -> Self {
        Self::start_with(db, None)
    }

    /// With `claude`, merge is on and runs that stand-in CLI, which the
    /// server is told is logged in.
    fn start_with(db: &Path, claude: Option<&Path>) -> Self {
        let rt = tokio::runtime::Runtime::new().unwrap();
        let store = Arc::new(Store::open(db).unwrap());
        let server = Server::new(
            Config {
                token: TOKEN.into(),
                merge_enabled: claude.is_some(),
                claude_bin: claude.map_or("definitely-not-a-real-binary".into(), |p| {
                    p.to_str().unwrap().to_string()
                }),
                merge_timeout: Duration::from_secs(20),
                rate_limit_max: 100_000,
                trusted_ip_header: String::new(),
                ..Config::default()
            },
            store,
        );
        server.set_claude_status(Status {
            checked_at: recall_server::now(),
            available: claude.is_some(),
            logged_in: claude.is_some(),
            error: String::new(),
        });
        let listener = rt
            .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
            .unwrap();
        let addr = listener.local_addr().unwrap();
        let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
        // The router, not `serve_with_shutdown`: that also starts the
        // background status probe, which would run the stand-in as `claude
        // auth status` and overwrite the status set above.
        let app = server
            .router()
            .into_make_service_with_connect_info::<SocketAddr>();
        rt.spawn(async move {
            let _ = axum::serve(listener, app)
                .with_graceful_shutdown(async {
                    let _ = stopped.await;
                })
                .await;
        });
        Self {
            addr,
            stop: Some(stop),
            _rt: rt,
        }
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        if let Some(stop) = self.stop.take() {
            let _ = stop.send(());
        }
        // Give graceful shutdown a moment; dropping the runtime does the
        // rest.
        thread::sleep(Duration::from_millis(50));
    }
}

/// One HTTP/1.1 request over a plain socket, returning status and body.
fn http(addr: SocketAddr, method: &str, path: &str, body: &str) -> (u16, String) {
    let mut s = TcpStream::connect(addr).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(30))).unwrap();
    write!(
        s,
        "{method} {path} HTTP/1.1\r\nHost: test\r\nAuthorization: Bearer {TOKEN}\r\n\
         Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .unwrap();
    let mut resp = String::new();
    s.read_to_string(&mut resp).unwrap();
    let status = resp[9..12].parse().unwrap();
    let body = resp.split_once("\r\n\r\n").map(|(_, b)| b).unwrap_or("");
    (status, body.to_string())
}

fn push(addr: SocketAddr, key: &str, path: &str) -> u16 {
    let body = format!(
        r#"{{"project_key":"{key}","file_path":"{path}","content":"x","source_env":"test"}}"#
    );
    http(addr, "POST", "/sync", &body).0
}

fn pulled(addr: SocketAddr, key: &str) -> usize {
    let (status, body) = http(addr, "GET", &format!("/sync?project_key={key}"), "");
    assert_eq!(status, 200, "{body}");
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    v["files"].as_array().unwrap().len()
}

/// The ordinary case: the server is up and taking pushes, a write is in
/// flight when the command starts, and the command waits its turn instead
/// of failing; the server sees the result on its next request.
#[test]
fn a_change_works_while_a_server_is_serving_the_same_file() {
    let fx = Fixture::new();
    let server = Running::start(&fx.db());
    let addr = server.addr;

    // Someone else holding the write lock as the command starts, as the
    // server does for the length of a push.
    let holder = Connection::open(fx.db()).unwrap();
    holder.execute_batch("BEGIN IMMEDIATE").unwrap();
    let release = thread::spawn(move || {
        thread::sleep(Duration::from_millis(700));
        holder.execute_batch("COMMIT").unwrap();
    });
    let pusher = thread::spawn(move || {
        (0..25)
            .map(|i| {
                thread::sleep(Duration::from_millis(10));
                push(addr, "busy/project", &format!("f{i}.md"))
            })
            .collect::<Vec<_>>()
    });

    let out = fx.admin(&["rename", OLD, "me/new", "--yes"]);
    release.join().unwrap();
    let statuses = pusher.join().unwrap();

    assert_exit(&out, 0);
    assert!(statuses.iter().all(|s| *s == 200), "{statuses:?}");
    assert_eq!(pulled(addr, "me/new"), 3);
    assert_eq!(pulled(addr, OLD), 0);
    assert_eq!(pulled(addr, "busy/project"), 25);
    assert_eq!(push(addr, "me/new", "after.md"), 200);
}

/// The case that has to fail cleanly: a writer that never lets go, so the
/// command cannot get the write lock. (Under WAL, which the server keeps
/// the file in, nothing else can hold a change up: a reader no longer
/// blocks a commit.) The command waits out the busy timeout, rolls back,
/// says so, and leaves the file as it found it, with the server still
/// serving.
#[test]
fn a_lock_held_past_the_busy_timeout_fails_cleanly() {
    let fx = Fixture::new();
    let server = Running::start(&fx.db());
    let before = fx.dump();

    let writer = Connection::open(fx.db()).unwrap();
    writer.execute_batch("BEGIN IMMEDIATE").unwrap();
    writer
        .execute_batch("UPDATE memory_files SET content = 'uncommitted' WHERE deleted = 0")
        .unwrap();

    let started = Instant::now();
    let out = fx.admin(&["remove", OLD, "--yes"]);
    let waited = started.elapsed();
    writer.execute_batch("ROLLBACK").unwrap();

    assert_exit(&out, 1);
    let (_, stderr) = text(&out);
    assert!(stderr.contains("stayed locked"), "{stderr}");
    assert!(stderr.contains("The database was not changed."), "{stderr}");
    assert!(
        waited >= Duration::from_secs(4),
        "it waited for the lock rather than giving up at once: {waited:?}"
    );

    assert_eq!(fx.dump(), before);
    let journal = fx.dir.path().join("recall.db-journal");
    assert!(!journal.exists(), "no hot journal left behind");
    let ok: String = Connection::open(fx.db())
        .unwrap()
        .query_row("PRAGMA integrity_check", [], |r| r.get(0))
        .unwrap();
    assert_eq!(ok, "ok");
    assert_eq!(push(server.addr, OLD, "after.md"), 200);
    assert_eq!(pulled(server.addr, OLD), 4);
}

/// The other case that stays locked: a file still in the rollback journal,
/// which an admin command leaves as it finds it (only a server switches a
/// file to WAL), where a reader mid-read does hold a change up past the
/// busy timeout. The message names that case too, rather than sending the
/// owner to look for a writer that is not there.
#[test]
fn a_reader_holds_up_a_change_on_a_file_still_in_the_rollback_journal() {
    let fx = Fixture::new();
    let before = fx.dump();
    let mode: String = Connection::open(fx.db())
        .unwrap()
        .query_row("PRAGMA journal_mode = DELETE", [], |r| r.get(0))
        .unwrap();
    assert_eq!(mode, "delete");

    let reader = Connection::open(fx.db()).unwrap();
    reader.execute_batch("BEGIN").unwrap();
    let _: i64 = reader
        .query_row("SELECT count(*) FROM memory_files", [], |r| r.get(0))
        .unwrap();
    let out = fx.admin(&["remove", OLD, "--yes"]);
    reader.execute_batch("COMMIT").unwrap();

    assert_exit(&out, 1);
    let (_, stderr) = text(&out);
    assert!(stderr.contains("stayed locked"), "{stderr}");
    assert!(
        stderr.contains("still in the rollback journal") && stderr.contains("a reader too"),
        "{stderr}"
    );
    assert!(stderr.contains("The database was not changed."), "{stderr}");
    assert_eq!(fx.dump(), before);
    let mode: String = Connection::open(fx.db())
        .unwrap()
        .query_row("PRAGMA journal_mode", [], |r| r.get(0))
        .unwrap();
    assert_eq!(mode, "delete", "the command left the journal mode alone");
}

/// What WAL changed for these commands: a reader mid-read, as sqlite-web
/// is while a page loads, no longer holds a change up (under the rollback
/// journal this was the case that timed out). The change commits at once,
/// the reader goes on seeing the moment it began at until it is done, the
/// server keeps serving throughout, and `list` works beside all three.
#[test]
fn a_reader_mid_read_no_longer_holds_up_a_change() {
    let fx = Fixture::new();
    let server = Running::start(&fx.db());
    let count = |c: &Connection| -> i64 {
        c.query_row(
            "SELECT count(*) FROM memory_files WHERE project_key = ?1",
            (OLD,),
            |r| r.get(0),
        )
        .unwrap()
    };

    let reader = Connection::open(fx.db()).unwrap();
    let mode: String = reader
        .query_row("PRAGMA journal_mode", [], |r| r.get(0))
        .unwrap();
    assert_eq!(mode, "wal", "the server left the file in WAL");
    reader.execute_batch("BEGIN").unwrap();
    assert_eq!(count(&reader), 3);

    let started = Instant::now();
    let out = fx.admin(&["remove", OLD, "--yes"]);
    assert_exit(&out, 0);
    assert!(
        started.elapsed() < Duration::from_secs(4),
        "it did not wait out the busy timeout behind a reader: {:?}",
        started.elapsed()
    );
    assert_eq!(count(&reader), 3, "the reader's moment, unchanged");
    let listed = fx.admin(&["list"]);
    assert_exit(&listed, 0);
    assert!(!text(&listed).0.contains(OLD), "{}", text(&listed).0);
    assert_eq!(push(server.addr, "me/thing", "during.md"), 200);

    reader.execute_batch("COMMIT").unwrap();
    assert_eq!(count(&reader), 0);
    assert_eq!(pulled(server.addr, OLD), 0);
    assert_eq!(pulled(server.addr, "me/thing"), 2);
}

/// A stand-in `claude` that takes `seconds` to merge, which is how long a
/// push holds a row it has read without holding any lock.
#[cfg(unix)]
fn slow_claude(dir: &Path, seconds: u32) -> PathBuf {
    use std::os::unix::fs::PermissionsExt;
    let path = dir.join("slow-claude");
    std::fs::write(
        &path,
        format!(
            "#!/bin/sh\ncat > /dev/null\nsleep {seconds}\n\
             printf '%s' '{{\"is_error\":false,\"result\":\"merged late\"}}'\n"
        ),
    )
    .unwrap();
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
    path
}

/// The race a lock cannot close: a push reads a row, merges for a while
/// holding no lock, and writes after the change has committed, partly
/// undoing it. The command cannot prevent that from its side, so it waits
/// out the merge window and says exactly what came back and what to do.
#[cfg(unix)]
#[test]
fn a_push_being_merged_as_a_change_commits_is_reported_afterwards() {
    let fx = Fixture::new();
    let claude = slow_claude(fx.dir.path(), 2);
    let server = Running::start_with(&fx.db(), Some(&claude));
    let addr = server.addr;
    let snapshot = Store::open(fx.db())
        .unwrap()
        .backup(fx.dir.path().join("periodic"), 7)
        .unwrap();
    let snap = snapshot.to_str().unwrap().to_string();

    // A remove, with a push to notes.md already merging as it commits.
    let pushing = thread::spawn(move || push(addr, OLD, "notes.md"));
    thread::sleep(Duration::from_millis(500));
    let out = fx
        .command(&["remove", OLD, "--yes"])
        .env("RECALL_MERGE_TIMEOUT_MS", "4000")
        .stdin(Stdio::null())
        .output()
        .unwrap();
    assert_eq!(pushing.join().unwrap(), 200);
    assert_exit(&out, 3);
    let (stdout, stderr) = text(&out);
    assert!(stdout.contains("Done: removed 3 row(s)"), "{stdout}");
    assert!(stdout.contains("Waiting 5.0s"), "{stdout}");
    assert!(
        stderr.contains(&format!(
            "1 row(s) are under \"{OLD}\" again:\n  notes.md\n"
        )),
        "{stderr}"
    );
    assert!(
        stderr.contains(&format!("run `recall-server admin remove {OLD}` again")),
        "{stderr}"
    );
    assert!(!stderr.contains("was not changed"), "{stderr}");
    let back = under(&fx.dump(), OLD)
        .into_iter()
        .cloned()
        .collect::<Vec<_>>();
    assert_eq!(back.len(), 1);
    assert_eq!(back[0].2, "merged late", "what the push wrote after");

    // A restore --overwrite, with a push to notes.md merging the version
    // the restore replaces.
    fx.sql(&format!(
        "UPDATE memory_files SET content = 'edited since' WHERE project_key = '{OLD}'"
    ));
    let pushing = thread::spawn(move || push(addr, OLD, "notes.md"));
    thread::sleep(Duration::from_millis(500));
    let out = fx
        .command(&["restore", &snap, OLD, "--yes", "--overwrite"])
        .env("RECALL_MERGE_TIMEOUT_MS", "4000")
        .stdin(Stdio::null())
        .output()
        .unwrap();
    assert_eq!(pushing.join().unwrap(), 200);
    assert_exit(&out, 3);
    let (_, stderr) = text(&out);
    assert!(
        stderr.contains(&format!(
            "1 restored row(s) under \"{OLD}\" are no longer what it wrote:\n  notes.md\n"
        )),
        "{stderr}"
    );
    assert!(
        stderr.contains(&format!(
            "run `recall-server admin restore --overwrite {snap} {OLD}` again"
        )),
        "{stderr}"
    );

    // And with nothing in flight, the same check passes quietly.
    let out = fx.admin(&["restore", &snap, OLD, "--yes", "--overwrite"]);
    assert_exit(&out, 0);
    assert!(text(&out).0.contains("Checked: the change held."));
}

// ---------------------------------------------------------------- not over HTTP

/// The property `ARCHITECTURE.md` names: no route on the public listener
/// renames, removes or restores anything, whatever the credential. These
/// commands exist only as a subcommand.
#[tokio::test]
async fn no_http_route_renames_removes_or_restores() {
    let fx = Fixture::new();
    let before = fx.dump();
    let server = Server::new(
        Config {
            token: TOKEN.into(),
            merge_enabled: false,
            rate_limit_max: 100_000,
            ..Config::default()
        },
        Arc::new(Store::open(fx.db()).unwrap()),
    );
    for (method, path) in [
        ("POST", "/admin/rename"),
        ("POST", "/admin/remove"),
        ("POST", "/admin/restore"),
        ("POST", "/admin/stats"),
        ("DELETE", "/admin/stats"),
        ("DELETE", "/admin/projects/me%2Fthing"),
        ("POST", "/admin/projects/me%2Fthing/rename"),
        ("DELETE", "/sync?project_key=me/thing"),
        ("PUT", "/sync"),
        ("PATCH", "/sync"),
    ] {
        let req = Request::builder()
            .method(method)
            .uri(path)
            .header("authorization", format!("Bearer {TOKEN}"))
            .header("content-type", "application/json")
            .body(Body::from(
                r#"{"from":"me/thing","to":"x","project_key":"me/thing"}"#,
            ))
            .unwrap();
        let resp = server.router().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND, "{method} {path}");
    }
    assert_eq!(fx.dump(), before);
}

// ---------------------------------------------------------------- arguments

#[test]
fn admin_arguments_are_checked_like_the_servers_own() {
    let fx = Fixture::new();
    for args in [
        vec![],
        vec!["bogus"],
        vec!["remove"],
        vec!["remove", OLD, "--force"],
        vec!["rename", OLD],
    ] {
        let out = fx.admin(&args);
        assert_exit(&out, 2);
        assert!(text(&out).1.contains("Usage:"), "{args:?}");
    }
    let out = fx.admin(&["help"]);
    assert_exit(&out, 0);
    assert!(text(&out).0.contains("recall-server admin restore"));
}