//! A repository on disk: the `.oxy` directory, one workspace, the operation
//! log, and the working-copy change (spec §12, §13.1).
//!
//! Not yet implemented from the spec: packs, checkpoints, multiple
//! workspaces, multi-signer policy updates, and the `objects`/`checkpoint`
//! endpoints. A repository without a policy is unverified and accepts
//! unsigned pushes; with one, everything received is checked (spec §14.5). Changes are signed individually when `core.sign` is
//! `change`; the default `seal` leaves them unsigned until push exists.

use crate::attest::{Attest, AttestBody};
use crate::diff::{ranges, split_lines, text_edits};
use crate::hash::{hash_object, ChangeHash, KeyId, ObjectRef, ObjectType};
use crate::identity::{verify as verify_signature, Domain, Identity, SecretKey, Signature};
use crate::ids::{ChangeId, NodeId, WorkspaceId};
use crate::objects::{
    Blob, Change, ChangeKind, Edit, Hlc, Mode, NodeKind, OpRef, StateObject, TreeOp,
};
use crate::oplog::{Format1Op, Op, View, SNAPSHOT_SUFFIX};
use crate::protocol::{
    open_remote, Bundle, RefResult, RefUpdate, RefsResponse, RemoteOptions, RemoteRef,
    SignedAttest, SignedPolicy,
};
use crate::state::{FileContent, Snapshot, State};
use crate::store::{Store, StoreError};
#[allow(unused_imports)]
use crate::subrepo::Pin;
use crate::tree::OpKey;
use rand::RngCore;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};

pub const OXY_DIR: &str = ".oxy";
pub const FORMAT_LINE: &str = "oxy repository format 2\n";
const FORMAT1_LINE: &str = "oxy repository format 1\n";
const BINARY_SNIFF: usize = 8 * 1024;
/// Loose objects tolerated before a commit packs them (spec §13.4).
pub const AUTOPACK_THRESHOLD: usize = 4096;

#[derive(Debug, thiserror::Error)]
pub enum RepoError {
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error(transparent)]
    Store(#[from] StoreError),
    #[error(transparent)]
    Object(#[from] crate::objects::ObjectError),
    #[error(transparent)]
    State(#[from] crate::state::StateError),
    #[error("{0}")]
    Msg(String),
    /// The operation would leave a conflict where none may be (a rebased
    /// change cannot hold one); the caller may offer another way.
    #[error("{0}")]
    Conflicted(String),
}

pub type Result<T> = std::result::Result<T, RepoError>;

fn fail<T>(s: impl Into<String>) -> Result<T> {
    Err(RepoError::Msg(s.into()))
}

mod bookmarks;
mod checkout;
mod checkpoint;
mod config;
mod doctor;
mod fsck;
mod gc;
mod git;
mod history;
mod hooks;
mod hosted;
mod identity;
mod inspect;
mod ops;
mod policy;
mod prompt;
mod receipts;
mod snapshot;
mod subrepos;
mod sync;
mod tags;
#[cfg(test)]
mod tests;
mod walk;
mod workspace;

pub use bookmarks::{remote_bookmark, BookmarkInfo, MergeReport, RemoteOnlyBookmark, StateGap};
pub use checkout::CheckoutReport;
pub use config::{parse_size, read_config_file, Config};
pub use doctor::{Check, DoctorReport, Health};
pub use fsck::{FsckReport, Problem, Severity};
pub use gc::GcReport;
pub use git::{GitExportReport, GitImportReport, GitVerifyReport};
pub use hooks::{json_array, json_str, HookOutcome};
pub use hosted::FileEdit;
pub use identity::{Rotation, SignatureStatus};
pub use inspect::{BisectStep, BlameLine, Description};
pub use ops::{RestoreReport, RevertReport, ViewChange};
pub use policy::PolicyUpdate;
pub use prompt::{PromptInfo, DEFAULT_PROMPT_FORMAT, GLYPH_PROMPT_FORMAT};
pub use receipts::ReceiptInfo;
pub use snapshot::{
    added_node_paths, diff_files_of, has_conflict_markers, is_binary, referenced_changes,
    render_change, render_hunks, similarity, ChangeStatus, DiffHunk, FileDiff, Hunk, PathStatus,
    Selection, SnapshotResult,
};
pub use subrepos::{SubrepoInfo, SubrepoSync};
pub use sync::{FetchReport, PullReport, PushReport, RefUpdated};
pub use tags::TagInfo;
pub use workspace::{FileLock, SnapshotLock, WorkingCopyLock, WorkspaceInfo};

/// The file name under which per-bookmark or per-tag data is kept: `/`
/// (and `%`) escaped, so `feature/x` is one file beside `feature` rather
/// than a directory that would clash with it.
pub(crate) fn name_file(name: &str) -> String {
    name.replace('%', "%25").replace('/', "%2F")
}

/// The name a [`name_file`] file name stands for.
pub(crate) fn file_name_to_name(file: &str) -> String {
    file.replace("%2F", "/").replace("%25", "%")
}

fn write_atomic(path: &Path, data: &[u8]) -> Result<()> {
    Ok(crate::durable::write_atomic(path, data)?)
}

/// Replace a cache file: whole or not at all, but without waiting for the
/// disk. A cache checks its own bytes when read, so one a crash leaves
/// short or stale is rebuilt rather than trusted.
/// How many changes a state index may fall behind its working copy before
/// it is written again; until then each load replays them.
const INDEX_REPLAY_LIMIT: usize = 16;
/// The changes whose encodings are kept for building graphs: a batch of
/// files often takes inline text from a few changes in turn.
const CHANGE_CACHE_COUNT: usize = 4;
const CHANGE_CACHE_BYTES: usize = 512 << 20;
/// How many state indexes a workspace keeps.
const STATE_INDEXES_KEPT: usize = 2;
const INDEX_REPLAY_TIME: std::time::Duration = std::time::Duration::from_millis(50);

pub(crate) fn write_cache(path: &Path, data: &[u8]) -> Result<()> {
    let tmp = path.with_extension(format!("tmp-{}", std::process::id()));
    fs::write(&tmp, data)?;
    fs::rename(&tmp, path)?;
    Ok(())
}

pub struct LogEntry {
    pub hash: ChangeHash,
    /// Metadata from the state; the full change may be absent in a shallow repository.
    pub change: crate::state::ChangeMeta,
    pub is_working_copy: bool,
    pub is_head: bool,
}

/// A text file's base graph key, by path, as a walk asks for it.
pub(crate) type KeyOf<'a> = dyn Fn(&str) -> Option<[u8; 32]> + Sync + 'a;

/// A file as found on disk or in an imported tree.
#[derive(Clone)]
pub(crate) struct FsEntry {
    pub kind: NodeKind,
    pub exec: bool,
    /// File content, or a symlink's target. Empty when `blob` is set.
    /// Shared, so a new file's insert holds it without a copy.
    pub content: std::sync::Arc<Vec<u8>>,
    /// The stored blob of an opaque file whose size and mtime are unchanged
    /// since the last manifest (spec §12.3 stat cache); the content was not
    /// read.
    pub blob: Option<ObjectRef>,
    /// BLAKE3 of the content, carried over from the manifest when the stat
    /// is unchanged, so the manifest can be rewritten without rehashing.
    pub content_hash: Option<[u8; 32]>,
    /// A text file known equal to the base's rendering without reading it:
    /// its stat matches the manifest and the manifest recorded it equal to
    /// a base file graph with the same key (spec §12.3). `content` is empty.
    pub unchanged: bool,
    /// The base graph key this file equals, for the next manifest.
    pub graph_key: Option<[u8; 32]>,
    /// Size and mtime already known to be current (a checkout leaving the
    /// file as it was), so the manifest need not stat it again.
    pub stat: Option<(u64, u128)>,
}

impl FsEntry {
    pub(crate) fn new(kind: NodeKind, exec: bool, content: Vec<u8>) -> FsEntry {
        FsEntry {
            kind,
            exec,
            content: std::sync::Arc::new(content),
            blob: None,
            content_hash: None,
            unchanged: false,
            graph_key: None,
            stat: None,
        }
    }
}

pub struct Repo {
    /// Oxygen-pinned subrepos skipped by the running Git export.
    pub(crate) git_export_skipped: std::cell::Cell<usize>,
    /// The manifest as last read, with the file identity it was read from,
    /// so one command parses it once (see `read_manifest`).
    pub(crate) manifest_cache: std::cell::RefCell<Option<prompt::CachedManifest>>,
    /// A listing of the working copy left by a failed clean check, for the
    /// snapshot that follows it to reuse; taken by the next walk and
    /// dropped whenever the manifest is written.
    pub(crate) listing_cache: std::cell::RefCell<Option<walk::Listing>>,
    /// The nested repositories the last walk left out, being neither
    /// registered nor tracked subrepos (oxygen#43).
    pub(crate) untracked_nested: std::cell::RefCell<Vec<String>>,
    root: PathBuf,
    oxy: PathBuf,
    store: Store,
    workspace: WorkspaceId,
    /// A hosted repository with no working copy (spec §12.2, hub spec §4.1).
    bare: bool,
    keys_dir: PathBuf,
    user_config: PathBuf,
    pub config: Config,
    /// What the working copy's file system can represent (spec §7.7).
    platform: crate::portable::Platform,
    /// The encodings of the changes last read for their inline inserts,
    /// oldest first.
    last_change: std::cell::RefCell<Vec<(ChangeHash, std::sync::Arc<Vec<u8>>)>>,
    /// `--at-op`: read the repository as it was after this operation;
    /// recording an operation is refused.
    at_op: Option<ObjectRef>,
}

/// Marker file of a bare repository, inside `.oxy/`.
const BARE_MARKER: &str = "bare";
/// Marker of a hosted repository that has no id yet: the first push binds
/// it to the pusher's id (spec §16.1).
const UNBOUND_MARKER: &str = "unbound";

/// The user's oxy directory: `$OXY_HOME`, else `$XDG_CONFIG_HOME/oxy`, else
/// `~/.config/oxy`. It holds `config.toml` (spec §19) and `keys/`.
pub fn default_home() -> PathBuf {
    if let Ok(h) = std::env::var("OXY_HOME") {
        return PathBuf::from(h);
    }
    if let Ok(x) = std::env::var("XDG_CONFIG_HOME") {
        return PathBuf::from(x).join("oxy");
    }
    // Windows has no HOME as a rule: the roaming profile's application
    // data, as Windows programs keep theirs.
    #[cfg(windows)]
    if let Some(d) = std::env::var_os("APPDATA").or_else(|| std::env::var_os("USERPROFILE")) {
        return PathBuf::from(d).join("oxy");
    }
    let home = std::env::var("HOME").unwrap_or_else(|_| ".".into());
    PathBuf::from(home).join(".config").join("oxy")
}

pub fn default_keys_dir() -> PathBuf {
    default_home().join("keys")
}

pub fn default_user_config() -> PathBuf {
    default_home().join("config.toml")
}

/// Where `create_identity` records the author.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum IdentityScope {
    /// The user file: every repository of this user signs with it.
    User,
    /// This repository only.
    Repo,
}

impl Repo {
    // ------------------------------------------------------------ lifecycle

    pub fn init(root: &Path) -> Result<Repo> {
        let mut id = [0u8; 16];
        rand::rngs::OsRng.fill_bytes(&mut id);
        Self::init_with_id(root, id)
    }

    /// Initialize a repository with a given repository ID (clones share the
    /// ID of their origin, spec §14.4).
    pub fn init_with_id(root: &Path, id: [u8; 16]) -> Result<Repo> {
        fs::create_dir_all(root)?;
        let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
        let oxy = root.join(OXY_DIR);
        if oxy.exists() {
            return fail(format!("{} is already an oxy repository", root.display()));
        }
        fs::create_dir_all(oxy.join("ops"))?;
        write_atomic(&oxy.join("format"), FORMAT_LINE.as_bytes())?;
        write_atomic(&oxy.join("id"), hex::encode(id).as_bytes())?;
        let config = Config::load(Some(&default_user_config()), &oxy.join("config"))?;
        config.save(&oxy.join("config"))?;
        let workspace = WorkspaceId::random();
        write_atomic(&oxy.join("workspace"), workspace.to_hex().as_bytes())?;
        let store = Store::open(oxy.join("objects"))?;
        let repo = Repo {
            last_change: Default::default(),
            at_op: None,
            git_export_skipped: Default::default(),
            manifest_cache: Default::default(),
            listing_cache: Default::default(),
            untracked_nested: Default::default(),
            root,
            oxy,
            store,
            workspace,
            bare: false,
            keys_dir: default_keys_dir(),
            platform: crate::portable::Platform::current(),
            user_config: default_user_config(),
            config,
        };
        repo.register_workspace(&workspace, &repo.root)?;

        let wc = repo.new_working_copy(vec![], "");
        let wc_hash = repo.store.put_change(&wc)?;
        let mut view = View::default();
        view.bookmarks.insert(
            repo.config.bookmark.clone(),
            repo.store.put_state(&StateObject::new(vec![]))?,
        );
        view.workspaces.insert(workspace, wc_hash);
        repo.record_op(&[], "oxy init", view)?;
        Ok(repo)
    }

    /// Create a bare repository: an object store, operation log and
    /// references with no workspace and no working copy, for hosting
    /// (hub spec §4.1). `root` holds `.oxy/` and nothing else.
    pub fn init_bare(root: &Path) -> Result<Repo> {
        let mut id = [0u8; 16];
        rand::rngs::OsRng.fill_bytes(&mut id);
        Self::init_bare_with_id(root, id)
    }

    pub fn init_bare_with_id(root: &Path, id: [u8; 16]) -> Result<Repo> {
        fs::create_dir_all(root)?;
        let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
        let oxy = root.join(OXY_DIR);
        if oxy.exists() {
            return fail(format!("{} is already an oxy repository", root.display()));
        }
        fs::create_dir_all(oxy.join("ops"))?;
        write_atomic(&oxy.join("format"), FORMAT_LINE.as_bytes())?;
        write_atomic(&oxy.join("id"), hex::encode(id).as_bytes())?;
        write_atomic(&oxy.join(BARE_MARKER), b"")?;
        let config = Config::load(Some(&default_user_config()), &oxy.join("config"))?;
        config.save(&oxy.join("config"))?;
        let store = Store::open(oxy.join("objects"))?;
        let repo = Repo {
            last_change: Default::default(),
            at_op: None,
            git_export_skipped: Default::default(),
            manifest_cache: Default::default(),
            listing_cache: Default::default(),
            untracked_nested: Default::default(),
            root,
            oxy,
            store,
            workspace: WorkspaceId::ZERO,
            bare: true,
            keys_dir: default_keys_dir(),
            platform: crate::portable::Platform::current(),
            user_config: default_user_config(),
            config,
        };
        let mut view = View::default();
        view.bookmarks.insert(
            repo.config.bookmark.clone(),
            repo.store.put_state(&StateObject::new(vec![]))?,
        );
        repo.record_op(&[], "oxy init --bare", view)?;
        Ok(repo)
    }

    /// A bare repository with no id yet. Its `id()` is all zero until the
    /// first push binds it (`bind_id`), so an existing repository can be
    /// pushed to a freshly created hosted one and keep its identity.
    pub fn init_bare_unbound(root: &Path) -> Result<Repo> {
        let repo = Self::init_bare_with_id(root, [0; 16])?;
        fs::remove_file(repo.oxy.join("id"))?;
        write_atomic(&repo.oxy.join(UNBOUND_MARKER), b"")?;
        Ok(repo)
    }

    pub fn is_unbound(&self) -> bool {
        self.oxy.join(UNBOUND_MARKER).is_file()
    }

    /// Give an unbound repository its id.
    pub fn bind_id(&self, id: [u8; 16]) -> Result<()> {
        if !self.is_unbound() {
            return fail("this repository already has an id");
        }
        if id == [0; 16] {
            return fail("a repository id cannot be all zero");
        }
        write_atomic(&self.oxy.join("id"), hex::encode(id).as_bytes())?;
        fs::remove_file(self.oxy.join(UNBOUND_MARKER))?;
        Ok(())
    }

    /// Whether this repository has no working copy (`init_bare`).
    pub fn is_bare(&self) -> bool {
        self.bare
    }

    /// Open the repository containing `start` (searching upwards).
    pub fn open(start: &Path) -> Result<Repo> {
        let start = start.canonicalize()?;
        let mut dir: Option<&Path> = Some(&start);
        while let Some(d) = dir {
            let oxy = d.join(OXY_DIR);
            if oxy.is_file() {
                // A secondary workspace (spec §12.2): the file points at the repository.
                let text = fs::read_to_string(&oxy)?;
                let Some((shared, workspace)) = workspace::parse_workspace_file(&text) else {
                    return fail(format!("{} is not a valid workspace file", oxy.display()));
                };
                let format = fs::read_to_string(shared.join("format"))?;
                if format != FORMAT_LINE {
                    return fail(format!("unsupported repository format: {}", format.trim()));
                }
                let config = Config::load(Some(&default_user_config()), &shared.join("config"))?;
                let store = Store::open(shared.join("objects"))?;
                let mut repo = Repo {
                    last_change: Default::default(),
                    at_op: None,
                    git_export_skipped: Default::default(),
                    manifest_cache: Default::default(),
                    listing_cache: Default::default(),
                    untracked_nested: Default::default(),
                    root: d.to_path_buf(),
                    oxy: shared,
                    store,
                    workspace,
                    bare: false,
                    keys_dir: default_keys_dir(),
                    platform: crate::portable::Platform::current(),
                    user_config: default_user_config(),
                    config,
                };
                repo.apply_workspace_bookmark();
                return Ok(repo);
            }
            if oxy.is_dir() {
                let format = fs::read_to_string(oxy.join("format"))?;
                let config = Config::load(Some(&default_user_config()), &oxy.join("config"))?;
                let bare = oxy.join(BARE_MARKER).is_file();
                let workspace = if bare {
                    WorkspaceId::ZERO
                } else {
                    WorkspaceId::from_hex(fs::read_to_string(oxy.join("workspace"))?.trim())
                        .ok_or_else(|| RepoError::Msg("corrupt .oxy/workspace".into()))?
                };
                let store = Store::open(oxy.join("objects"))?;
                let mut repo = Repo {
                    last_change: Default::default(),
                    at_op: None,
                    git_export_skipped: Default::default(),
                    manifest_cache: Default::default(),
                    listing_cache: Default::default(),
                    untracked_nested: Default::default(),
                    root: d.to_path_buf(),
                    oxy,
                    store,
                    workspace,
                    bare,
                    keys_dir: default_keys_dir(),
                    platform: crate::portable::Platform::current(),
                    user_config: default_user_config(),
                    config,
                };
                if !bare {
                    repo.apply_workspace_bookmark();
                }
                if format == FORMAT1_LINE {
                    repo.migrate_format1()?;
                } else if format != FORMAT_LINE {
                    return fail(format!("unsupported repository format: {}", format.trim()));
                }
                return Ok(repo);
            }
            dir = d.parent();
        }
        fail(format!(
            "no oxy repository found at or above {}; start one with `oxy init` or `oxy clone URL`",
            start.display()
        ))
    }

    /// Rewrite a format-1 operation log, whose bookmarks were head lists,
    /// into format 2, where each bookmark names a `state` object.
    fn migrate_format1(&self) -> Result<()> {
        let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;
        let head = ObjectRef::from_hex(hex.trim())
            .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))?;
        // Collect the chain root-first.
        let mut chain: Vec<Format1Op> = Vec::new();
        let mut cur = Some(head);
        while let Some(r) = cur {
            let op = Op::decode_format1(&self.store.get_typed(&r, ObjectType::Op)?)?;
            cur = op.parents.first().copied();
            chain.push(op);
        }
        chain.reverse();
        let mut new_parent: Option<ObjectRef> = None;
        for old in chain {
            let mut view = View::default();
            for (name, heads) in &old.view.bookmarks {
                view.bookmarks.insert(
                    name.clone(),
                    self.store.put_state(&StateObject::new(heads.clone()))?,
                );
            }
            view.workspaces = old.view.workspaces.clone();
            let op = Op {
                parents: new_parent.into_iter().collect(),
                hlc: old.hlc,
                command: old.command,
                view,
            };
            new_parent = Some(self.store.put(ObjectType::Op, &op.encode())?);
        }
        write_atomic(
            &self.oxy.join("ops").join("head"),
            new_parent.unwrap().to_hex().as_bytes(),
        )?;
        write_atomic(&self.oxy.join("format"), FORMAT_LINE.as_bytes())?;
        Ok(())
    }

    /// Pack automatically once many loose objects have accumulated.
    pub(crate) fn maybe_autopack(&self) -> Result<()> {
        if self.store.loose_count()? > AUTOPACK_THRESHOLD {
            self.store.pack_loose()?;
            self.store.merge_packs()?;
        }
        Ok(())
    }

    /// The repository ID (spec §14.4). Repositories made before IDs existed get one on first use.
    pub fn id(&self) -> Result<[u8; 16]> {
        if self.is_unbound() {
            return Ok([0; 16]);
        }
        let path = self.oxy.join("id");
        match fs::read_to_string(&path) {
            Ok(h) => hex::decode(h.trim())
                .ok()
                .and_then(|v| v.try_into().ok())
                .ok_or_else(|| RepoError::Msg("corrupt .oxy/id".into())),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                let mut id = [0u8; 16];
                rand::rngs::OsRng.fill_bytes(&mut id);
                write_atomic(&path, hex::encode(id).as_bytes())?;
                Ok(id)
            }
            Err(e) => Err(e.into()),
        }
    }

    pub fn root(&self) -> &Path {
        &self.root
    }
    pub fn store(&self) -> &Store {
        &self.store
    }
    pub fn workspace(&self) -> WorkspaceId {
        self.workspace
    }

    /// Use a different user directory for keys and the user config file
    /// (tests, or `$OXY_HOME`). Reloads the effective configuration.
    /// Act as if the working copy were on another platform's file system
    /// (tests of Windows and macOS behaviour on any host).
    pub fn with_platform(mut self, platform: crate::portable::Platform) -> Repo {
        self.platform = platform;
        self
    }

    pub fn platform(&self) -> crate::portable::Platform {
        self.platform
    }

    pub fn with_home(mut self, dir: impl Into<PathBuf>) -> Repo {
        let dir = dir.into();
        self.keys_dir = dir.join("keys");
        self.user_config = dir.join("config.toml");
        if let Ok(c) = Config::load(Some(&self.user_config), &self.oxy.join("config")) {
            self.config = c;
        }
        self.apply_workspace_bookmark();
        self
    }

    /// The workspace's current bookmark (spec §10.1) overrides the
    /// configured default when `oxy bookmark switch` has recorded one.
    fn apply_workspace_bookmark(&mut self) {
        if let Ok(b) = fs::read_to_string(
            self.oxy
                .join("workspaces")
                .join(self.workspace.to_hex())
                .join("bookmark"),
        ) {
            let b = b.trim();
            if !b.is_empty() {
                self.config.bookmark = b.to_string();
            }
        }
    }

    pub fn user_config_path(&self) -> &Path {
        &self.user_config
    }

    /// Set a key in the user file and refresh the effective configuration.
    pub fn set_user_config(&mut self, key: &str, value: &str) -> Result<()> {
        Config::set_user(&self.user_config, key, value)?;
        self.config = Config::load(Some(&self.user_config), &self.oxy.join("config"))?;
        Ok(())
    }

    pub fn save_config(&self) -> Result<()> {
        self.config.save(&self.oxy.join("config"))
    }

    fn author(&self) -> KeyId {
        self.config.author_key.unwrap_or(ObjectRef::ZERO)
    }
}

impl Repo {
    fn new_working_copy(&self, deps: Vec<ChangeHash>, message: &str) -> Change {
        let mut c = Change {
            version: crate::objects::CHANGE_VERSION,
            change_id: ChangeId::random(),
            deps,
            author: self.author(),
            hlc: Hlc::now(),
            message: message.into(),
            tree_ops: vec![],
            edits: vec![],
            kind: ChangeKind::WorkingCopy,
        };
        c.normalize();
        c
    }

    // ------------------------------------------------------------ operation log

    pub fn head_op(&self) -> Result<(ObjectRef, Op)> {
        if let Some(r) = self.at_op {
            return Ok((r, Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?));
        }
        let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;
        let r = ObjectRef::from_hex(hex.trim())
            .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))?;
        let op = Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?;
        Ok((r, op))
    }

    pub fn view(&self) -> Result<View> {
        Ok(self.head_op()?.1.view)
    }

    fn record_op(&self, parents: &[ObjectRef], command: &str, view: View) -> Result<ObjectRef> {
        if self.at_op.is_some() {
            return fail(
                "--at-op shows the repository as it was; this command would change it (run it without --at-op)",
            );
        }
        let op = Op {
            parents: parents.to_vec(),
            hlc: Hlc::now(),
            command: command.into(),
            view,
        };
        let r = self.store.put(ObjectType::Op, &op.encode())?;
        // The head is the commit point: everything it names must be on disk first.
        self.store.flush()?;
        // Compare and swap under a short lock (spec §12.1): when another
        // process recorded an operation since this one read the head, the
        // two are merged instead of one overwriting the other.
        let _lock = self.lock_ops()?;
        let current = match parents.first() {
            // The first operation (init): there is no head yet.
            None => None,
            Some(_) => Some(self.head_ref()?),
        };
        let Some(current) = current.filter(|c| Some(c) != parents.first()) else {
            write_atomic(&self.oxy.join("ops").join("head"), r.to_hex().as_bytes())?;
            return Ok(r);
        };
        let base = match parents.first() {
            Some(p) => Op::decode(&self.store.get_typed(p, ObjectType::Op)?)?.view,
            None => View::default(),
        };
        let theirs = Op::decode(&self.store.get_typed(&current, ObjectType::Op)?)?.view;
        let merged = Op {
            parents: vec![current, r],
            hlc: Hlc::now(),
            command: format!("{command} (merged with a concurrent operation)"),
            view: self.merge_views(&base, &op.view, &theirs)?,
        };
        let m = self.store.put(ObjectType::Op, &merged.encode())?;
        self.store.flush()?;
        write_atomic(&self.oxy.join("ops").join("head"), m.to_hex().as_bytes())?;
        Ok(m)
    }

    /// The operation head's lock, held only while the head pointer moves.
    fn lock_ops(&self) -> Result<FileLock> {
        fs::create_dir_all(self.oxy.join("ops"))?;
        FileLock::acquire(
            self.oxy.join("ops").join("lock"),
            "another oxy process holds the operation log",
        )
    }

    fn head_ref(&self) -> Result<ObjectRef> {
        let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;
        ObjectRef::from_hex(hex.trim())
            .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))
    }

    /// Three-way merge of two views that both started from `base` (spec
    /// §12.1). A reference only one side changed takes that side's value.
    /// When both changed it differently: a bookmark keeps both sides'
    /// heads (it may then have several), a change beats a deletion, a
    /// workspace changed by both is an error (only one of them matches the
    /// files on disk), and otherwise this process's value wins for remote
    /// references and the policy, the first value for a tag (tags never
    /// move).
    fn merge_views(&self, base: &View, ours: &View, theirs: &View) -> Result<View> {
        fn pick<K: Ord + Clone, V: Clone + PartialEq>(
            b: &BTreeMap<K, V>,
            o: &BTreeMap<K, V>,
            t: &BTreeMap<K, V>,
            mut both: impl FnMut(&V, &V) -> Result<V>,
        ) -> Result<BTreeMap<K, V>> {
            let keys: BTreeSet<&K> = b.keys().chain(o.keys()).chain(t.keys()).collect();
            let mut out = BTreeMap::new();
            for k in keys {
                let (bv, ov, tv) = (b.get(k), o.get(k), t.get(k));
                let v = if ov == tv || tv == bv {
                    ov.cloned()
                } else if ov == bv {
                    tv.cloned()
                } else {
                    match (ov, tv) {
                        (Some(x), Some(y)) => Some(both(x, y)?),
                        (Some(x), None) | (None, Some(x)) => Some(x.clone()),
                        (None, None) => None,
                    }
                };
                if let Some(v) = v {
                    out.insert(k.clone(), v);
                }
            }
            Ok(out)
        }
        let bookmarks = pick(
            &base.bookmarks,
            &ours.bookmarks,
            &theirs.bookmarks,
            |x, y| {
                let mut heads = self.store.get_state(x)?.heads;
                heads.extend(self.store.get_state(y)?.heads);
                let heads = self.load_state(&heads)?.maximal(&heads);
                Ok(self.store.put_state(&StateObject::new(heads))?)
            },
        )?;
        // Two operations that both replaced one workspace's working copy
        // cannot both be right: the files on disk match only one of them.
        // The working-copy lock keeps this from happening; should it happen
        // anyway, this operation is not recorded.
        let workspaces = pick(
            &base.workspaces,
            &ours.workspaces,
            &theirs.workspaces,
            |_, _| {
                fail(
                    "another oxy process changed this working copy at the same time; \
                     this command was not recorded: check `oxy status` and run it again",
                )
            },
        )?;
        let tags = pick(&base.tags, &ours.tags, &theirs.tags, |_, y| Ok(*y))?;
        let empty = BTreeMap::new();
        let names: BTreeSet<&String> = base
            .remotes
            .keys()
            .chain(ours.remotes.keys())
            .chain(theirs.remotes.keys())
            .collect();
        let mut remotes = BTreeMap::new();
        for n in names {
            let m = pick(
                base.remotes.get(n).unwrap_or(&empty),
                ours.remotes.get(n).unwrap_or(&empty),
                theirs.remotes.get(n).unwrap_or(&empty),
                |x, _| Ok(*x),
            )?;
            if !m.is_empty() {
                remotes.insert(n.clone(), m);
            }
        }
        let policy = if ours.policy == theirs.policy || theirs.policy == base.policy {
            ours.policy
        } else if ours.policy == base.policy {
            theirs.policy
        } else {
            ours.policy.or(theirs.policy)
        };
        self.store.flush()?;
        Ok(View {
            bookmarks,
            workspaces,
            remotes,
            policy,
            tags,
        })
    }

    pub fn op_log(&self) -> Result<Vec<(ObjectRef, Op)>> {
        let mut out = Vec::new();
        let (mut r, mut op) = self.head_op()?;
        loop {
            let next = op.parents.first().copied();
            out.push((r, op));
            match next {
                Some(p) => {
                    r = p;
                    op = Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?;
                }
                None => break,
            }
        }
        Ok(out)
    }

    /// Undo the last user command (spec §12.1). Operations that only recorded
    /// a working-copy snapshot are skipped, since the files are still on disk
    /// and the next snapshot recreates them, and a command's own checkout
    /// (`… (checkout)`) goes with the command. When the working copy is
    /// clean and was clean after the operation undone to, its files follow
    /// too, since nothing on disk can be lost; otherwise they stay as they
    /// are (an undone commit's edits come back as uncommitted work).
    pub fn undo(&self) -> Result<Op> {
        let _wc = self.lock_working_copy()?;
        let clean_now =
            self.clean_by_manifest()? || self.snapshot_status("oxy undo")?.statuses.is_empty();
        let (read_head, mut op) = self.head_op()?;
        let wc_now = op.view.workspaces.get(&self.workspace).copied();
        let up = |op: Op| -> Result<Option<Op>> {
            match op.parents.first() {
                Some(p) => Ok(Some(Op::decode(&self.store.get_typed(p, ObjectType::Op)?)?)),
                None => Ok(None),
            }
        };
        loop {
            while op.is_snapshot() {
                match up(op.clone())? {
                    Some(p) => op = p,
                    None => break,
                }
            }
            if !op.command.ends_with(" (checkout)") {
                break;
            }
            match up(op.clone())? {
                Some(p) => op = p,
                None => break,
            }
        }
        let Some(parent) = op.parents.first() else {
            return fail("nothing to undo");
        };
        let parent_op = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?;
        {
            let _lock = self.lock_ops()?;
            if self.head_ref()? != read_head {
                return fail("another oxy command ran while undoing; run undo again");
            }
            write_atomic(
                &self.oxy.join("ops").join("head"),
                parent.to_hex().as_bytes(),
            )?;
        }
        self.drop_manifest();
        if let Some(then) = parent_op.view.workspaces.get(&self.workspace) {
            if clean_now && Some(*then) != wc_now {
                let wc = self.store.get_change(then)?;
                if wc.edits.is_empty() && wc.tree_ops.is_empty() {
                    let base = self.load_state_history(&wc.deps)?.maximal(&wc.deps);
                    // The files differ from the restored working copy by
                    // exactly what the undone command did: nothing else is
                    // on disk, as the working copy was clean.
                    self.checkout(&base, true, "oxy undo (checkout)")?;
                }
            }
        }
        Ok(parent_op)
    }

    /// The last operation that was not a snapshot.
    pub fn last_command(&self) -> Result<Op> {
        let (_, mut op) = self.head_op()?;
        while op.is_snapshot() {
            let Some(parent) = op.parents.first() else {
                break;
            };
            op = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?;
        }
        Ok(op)
    }

    // ------------------------------------------------------------ states

    /// Load the dependency closure of `heads`, through a checkpoint when one applies.
    pub fn load_state(&self, heads: &[ChangeHash]) -> Result<State> {
        if let Some(st) = self.load_state_checkpointed(heads)? {
            return Ok(st);
        }
        self.load_state_full(heads)
    }

    /// The state of `heads` for its history (changes, tree, metadata) with
    /// no file graph built: much cheaper than [`Repo::load_state`] for log
    /// and friends. Build the graphs a command needs with
    /// [`Repo::load_graphs_of`].
    pub fn load_state_history(&self, heads: &[ChangeHash]) -> Result<State> {
        if let Some(st) = self.load_state_checkpointed_as(heads, true)? {
            return Ok(st);
        }
        self.load_state_replayed(heads, State::new_history_only())
    }

    /// Build these file graphs in a history-only state.
    pub fn load_graphs_of(&self, state: &mut State, nodes: &[crate::ids::NodeId]) -> Result<()> {
        Ok(state.load_graphs(
            nodes,
            |refs| {
                self.read_blobs(refs).map_err(|e| {
                    crate::objects::ObjectError::Schema(format!("graph and text blobs: {e}"))
                })
            },
            |h| self.change_bytes(h),
        )?)
    }

    /// The graphs of `nodes` of a history-only state, built beside it (see
    /// [`State::build_graphs_detached`]).
    pub fn build_graphs_detached(
        &self,
        state: &State,
        nodes: &[crate::ids::NodeId],
    ) -> Result<HashMap<crate::ids::NodeId, crate::graph::FileGraph>> {
        Ok(state
            .build_graphs_detached(
                nodes,
                |refs| {
                    self.read_blobs(refs).map_err(|e| {
                        crate::objects::ObjectError::Schema(format!("graph and text blobs: {e}"))
                    })
                },
                |h| self.change_bytes(h),
            )?
            .into_iter()
            .collect())
    }

    /// The contents of texts [`State::single_text`] gave: blobs read in
    /// one go, inline ones from their changes.
    pub fn single_texts(&self, texts: &[crate::objects::TextRef]) -> Result<Vec<Vec<u8>>> {
        use crate::objects::TextRef;
        let refs: Vec<ObjectRef> = texts
            .iter()
            .filter_map(|t| match t {
                TextRef::Blob(r, _, _) => Some(*r),
                TextRef::Inline { .. } => None,
            })
            .collect();
        let mut blobs = self.read_blobs(&refs)?.into_iter();
        let mut out = Vec::with_capacity(texts.len());
        for t in texts {
            let lines = match t {
                TextRef::Blob(_, count, form) => {
                    let content = blobs
                        .next()
                        .ok_or_else(|| RepoError::Msg("text blobs: wrong count".into()))?;
                    crate::objects::text_lines(&content, *count, *form)?
                }
                TextRef::Inline { change, start, end } => {
                    let bytes = self.change_bytes(change)?;
                    let range = bytes.get(*start as usize..*end as usize).ok_or_else(|| {
                        RepoError::Msg("an inline insert lies outside its change".into())
                    })?;
                    crate::objects::inline_lines(range)?
                }
            };
            out.push(lines.bytes().to_vec());
        }
        Ok(out)
    }

    /// Build every graph a history-only state still lacks, making it full.
    pub fn load_all_graphs(&self, state: &mut State) -> Result<()> {
        Ok(state.load_all_graphs(
            |refs| {
                self.read_blobs(refs).map_err(|e| {
                    crate::objects::ObjectError::Schema(format!("graph and text blobs: {e}"))
                })
            },
            |h| self.change_bytes(h),
        )?)
    }

    /// A change's encoding, for building graphs of its inline inserts. The
    /// last few read are kept: a checkout building graphs a batch at a
    /// time reads the same large changes for every batch, often in turn.
    fn change_bytes(
        &self,
        h: &ChangeHash,
    ) -> std::result::Result<std::sync::Arc<Vec<u8>>, crate::objects::ObjectError> {
        let mut cache = self.last_change.borrow_mut();
        if let Some(i) = cache.iter().position(|(k, _)| k == h) {
            let hit = cache.remove(i);
            let b = hit.1.clone();
            cache.push(hit);
            return Ok(b);
        }
        let b = std::sync::Arc::new(self.store.get_typed(h, ObjectType::Change).map_err(|e| {
            crate::objects::ObjectError::Schema(format!("change {}: {e}", h.short(12)))
        })?);
        cache.push((*h, b.clone()));
        // The last few, within a memory budget; the newest always stays.
        while cache.len() > CHANGE_CACHE_COUNT
            || (cache.len() > 1
                && cache.iter().map(|(_, b)| b.len()).sum::<usize>() > CHANGE_CACHE_BYTES)
        {
            cache.remove(0);
        }
        Ok(b)
    }

    /// Load the dependency closure of `heads` change by change.
    pub fn load_state_full(&self, heads: &[ChangeHash]) -> Result<State> {
        self.load_state_replayed(heads, State::new())
    }

    fn load_state_replayed(&self, heads: &[ChangeHash], mut state: State) -> Result<State> {
        let history = state.is_history_only();
        let mut stack: Vec<ChangeHash> = heads.to_vec();
        let mut seen: BTreeSet<ChangeHash> = BTreeSet::new();
        {
            let _t = crate::timing::phase("load: read and add changes");
            let (mut read, mut add, mut fetch) = (
                std::time::Duration::ZERO,
                std::time::Duration::ZERO,
                std::time::Duration::ZERO,
            );
            while let Some(h) = stack.pop() {
                // A change the state holds brings its history with it.
                if !seen.insert(h) || state.contains(&h) {
                    continue;
                }
                let t0 = std::time::Instant::now();
                // A history-only state reads text only for the graphs it
                // builds; either way the hash is the one asked for.
                let (c, texts) = if history {
                    let bytes = self.store.get_typed(&h, ObjectType::Change)?;
                    fetch += t0.elapsed();
                    Change::decode_skeleton(&bytes, h)
                        .map_err(|e| crate::store::StoreError::Corrupt(h, e.to_string()))?
                } else {
                    (self.store.get_change(&h)?, Vec::new())
                };
                read += t0.elapsed();
                stack.extend(c.deps.iter().copied());
                let t1 = std::time::Instant::now();
                state.add_known(h, c, texts);
                add += t1.elapsed();
            }
            crate::timing::record("load:   of which reading", read);
            crate::timing::record("load:     of which fetching", fetch);
            crate::timing::record("load:   of which applying", add);
        }
        let _t = crate::timing::phase("load: validate");
        state.validate()?;
        Ok(state)
    }

    pub fn working_copy(&self) -> Result<(ChangeHash, Change)> {
        if self.bare {
            return fail("this is a bare repository: it has no working copy");
        }
        let view = self.view()?;
        let h = *view
            .workspaces
            .get(&self.workspace)
            .ok_or_else(|| RepoError::Msg("this workspace is not in the current view".into()))?;
        Ok((h, self.store.get_change(&h)?))
    }

    /// Heads of a bookmark's state (spec §10.1).
    pub fn heads_of(&self, view: &View, bookmark: &str) -> Result<Vec<ChangeHash>> {
        match view.bookmarks.get(bookmark) {
            Some(st) => Ok(self.store.get_state(st)?.heads),
            None => Ok(vec![]),
        }
    }

    pub fn bookmark_heads(&self) -> Result<Vec<ChangeHash>> {
        self.heads_of(&self.view()?, &self.config.bookmark)
    }

    // ------------------------------------------------------------ blobs

    fn blob_for(content: &[u8]) -> Blob {
        Blob::of(content)
    }

    pub fn blob_ref(content: &[u8]) -> ObjectRef {
        Self::blob_for(content).hash()
    }

    pub fn store_blob(&self, content: &[u8]) -> Result<ObjectRef> {
        for c in crate::chunk::chunks(content) {
            self.store.put_chunk(c)?;
        }
        Ok(self
            .store
            .put(ObjectType::Blob, &Self::blob_for(content).encode())?)
    }

    /// Read many blobs at once (checkpoint graphs): the blob objects, then
    /// all their chunks, are read in parallel; a blob the store lacks goes
    /// through [`Repo::read_blob`], which may fetch it lazily.
    pub fn read_blobs(&self, refs: &[ObjectRef]) -> Result<Vec<Vec<u8>>> {
        let mut blobs: Vec<Option<Blob>> = Vec::with_capacity(refs.len());
        for (r, got) in refs.iter().zip(self.store.get_many(refs)) {
            blobs.push(match got {
                Ok((ObjectType::Blob, body)) => Some(Blob::decode(&body)?),
                Ok((ty, _)) => return fail(format!("{} is a {ty:?}, not a blob", r.short(12))),
                Err(crate::store::StoreError::NotFound(_)) => None,
                Err(e) => return Err(e.into()),
            });
        }
        let chunk_refs: Vec<ObjectRef> = blobs
            .iter()
            .flatten()
            .flat_map(|b| b.chunks.iter().copied())
            .collect();
        let mut chunks = self.store.get_many(&chunk_refs).into_iter();
        let mut out = Vec::with_capacity(refs.len());
        for (r, blob) in refs.iter().zip(blobs) {
            match blob {
                None => out.push(self.read_blob(r)?),
                Some(b) => {
                    let mut content = Vec::with_capacity(b.length as usize);
                    for c in &b.chunks {
                        match chunks.next() {
                            Some(Ok((ObjectType::Chunk, body))) => content.extend(body),
                            Some(Ok(_)) => return fail(format!("{} is not a chunk", c.short(12))),
                            Some(Err(e)) => return Err(e.into()),
                            None => return fail("chunk list ended early"),
                        }
                    }
                    out.push(content);
                }
            }
        }
        Ok(out)
    }

    /// Read a blob's content, fetching it from the remote that withheld it
    /// when it is lazily absent (spec §13.5).
    pub fn read_blob(&self, r: &ObjectRef) -> Result<Vec<u8>> {
        if !self.store.has(r) && self.lazy_remote(r).is_some() {
            self.fetch_lazy_blob(r)?;
        }
        let blob = Blob::decode(&self.store.get_typed(r, ObjectType::Blob)?)?;
        let mut out = Vec::with_capacity(blob.length as usize);
        for c in &blob.chunks {
            out.extend(self.store.get_typed(c, ObjectType::Chunk)?);
        }
        Ok(out)
    }

    // ------------------------------------------------------------ working copy

    /// Attributes from the `.oxyattributes` files in the working directory.
    pub fn disk_attributes(&self) -> crate::attributes::Attributes {
        // Nested repositories and ignored paths are left out, as in a walk.
        let listing = walk::list(&self.root, &|p: &str| is_attributes(p)).unwrap_or_default();
        let files: Vec<(String, Vec<u8>)> = listing
            .files
            .into_iter()
            .filter_map(|(rel, path, _)| Some((rel, fs::read(path).ok()?)))
            .collect();
        crate::attributes::Attributes::from_files(
            files.iter().map(|(p, c)| (p.as_str(), c.as_slice())),
        )
    }

    /// Folders holding a repository of their own that the last look at the
    /// working copy left out, not being subrepos (oxygen#43).
    pub fn untracked_nested(&self) -> Vec<String> {
        self.untracked_nested.borrow().clone()
    }

    /// Attributes from a materialized state's `.oxyattributes` files.
    pub fn state_attributes(snapshot: &Snapshot) -> crate::attributes::Attributes {
        crate::attributes::Attributes::from_files(snapshot.files.iter().filter_map(|(p, f)| {
            match &f.content {
                FileContent::Text { bytes, .. }
                    if p == ".oxyattributes" || p.ends_with("/.oxyattributes") =>
                {
                    Some((p.as_str(), bytes.as_slice()))
                }
                _ => None,
            }
        }))
    }

    /// Store a blob, raw when the path is `nocompress`.
    pub fn store_blob_for(&self, content: &[u8], nocompress: bool) -> Result<ObjectRef> {
        if !nocompress {
            return self.store_blob(content);
        }
        for c in crate::chunk::chunks(content) {
            self.store.put_chunk_nocompress(c)?;
        }
        Ok(self
            .store
            .put(ObjectType::Blob, &Self::blob_for(content).encode())?)
    }

    /// This workspace's state index (oxygen#41): its working copy's base,
    /// saved so a snapshot reads it rather than replaying history.
    /// The state index of the state with these heads: this workspace keeps
    /// the last few it used, named by their heads, so moving between two
    /// states finds both.
    fn state_index_path(&self, heads: &[ChangeHash]) -> PathBuf {
        let mut heads = heads.to_vec();
        heads.sort();
        let mut h = blake3::Hasher::new();
        for c in &heads {
            h.update(&c.0);
        }
        self.oxy.join("index").join(format!(
            "{}{}",
            self.state_index_prefix(),
            &h.finalize().to_hex()[..16]
        ))
    }

    fn state_index_prefix(&self) -> String {
        format!("state-{}-", self.workspace.to_hex())
    }

    /// This workspace's state indexes, newest first.
    fn state_indexes(&self) -> Vec<PathBuf> {
        let prefix = self.state_index_prefix();
        let mut found: Vec<(std::time::SystemTime, PathBuf)> = fs::read_dir(self.oxy.join("index"))
            .into_iter()
            .flatten()
            .flatten()
            .filter(|e| e.file_name().to_string_lossy().starts_with(&prefix))
            .filter_map(|e| Some((e.metadata().ok()?.modified().ok()?, e.path())))
            .collect();
        found.sort_by_key(|f| std::cmp::Reverse(f.0));
        found.into_iter().map(|(_, p)| p).collect()
    }

    /// The history-only state of `deps` (as [`Repo::load_state_history`]),
    /// from this workspace's state index when it holds that history: every
    /// one of `deps` is in it and their heads are its heads. Otherwise it
    /// is loaded from history and indexed for next time.
    pub fn load_state_indexed(&self, deps: &[ChangeHash]) -> Result<State> {
        if let Some((st, heads)) = self.read_state_index(deps) {
            let mut have = st.maximal(deps);
            have.sort();
            if deps.iter().all(|d| st.contains(d)) && have == heads {
                #[cfg(test)]
                assert_eq!(
                    st.write_index(),
                    self.load_state_history(deps)?.write_index(),
                    "a state index holds the state history gives"
                );
                return Ok(st);
            }
            // An index a few changes behind (a commit, a pull since): add
            // what it lacks, as long as all it holds is history of `deps`.
            // Anything amiss (a change beyond a shallow boundary) takes the
            // long way instead.
            let (indexed, started) = (st.len(), std::time::Instant::now());
            if let Ok(st) = self.load_state_replayed(deps, st) {
                let mut all = deps.to_vec();
                all.extend(heads.iter().copied());
                let (mut within, mut want) = (st.maximal(&all), st.maximal(deps));
                within.sort();
                want.sort();
                if within == want {
                    #[cfg(test)]
                    self.assert_same_state(&st, &self.load_state_history(deps)?);
                    // Writing the index costs far more than replaying a
                    // few small changes: it is rewritten once it falls
                    // behind, or when replaying took long (large changes).
                    if st.len() - indexed > INDEX_REPLAY_LIMIT
                        || started.elapsed() > INDEX_REPLAY_TIME
                    {
                        self.write_state_index(deps, &st);
                    }
                    return Ok(st);
                }
            }
        }
        let st = self.load_state_history(deps)?;
        self.write_state_index(deps, &st);
        Ok(st)
    }

    /// Two history-only states hold the same changes and render the same
    /// files. Their indexes may differ: one loaded from a checkpoint keeps
    /// the checkpoint's graphs as the bases of its files' histories.
    #[cfg(test)]
    fn assert_same_state(&self, a: &State, b: &State) {
        let ids = |s: &State| s.changes().map(|(h, _)| *h).collect::<BTreeSet<_>>();
        assert_eq!(ids(a), ids(b), "the same changes");
        let files = |s: &State| {
            let mut s = s.clone();
            self.load_all_graphs(&mut s).unwrap();
            s.materialize().files
        };
        assert_eq!(files(a), files(b), "the same files");
    }

    /// The state index file: its heads, a hash of its changes block and one
    /// of the rest, then the body [`State::write_index`] gives, whose first
    /// block is the changes. Hashed in two parts so the log can read and
    /// check the changes alone.
    /// The index of `deps` if there is one, else the newest, which the
    /// caller may bring up to date.
    fn read_state_index(&self, deps: &[ChangeHash]) -> Option<(State, Vec<ChangeHash>)> {
        let _t = crate::timing::phase("load: state index");
        let exact = self.state_index_path(deps);
        let bytes = match fs::read(&exact) {
            Ok(b) => {
                // Used now: the newest, so the last to be dropped.
                let _ = fs::File::options()
                    .write(true)
                    .open(&exact)
                    .and_then(|f| f.set_modified(std::time::SystemTime::now()));
                b
            }
            Err(_) => fs::read(self.state_indexes().first()?).ok()?,
        };
        let mut r = crate::flat::Reader::new(&bytes);
        let mut heads = Vec::new();
        for _ in 0..r.count().ok()? {
            heads.push(r.object().ok()?);
        }
        let (changes_sum, rest_sum) = (r.object().ok()?, r.object().ok()?);
        let body = &bytes[4 + 32 * heads.len() + 64..];
        let split = 4 + r.count().ok()?;
        // Checked whole, so a part decoded later (the tree's maps, a
        // file's history) can trust its bytes.
        let (changes, rest) = (body.get(..split)?, body.get(split..)?);
        let mut h = blake3::Hasher::new();
        h.update_rayon(rest);
        if *blake3::hash(changes).as_bytes() != changes_sum.0
            || *h.finalize().as_bytes() != rest_sum.0
        {
            return None;
        }
        Some((State::read_index(body).ok()?, heads))
    }

    /// The changes of this workspace's state index (the closure of its
    /// heads), read without the rest of the index.
    pub(crate) fn read_state_index_changes(
        &self,
    ) -> Option<std::collections::HashMap<ChangeHash, std::sync::Arc<crate::state::ChangeMeta>>>
    {
        use std::io::Read;
        let _t = crate::timing::phase("load: state index changes");
        let mut f = fs::File::open(self.state_indexes().first()?).ok()?;
        let mut word = [0u8; 4];
        f.read_exact(&mut word).ok()?;
        let n = u32::from_le_bytes(word) as usize;
        let mut head = vec![0u8; 32 * n + 64 + 4];
        f.read_exact(&mut head).ok()?;
        let sum = &head[32 * n..32 * n + 32];
        let len = u32::from_le_bytes(head[32 * n + 64..].try_into().unwrap()) as usize;
        let mut changes = vec![0u8; 4 + len];
        changes[..4].copy_from_slice(&head[32 * n + 64..]);
        f.read_exact(&mut changes[4..]).ok()?;
        if blake3::hash(&changes).as_bytes() != sum {
            return None;
        }
        State::read_index_changes(&changes[4..]).ok()
    }

    /// Save `st`, the history-only state of `deps`, as this workspace's
    /// state index; quietly nothing when it cannot be (it is only a cache).
    pub(crate) fn write_state_index(&self, deps: &[ChangeHash], st: &State) {
        let _t = crate::timing::phase("write: state index");
        let encoding = crate::timing::phase("write:   of which encoding");
        let Some(body) = st.write_index() else {
            return;
        };
        drop(encoding);
        #[cfg(test)]
        assert_eq!(
            State::read_index(&body).ok().and_then(|s| s.write_index()),
            Some(body.clone()),
            "a state index reads back as written"
        );
        let mut heads = st.maximal(deps);
        heads.sort();
        let mut w = crate::flat::Writer::default();
        w.count(heads.len());
        for h in &heads {
            w.object(h);
        }
        let split = 4 + u32::from_le_bytes(body[..4].try_into().unwrap()) as usize;
        w.fixed(blake3::hash(&body[..split]).as_bytes());
        let mut sum = blake3::Hasher::new();
        sum.update_rayon(&body[split..]);
        w.fixed(sum.finalize().as_bytes());
        w.fixed(&body);
        let path = self.state_index_path(&heads);
        let _ = fs::create_dir_all(self.oxy.join("index"));
        let _ = write_cache(&path, &w.0);
        // The last two used: the index of a state moved away from stays for
        // a move back. Older ones (and the unnamed one of before) go.
        for old in self.state_indexes().iter().skip(STATE_INDEXES_KEPT) {
            let _ = fs::remove_file(old);
        }
        let _ = fs::remove_file(
            self.oxy
                .join("index")
                .join(format!("state-{}", self.workspace.to_hex())),
        );
    }

    fn locks_dir(&self) -> PathBuf {
        self.oxy.join("locks")
    }

    /// Whether `path` is locked locally for editing (spec §18, advisory).
    pub fn is_locked(&self, path: &str) -> bool {
        self.locks_dir().join(path.replace('/', "%2F")).exists()
    }

    /// Take the advisory local lock on a `lockable` file and make it writable.
    pub fn lock_path(&self, path: &str) -> Result<()> {
        let attrs = self.disk_attributes();
        if !attrs.lookup(path).lockable {
            return fail(format!("{path} is not marked lockable in .oxyattributes"));
        }
        fs::create_dir_all(self.locks_dir())?;
        write_atomic(
            &self.locks_dir().join(path.replace('/', "%2F")),
            std::process::id().to_string().as_bytes(),
        )?;
        set_readonly(&self.root.join(path), false)?;
        Ok(())
    }

    /// Release the local lock and make the file read-only again.
    pub fn unlock_path(&self, path: &str) -> Result<()> {
        let _ = fs::remove_file(self.locks_dir().join(path.replace('/', "%2F")));
        if self.root.join(path).exists() {
            set_readonly(&self.root.join(path), true)?;
        }
        Ok(())
    }

    /// Whether a repository path is inside the sparse filter (spec §17).
    pub fn in_scope(&self, path: &str) -> bool {
        self.config.sparse_paths.is_empty()
            || self
                .config
                .sparse_paths
                .iter()
                .any(|p| path == p || path.starts_with(&format!("{p}/")))
    }

    fn walk_working_copy(&self) -> Result<BTreeMap<String, FsEntry>> {
        self.walk_working_copy_with(None, &BTreeMap::new(), None)
    }

    /// Walk the working copy; with a manifest, a file whose size and mtime
    /// match it is not read: an opaque file yields its known blob, and a
    /// text file whose recorded graph key equals its base graph's key in
    /// `keys` (path → key) is marked unchanged; other text files are read
    /// but keep their known hash. Files modified within two seconds of the
    /// manifest are never trusted (a racy stat).
    ///
    /// `.oxyattributes` files are always read, since their content is
    /// needed; and when any of them is not known unchanged (added, removed,
    /// edited, or changed in the base), no text file is taken as unchanged,
    /// because attributes decide how a file compares with the base.
    ///
    /// `.oxyignore` applies to untracked paths only: every path in `tracked`
    /// (the base state's files) that exists on disk is included even when
    /// an ignore rule matches it, as in Git.
    fn walk_working_copy_with(
        &self,
        cache: Option<&prompt::Manifest>,
        tracked: &BTreeMap<String, (NodeKind, Option<ObjectRef>)>,
        keys: Option<&KeyOf>,
    ) -> Result<BTreeMap<String, FsEntry>> {
        let mut out: BTreeMap<String, FsEntry>;
        let sparse = &self.config.sparse_paths;
        let in_scope = |p: &str| {
            sparse.is_empty()
                || sparse
                    .iter()
                    .any(|s| p == s || p.starts_with(&format!("{s}/")))
        };
        let walk_phase = crate::timing::phase("walk:   of which listing");
        let walk::Listing {
            files,
            symlinks,
            nested,
        } = match self.listing_cache.take() {
            Some(l) => l,
            // Ignore rules apply to untracked paths only, so the base's
            // files are not matched against them (an added file is: a rule
            // written after it was snapshotted leaves it out again).
            None => walk::list_knowing(&self.root, &in_scope, &|p: &str| tracked.contains_key(p))?,
        };
        let keys = keys.filter(|k| {
            let on_disk: HashSet<&str> = files.iter().map(|(r, _, _)| r.as_str()).collect();
            let attrs_unchanged =
                files
                    .iter()
                    .filter(|(r, _, _)| is_attributes(r))
                    .all(|(rel, _, meta)| {
                        stat_cached(rel, meta, cache)
                            .is_some_and(|e| e.graph.is_some() && e.graph == k(rel))
                    });
            let none_removed = tracked
                .keys()
                .filter(|r| is_attributes(r))
                .all(|r| on_disk.contains(r.as_str()));
            attrs_unchanged && none_removed
        });
        drop(walk_phase);
        let _read_phase = crate::timing::phase("walk:   of which stat and compare");
        // Reading files is most of a walk's cost; do it in parallel.
        {
            use rayon::prelude::*;
            let mut read: Vec<(String, FsEntry)> = files
                .par_iter()
                .map(|(rel, path, meta)| {
                    // A key is worked out only for a file the manifest
                    // knows with the same stat and a key of its own.
                    let key = keys
                        .filter(|_| !is_attributes(rel))
                        .filter(|_| {
                            stat_cached(rel, meta, cache).is_some_and(|e| e.graph.is_some())
                        })
                        .and_then(|k| k(rel));
                    Ok((rel.clone(), read_file_entry(rel, path, meta, cache, key)?))
                })
                .collect::<Result<Vec<_>>>()?;
            for (rel, e) in read.iter_mut() {
                // A cached blob the store lost (gc, a partial clone): read the file after all.
                if e.content.is_empty() && e.blob.is_some_and(|b| !self.store.has(&b)) {
                    e.content = std::sync::Arc::new(fs::read(self.root.join(rel.as_str()))?);
                    e.blob = None;
                }
            }
            // The listing is in path order, so the map is built without
            // searching.
            out = read.into_iter().collect();
            for (rel, target) in symlinks {
                out.insert(rel, FsEntry::new(NodeKind::Symlink, false, target));
            }
        }
        drop(_read_phase);
        let _rest = crate::timing::phase("walk:   of which subrepos and hidden");
        // Nested repositories (spec §7.8): registered in `.oxymodules` or
        // already tracked as subrepo nodes, they contribute their pin;
        // unregistered ones are left alone.
        let modules = self.modules_on_disk();
        let removed = self.removed_subrepos();
        for rel in &removed {
            if tracked.get(rel).map(|(k, _)| *k) != Some(NodeKind::Subrepo) {
                self.clear_removed_subrepo(rel);
            }
        }
        let mut untracked_nested = Vec::new();
        for dir in nested {
            let rel = self.rel_path(&dir);
            if !self.in_scope(&rel) || removed.contains(&rel) {
                continue;
            }
            let registered = modules.entries.get(&rel);
            if registered.is_none() && tracked.get(&rel).map(|(k, _)| *k) != Some(NodeKind::Subrepo)
            {
                untracked_nested.push(rel);
                continue;
            }
            let pin = subrepos::read_pin(&dir, registered.and_then(|m| m.bookmark.as_deref()))?;
            out.insert(rel, FsEntry::new(NodeKind::Subrepo, false, pin.encode()));
        }
        untracked_nested.sort();
        *self.untracked_nested.borrow_mut() = untracked_nested;
        // Tracked paths the ignore rules hid.
        for (rel, (kind, blob)) in missing_from(tracked, &out) {
            if !self.in_scope(rel) {
                continue;
            }
            let path = self.root.join(rel);
            if *kind == NodeKind::Subrepo {
                if removed.contains(rel) {
                    continue;
                }
                let present =
                    path.is_dir() && (path.join(OXY_DIR).exists() || path.join(".git").exists());
                if present {
                    let pin = subrepos::read_pin(
                        &path,
                        modules.entries.get(rel).and_then(|m| m.bookmark.as_deref()),
                    )?;
                    out.insert(
                        rel.clone(),
                        FsEntry::new(NodeKind::Subrepo, false, pin.encode()),
                    );
                } else if let Some(b) = blob {
                    // Not cloned (or emptied): the pin stays as recorded, as a
                    // Git gitlink does when its directory is absent.
                    out.insert(
                        rel.clone(),
                        FsEntry {
                            blob: Some(*b),
                            ..FsEntry::new(NodeKind::Subrepo, false, vec![])
                        },
                    );
                }
                continue;
            }
            let Ok(meta) = fs::symlink_metadata(&path) else {
                continue;
            };
            let ft = meta.file_type();
            if ft.is_symlink() {
                let target = fs::read_link(&path)?;
                out.insert(
                    rel.clone(),
                    FsEntry::new(
                        NodeKind::Symlink,
                        false,
                        target.to_string_lossy().as_bytes().to_vec(),
                    ),
                );
            } else if ft.is_file() {
                let e = self.file_entry(rel, &path, &meta, cache)?;
                out.insert(rel.clone(), e);
            }
        }
        Ok(out)
    }

    /// The entry for one regular file, read or served from the stat cache.
    fn file_entry(
        &self,
        rel: &str,
        path: &Path,
        meta: &fs::Metadata,
        cache: Option<&prompt::Manifest>,
    ) -> Result<FsEntry> {
        let mut e = read_file_entry(rel, path, meta, cache, None)?;
        if e.content.is_empty() && e.blob.is_some_and(|b| !self.store.has(&b)) {
            e.content = std::sync::Arc::new(fs::read(path)?);
            e.blob = None;
        }
        Ok(e)
    }

    /// The non-directory paths of a snapshot with their kinds, for the
    /// walker's `tracked` set.
    pub(crate) fn tracked_kinds(
        snapshot: &Snapshot,
    ) -> BTreeMap<String, (NodeKind, Option<ObjectRef>)> {
        snapshot
            .files
            .iter()
            .filter(|(_, f)| f.kind != NodeKind::Dir)
            .map(|(p, f)| {
                let blob = match &f.content {
                    FileContent::Opaque { blobs } => blobs.first().copied(),
                    _ => None,
                };
                (p.clone(), (f.kind, blob))
            })
            .collect()
    }

    /// A path under the root as a `/`-separated relative string.
    pub(crate) fn rel_path(&self, path: &Path) -> String {
        path.strip_prefix(&self.root)
            .unwrap_or(path)
            .components()
            .map(|c| c.as_os_str().to_string_lossy().to_string())
            .collect::<Vec<_>>()
            .join("/")
    }
}

impl Repo {
    /// Changes reachable from the bookmark and the working copy, newest first.
    pub fn log(&self) -> Result<Vec<LogEntry>> {
        let heads = self.bookmark_heads()?;
        if self.bare {
            return self.log_state(&heads);
        }
        let (wc_hash, _) = self.working_copy()?;
        let mut all = heads.clone();
        all.push(wc_hash);
        let head_set: BTreeSet<ChangeHash> = heads.into_iter().collect();
        let mut entries: Vec<LogEntry> = Vec::new();
        // A shallow clone knows changes beyond its boundary from a
        // checkpoint's metadata only: its log comes from the state.
        let metas = if self.is_shallow() {
            self.load_state_history(&all)?
                .changes()
                .map(|(h, m)| (*h, m.clone()))
                .collect()
        } else {
            self.metas_of(&all)?
        };
        for (h, meta) in metas {
            entries.push(LogEntry {
                hash: h,
                change: meta,
                is_working_copy: h == wc_hash,
                is_head: head_set.contains(&h),
            });
        }
        entries.sort_by(|a, b| {
            b.is_working_copy
                .cmp(&a.is_working_copy)
                .then(b.change.hlc.cmp(&a.change.hlc))
                .then(b.hash.cmp(&a.hash))
        });
        Ok(entries)
    }

    /// A window of `log_state`: entries `offset..offset+limit` and the total
    /// count, for paginated views.
    pub fn log_window(
        &self,
        heads: &[ChangeHash],
        offset: usize,
        limit: usize,
    ) -> Result<(Vec<LogEntry>, usize)> {
        let all = self.log_state(heads)?;
        let total = all.len();
        Ok((all.into_iter().skip(offset).take(limit).collect(), total))
    }

    /// The log of the state with these heads, newest first, with no working
    /// copy involved (for hosted repositories and any explicit state).
    pub fn log_state(&self, heads: &[ChangeHash]) -> Result<Vec<LogEntry>> {
        // The log needs change metadata only: neither graphs nor the tree.
        if self.is_shallow() {
            return Ok(self.log_of(&self.load_state_history(heads)?));
        }
        let metas = self.metas_of(heads)?;
        let depended: BTreeSet<ChangeHash> = metas
            .iter()
            .flat_map(|(_, m)| m.deps.iter().copied())
            .collect();
        let mut entries: Vec<LogEntry> = metas
            .into_iter()
            .map(|(h, meta)| LogEntry {
                hash: h,
                is_head: !depended.contains(&h),
                change: meta,
                is_working_copy: false,
            })
            .collect();
        entries.sort_by(|a, b| b.change.hlc.cmp(&a.change.hlc).then(b.hash.cmp(&a.hash)));
        Ok(entries)
    }

    /// Every change of the state with these heads and its metadata, read
    /// without its operations ([`Store::get_change_meta`]). A change beyond
    /// a shallow clone's boundary is left out.
    pub fn metas_of(
        &self,
        heads: &[ChangeHash],
    ) -> Result<Vec<(ChangeHash, crate::state::ChangeMeta)>> {
        let boundary = self.boundary();
        // The state index holds the metadata of its heads' closure: a
        // change found there is not read from the store (a change object
        // can be as large as the tree it adds).
        let indexed = self.read_state_index_changes();
        let (mut stored, mut from_store) = (0usize, std::time::Duration::ZERO);
        let mut seen: HashSet<ChangeHash> = HashSet::new();
        let mut stack: Vec<ChangeHash> = heads.to_vec();
        let mut out = Vec::new();
        while let Some(h) = stack.pop() {
            if !seen.insert(h) {
                continue;
            }
            if let Some(meta) = indexed.as_ref().and_then(|m| m.get(&h)) {
                #[cfg(test)]
                assert_eq!(
                    **meta,
                    self.store.get_change_meta(&h)?,
                    "the state index holds a change's metadata"
                );
                stack.extend(meta.deps.iter().copied());
                out.push((h, (**meta).clone()));
                continue;
            }
            let started = std::time::Instant::now();
            let meta = match self.store.get_change_meta(&h) {
                Ok(m) => m,
                Err(crate::store::StoreError::NotFound(_)) if boundary.contains(&h) => continue,
                Err(e) => return Err(e.into()),
            };
            stored += 1;
            from_store += started.elapsed();
            stack.extend(meta.deps.iter().copied());
            out.push((h, meta));
        }
        if stored > 0 && crate::timing::enabled() {
            eprintln!("timing metadata: {stored} changes read whole, {from_store:.2?}");
        }
        Ok(out)
    }

    /// The log of a state already loaded, newest first.
    /// The history of lines `first..=last` (1-based) of `path` in the
    /// state `heads` make (spec §27): every change that inserted or
    /// deleted one of their atoms, or an atom deleted from between them,
    /// newest first. Exact, as atoms keep their identity across edits.
    pub fn line_history(
        &self,
        heads: &[ChangeHash],
        path: &str,
        first: usize,
        last: usize,
    ) -> Result<Vec<LogEntry>> {
        if first == 0 || last < first {
            return fail(format!(
                "lines {first},{last}: expected FIRST,LAST from 1 with FIRST ≤ LAST"
            ));
        }
        let mut state = self.load_state_history(heads)?;
        let Some(node) = state.tree().view().entries.get(path).map(|e| e.node) else {
            return fail(format!("{path} is not in that state"));
        };
        self.load_graphs_of(&mut state, &[node])?;
        let live = state.live_lines(&node);
        if last > live.len() {
            return fail(format!(
                "{path} has {} lines; asked for {first},{last}",
                live.len()
            ));
        }
        let Some(g) = state.file(&node) else {
            return fail(format!("{path} is not a text file"));
        };
        let atoms = g.between(&live[first - 1].0, &live[last - 1].0);
        let touched = g.touching(&atoms);
        Ok(self
            .log_of(&state)
            .into_iter()
            .filter(|e| touched.contains(&e.hash))
            .collect())
    }

    pub fn log_of(&self, state: &crate::state::State) -> Vec<LogEntry> {
        let head_set: BTreeSet<ChangeHash> = state.heads().into_iter().collect();
        let mut entries: Vec<LogEntry> = state
            .changes()
            .map(|(h, meta)| LogEntry {
                hash: *h,
                change: meta.clone(),
                is_working_copy: false,
                is_head: head_set.contains(h),
            })
            .collect();
        entries.sort_by(|a, b| b.change.hlc.cmp(&a.change.hlc).then(b.hash.cmp(&a.hash)));
        entries
    }

    /// Resolve a spec to a set of heads: a bookmark's name, a remote's
    /// bookmark as `REMOTE/NAME`, a tag name, or a single change
    /// (`resolve_change`).
    pub fn resolve_heads(&self, spec: &str) -> Result<Vec<ChangeHash>> {
        let view = self.view()?;
        if let Some(st) = view.bookmarks.get(spec) {
            return Ok(self.store.get_state(st)?.heads);
        }
        if let Some((_, _, st)) = remote_bookmark(&view, spec) {
            return Ok(self.store.get_state(&st)?.heads);
        }
        if let Some(heads) = self.tag_heads(spec)? {
            return Ok(heads);
        }
        Ok(vec![self.resolve_change(spec)?])
    }

    /// Resolve a change by ChangeId prefix, `#hash` prefix, `@` (working
    /// copy), bookmark, `REMOTE/NAME` or tag (the one head of its state).
    pub fn resolve_change(&self, spec: &str) -> Result<ChangeHash> {
        if spec == "@" {
            return Ok(self.working_copy()?.0);
        }
        let view = self.view()?;
        let named = match view.bookmarks.get(spec) {
            Some(st) => Some(("bookmark", self.store.get_state(st)?.heads)),
            None => match remote_bookmark(&view, spec) {
                Some((_, _, st)) => Some(("bookmark", self.store.get_state(&st)?.heads)),
                None => self.tag_heads(spec)?.map(|h| ("tag", h)),
            },
        };
        if let Some((kind, heads)) = named {
            return match heads.len() {
                1 => Ok(heads[0]),
                n => fail(format!("{kind} {spec} has {n} heads; name a change, or use it where a state is expected")),
            };
        }
        if spec.is_empty() || spec == "#" {
            return fail("empty change specifier");
        }
        let entries = self.log()?;
        let matches: Vec<&LogEntry> = if let Some(hp) = spec.strip_prefix('#') {
            entries
                .iter()
                .filter(|e| e.hash.starts_with_hex(hp))
                .collect()
        } else {
            entries
                .iter()
                .filter(|e| e.change.change_id.starts_with_hex(spec))
                .collect()
        };
        match matches.len() {
            1 => Ok(matches[0].hash),
            0 => fail(format!("no change matches {spec:?}")),
            _ => {
                let ids: BTreeSet<ChangeId> = matches.iter().map(|e| e.change.change_id).collect();
                if ids.len() == 1 {
                    fail(format!(
                        "change {spec} is divergent: {} variants",
                        matches.len()
                    ))
                } else {
                    fail(format!("{spec:?} is ambiguous"))
                }
            }
        }
    }

    /// Human-readable description of what a change does, with line content.
    pub fn describe_change(&self, hash: &ChangeHash) -> Result<String> {
        let change = self.store.get_change(hash)?;
        let (base, base_snapshot) = self.change_base(&change)?;
        Ok(render_change(&change, &base, &base_snapshot))
    }

    /// The base state of `change` (its dependencies) and its rendering, as
    /// much as showing the change needs: the whole tree, and the graphs and
    /// contents of the files the change edits. Other text files are left
    /// unrendered ([`Snapshot::unrendered`]).
    pub fn change_base(&self, change: &Change) -> Result<(State, Snapshot)> {
        let mut base = self.load_state_history(&change.deps)?;
        let edited: std::collections::HashSet<NodeId> =
            change.edits.iter().map(|e| e.node()).collect();
        let nodes: Vec<NodeId> = edited.iter().copied().collect();
        self.load_graphs_of(&mut base, &nodes)?;
        let skip: std::collections::HashSet<NodeId> = base
            .tree()
            .view()
            .entries
            .values()
            .filter(|e| e.kind == crate::objects::NodeKind::File && !edited.contains(&e.node))
            .map(|e| e.node)
            .collect();
        let snapshot = base.materialize_except(&skip);
        Ok((base, snapshot))
    }
}

fn remove_any(path: &Path) -> std::io::Result<()> {
    let m = fs::symlink_metadata(path)?;
    if m.is_dir() {
        fs::remove_dir_all(path)
    } else {
        fs::remove_file(path)
    }
}

fn set_readonly(path: &Path, readonly: bool) -> std::io::Result<()> {
    let mut perms = fs::metadata(path)?.permissions();
    perms.set_readonly(readonly);
    fs::set_permissions(path, perms)
}

#[cfg(unix)]
fn set_executable(path: &Path, exec: bool) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt;
    let mut perms = fs::metadata(path)?.permissions();
    let mode = perms.mode();
    let want = if exec { mode | 0o111 } else { mode & !0o111 };
    if want != mode {
        perms.set_mode(want);
        fs::set_permissions(path, perms)?;
    }
    Ok(())
}
#[cfg(not(unix))]
fn set_executable(_path: &Path, _exec: bool) -> std::io::Result<()> {
    Ok(())
}

#[cfg(unix)]
fn is_executable(m: &fs::Metadata) -> bool {
    use std::os::unix::fs::PermissionsExt;
    m.permissions().mode() & 0o111 != 0
}
#[cfg(not(unix))]
fn is_executable(_m: &fs::Metadata) -> bool {
    false
}

fn op_ref(k: OpKey) -> OpRef {
    OpRef {
        change: k.0,
        index: k.1,
    }
}

fn split_path(path: &str) -> (&str, &str) {
    match path.rfind('/') {
        Some(i) => (&path[..i], &path[i + 1..]),
        None => ("", path),
    }
}

/// Whether `rel` is an attributes file (spec §7.6).
fn is_attributes(rel: &str) -> bool {
    rel == ".oxyattributes" || rel.ends_with("/.oxyattributes")
}

/// The manifest entry for `rel` when its size and mtime match it outside
/// the racy window.
fn stat_cached<'m>(
    rel: &str,
    meta: &fs::Metadata,
    cache: Option<&'m prompt::Manifest>,
) -> Option<&'m prompt::ManifestEntry> {
    let m = cache?;
    let e = m.entries.get(rel)?;
    let mtime_ns = prompt::mtime_ns(meta);
    let racy = mtime_ns + 2_000_000_000 >= m.written_ns;
    (e.size == meta.len() && e.mtime_ns == mtime_ns && !racy).then_some(e)
}

/// Read one working-copy file, or serve it from the stat cache (spec
/// §12.3). With a manifest whose stat for it is unchanged (and not racy),
/// an opaque file is not read at all (its stored blob stands in), and a
/// text file the manifest recorded equal to a base graph whose key is
/// `key`, the key the base has now, is not read either: equal keys mean
/// equal renderings, so it still equals the base. Touches no repository
/// state, so the walk runs it in parallel; the caller checks that a cached
/// blob is still in the store.
fn read_file_entry(
    rel: &str,
    path: &Path,
    meta: &fs::Metadata,
    cache: Option<&prompt::Manifest>,
    key: Option<[u8; 32]>,
) -> Result<FsEntry> {
    let exec = is_executable(meta);
    let stat = Some((meta.len(), prompt::mtime_ns(meta)));
    let mut e = match stat_cached(rel, meta, cache) {
        Some(e) if e.blob.is_some() => FsEntry {
            blob: e.blob,
            content_hash: Some(e.hash),
            ..FsEntry::new(NodeKind::File, exec, vec![])
        },
        Some(e) if key.is_some() && e.graph == key => FsEntry {
            content_hash: Some(e.hash),
            unchanged: true,
            graph_key: e.graph,
            ..FsEntry::new(NodeKind::File, exec, vec![])
        },
        Some(e) => FsEntry {
            content_hash: Some(e.hash),
            ..FsEntry::new(NodeKind::File, exec, fs::read(path)?)
        },
        None => FsEntry::new(NodeKind::File, exec, fs::read(path)?),
    };
    // The stat the manifest records: taken before the read, so a file that
    // changes meanwhile is read again next time.
    e.stat = stat;
    Ok(e)
}

/// Two maps side by side in key order: each key of either, with its value
/// in each. Linear, with no lookups, for maps of a whole tree.
pub(crate) fn join<'a, A, B>(
    a: &'a BTreeMap<String, A>,
    b: &'a BTreeMap<String, B>,
) -> impl Iterator<Item = (&'a String, Option<&'a A>, Option<&'a B>)> {
    let (mut a, mut b) = (a.iter().peekable(), b.iter().peekable());
    std::iter::from_fn(move || match (a.peek(), b.peek()) {
        (None, None) => None,
        (Some(_), None) => a.next().map(|(k, v)| (k, Some(v), None)),
        (None, Some(_)) => b.next().map(|(k, v)| (k, None, Some(v))),
        (Some((ka, _)), Some((kb, _))) => match ka.cmp(kb) {
            std::cmp::Ordering::Less => a.next().map(|(k, v)| (k, Some(v), None)),
            std::cmp::Ordering::Greater => b.next().map(|(k, v)| (k, None, Some(v))),
            std::cmp::Ordering::Equal => {
                let (k, va) = a.next()?;
                let (_, vb) = b.next()?;
                Some((k, Some(va), Some(vb)))
            }
        },
    })
}

/// The entries of `a` whose keys `b` lacks, in key order: linear, with no
/// lookups.
pub(crate) fn missing_from<'a, A, B>(
    a: &'a BTreeMap<String, A>,
    b: &BTreeMap<String, B>,
) -> Vec<(&'a String, &'a A)> {
    let mut keys = b.keys().peekable();
    a.iter()
        .filter(|(k, _)| {
            while keys.next_if(|kb| *kb < *k).is_some() {}
            keys.peek() != Some(k)
        })
        .collect()
}