//! Workspaces on disk: finding one, locking it, its stat index, saving the
//! folder into a tree, and writing a tree back into the folder.

use std::collections::BTreeMap;
use std::fs;
use std::io::Read;
use std::path::{Path, PathBuf};

use anyhow::{Context, Result, anyhow};
use rayon::prelude::*;
use vcs_core::Id;
use vcs_core::tree::{self, Entry, EntryKind, Files};

use crate::objects::Store;
use crate::sys;

pub const DIR: &str = ".vcs";

/// The workspace folder containing `start`, if any.
pub fn find(start: &Path) -> Option<PathBuf> {
    let mut dir = Some(start);
    while let Some(d) = dir {
        if d.join(DIR).join("workspace").is_file() {
            return Some(d.to_path_buf());
        }
        dir = d.parent();
    }
    None
}

pub fn read_id(root: &Path) -> Result<String> {
    Ok(fs::read_to_string(root.join(DIR).join("workspace"))?.trim().to_string())
}

pub fn write_id(root: &Path, id: &str) -> Result<()> {
    let dir = root.join(DIR);
    fs::create_dir_all(&dir)?;
    fs::write(dir.join("workspace"), format!("{id}\n"))?;
    // Keep the folder out of anything that walks the tree.
    fs::write(dir.join(".gitignore"), "*\n")?;
    Ok(())
}

/// An exclusive lock on the workspace, held while a command or the saver
/// works in it.
pub struct Lock {
    _file: fs::File,
}

pub fn lock(root: &Path) -> Result<Lock> {
    use fs4::fs_std::FileExt;
    let path = root.join(DIR).join("lock");
    let file = fs::OpenOptions::new().create(true).truncate(false).write(true).open(&path)?;
    file.lock_exclusive().context("locking the workspace")?;
    Ok(Lock { _file: file })
}

/// Like [`lock`], but gives up at once if someone else holds it.
pub fn try_lock(root: &Path) -> Result<Option<Lock>> {
    use fs4::fs_std::FileExt;
    let path = root.join(DIR).join("lock");
    let file = fs::OpenOptions::new().create(true).truncate(false).write(true).open(&path)?;
    match file.try_lock_exclusive() {
        Ok(true) => Ok(Some(Lock { _file: file })),
        Ok(false) => Ok(None),
        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(None),
        Err(e) => Err(e.into()),
    }
}

// The stat index: what each tracked file looked like when we last knew its
// blob, so unchanged files are never read again.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Stat {
    pub size: u64,
    pub mtime_ns: i64,
    pub ino: u64,
}

impl Stat {
    fn of(meta: &fs::Metadata) -> Stat {
        let (mtime_ns, ino) = sys::times(meta);
        Stat { size: meta.len(), mtime_ns, ino }
    }
}

#[derive(Debug, Clone, Copy)]
pub struct IndexEntry {
    pub entry: Entry,
    pub stat: Stat,
}

#[derive(Debug, Default, Clone)]
pub struct Index {
    pub entries: BTreeMap<String, IndexEntry>,
    /// When the index was written (ns); files changed just before it are
    /// re-read next time (git's "racy" rule, see `racy_window`).
    pub written_ns: i64,
    /// The tree these entries make, so a save that finds nothing new
    /// doesn't build it again.
    pub tree: Option<Id>,
}

const MAGIC: &[u8] = b"VCSI3\n";

impl Index {
    pub fn load(root: &Path) -> Index {
        Self::try_load(root, "index").unwrap_or_default()
    }

    /// Another index kept in the workspace's folder (a lazy workspace's tree).
    pub fn load_from(root: &Path, name: &str) -> Option<Index> {
        Self::try_load(root, name).ok()
    }

    fn try_load(root: &Path, name: &str) -> Result<Index> {
        let bytes = fs::read(root.join(DIR).join(name))?;
        if !bytes.starts_with(MAGIC) {
            return Err(anyhow!("bad index"));
        }
        let mut at = MAGIC.len();
        let take = |at: &mut usize, n: usize| -> Result<&[u8]> {
            let slice = bytes.get(*at..*at + n).ok_or_else(|| anyhow!("short index"))?;
            *at += n;
            Ok(slice)
        };
        let written_ns = i64::from_le_bytes(take(&mut at, 8)?.try_into()?);
        let tree = Id::from_slice(take(&mut at, 32)?).filter(|t| *t != Id::ZERO);
        let count = u32::from_le_bytes(take(&mut at, 4)?.try_into()?);
        // Written in order, so the map is built in one go, not one insert at a time.
        let mut list = Vec::with_capacity(count as usize);
        for _ in 0..count {
            let len = u16::from_le_bytes(take(&mut at, 2)?.try_into()?) as usize;
            let path = std::str::from_utf8(take(&mut at, len)?)?.to_string();
            let id = Id::from_slice(take(&mut at, 32)?).ok_or_else(|| anyhow!("bad index"))?;
            let kind = EntryKind::from_u8(take(&mut at, 1)?[0]).ok_or_else(|| anyhow!("bad index"))?;
            let size = u64::from_le_bytes(take(&mut at, 8)?.try_into()?);
            let mtime_ns = i64::from_le_bytes(take(&mut at, 8)?.try_into()?);
            let ino = u64::from_le_bytes(take(&mut at, 8)?.try_into()?);
            list.push((path, IndexEntry { entry: Entry { kind, id }, stat: Stat { size, mtime_ns, ino } }));
        }
        Ok(Index { entries: list.into_iter().collect(), written_ns, tree })
    }

    /// Writes the index. It's a cache (a lost one only means reading files
    /// again), so it isn't synced to disk.
    pub fn save(&mut self, root: &Path) -> Result<()> {
        self.save_as(root, "index")
    }

    pub fn save_as(&mut self, root: &Path, name: &str) -> Result<()> {
        self.written_ns = now_ns();
        let mut out = Vec::with_capacity(self.entries.len() * 80 + 64);
        out.extend_from_slice(MAGIC);
        out.extend_from_slice(&self.written_ns.to_le_bytes());
        out.extend_from_slice(&self.tree.unwrap_or(Id::ZERO).0);
        out.extend_from_slice(&(self.entries.len() as u32).to_le_bytes());
        for (path, e) in &self.entries {
            out.extend_from_slice(&(path.len() as u16).to_le_bytes());
            out.extend_from_slice(path.as_bytes());
            out.extend_from_slice(&e.entry.id.0);
            out.push(e.entry.kind as u8);
            out.extend_from_slice(&e.stat.size.to_le_bytes());
            out.extend_from_slice(&e.stat.mtime_ns.to_le_bytes());
            out.extend_from_slice(&e.stat.ino.to_le_bytes());
        }
        let tmp = root.join(DIR).join(format!("{name}.tmp{}", std::process::id()));
        fs::write(&tmp, &out)?;
        fs::rename(tmp, root.join(DIR).join(name))?;
        Ok(())
    }

    /// Whether writing `next` over this index would change anything: new
    /// entries, or entries still inside the racy second.
    pub fn same_as(&self, next: &Index) -> bool {
        self.tree.is_some()
            && self.tree == next.tree
            && self.entries.len() == next.entries.len()
            && self.entries.iter().zip(next.entries.iter()).all(|((p, a), (q, b))| {
                p == q && a.entry == b.entry && a.stat == b.stat && a.stat.mtime_ns < self.written_ns - racy_window(a.stat.mtime_ns)
            })
    }

    /// Whether `stat` proves the file still has the indexed content.
    fn unchanged(&self, entry: &IndexEntry, stat: Stat) -> bool {
        entry.stat == stat && stat.mtime_ns < self.written_ns - racy_window(stat.mtime_ns)
    }
}

/// How close to the index's writing a change can hide: a file written again
/// within the same timestamp tick looks unchanged. Timestamps here have
/// nanoseconds, and the kernel moves them every few milliseconds; a whole
/// second means a disk with coarse timestamps (FAT, old HFS+).
fn racy_window(mtime_ns: i64) -> i64 {
    if mtime_ns % 1_000_000_000 == 0 { 2_000_000_000 } else { 50_000_000 }
}

fn now_ns() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos() as i64)
        .unwrap_or(0)
}

// Rules for what a workspace keeps out of its trees.

/// What `vcs.toml` says.
#[derive(Debug, Default, Clone, serde::Deserialize)]
pub struct Config {
    #[serde(default)]
    pub workspace: WorkspaceConfig,
    #[serde(default)]
    pub mine: MineConfig,
    #[serde(default)]
    pub land: LandConfig,
}

/// `[land] run`: commands `vcs land` runs in the workspace first, in order
/// (lint, tests); if one fails, nothing lands. They're this machine's, as
/// git hooks are: the host's own rules are its settings.
#[derive(Debug, Default, Clone, serde::Deserialize)]
pub struct LandConfig {
    #[serde(default)]
    pub run: Vec<String>,
}

#[derive(Debug, Default, Clone, serde::Deserialize)]
pub struct WorkspaceConfig {
    #[serde(default)]
    pub copy: Vec<String>,
}

#[derive(Debug, Default, Clone, serde::Deserialize)]
pub struct MineConfig {
    #[serde(default)]
    pub paths: Vec<String>,
}

pub fn config(root: &Path) -> Config {
    fs::read_to_string(root.join("vcs.toml"))
        .ok()
        .and_then(|text| toml::from_str(&text).ok())
        .unwrap_or_default()
}

pub const DEFAULT_MINE: &[&str] = &[".env", ".env.*", "!.env.example", "!.env.sample", "!.env.template"];

/// Paths that stay the user's own: never saved into a change, never landed.
pub struct Mine {
    include: globset::GlobSet,
    exclude: globset::GlobSet,
    /// The rules as written, to tell when they change.
    pub rules: String,
}

impl Mine {
    /// The project's rules (vcs.toml, or the defaults) plus the person's own.
    pub fn new(config: &Config, personal: &[String]) -> Mine {
        let mut include = globset::GlobSetBuilder::new();
        let mut exclude = globset::GlobSetBuilder::new();
        let mut patterns: Vec<String> = if config.mine.paths.is_empty() {
            DEFAULT_MINE.iter().map(|s| s.to_string()).collect()
        } else {
            config.mine.paths.clone()
        };
        patterns.extend(personal.iter().cloned());
        let rules = patterns.join("\n");
        for pattern in patterns {
            let (set, p) = match pattern.strip_prefix('!') {
                Some(p) => (&mut exclude, p.to_string()),
                None => (&mut include, pattern),
            };
            // A bare name matches at any depth, as in .gitignore.
            let p = p.trim_end_matches('/').to_string();
            let globs = if p.contains('/') { vec![p.trim_start_matches('/').to_string()] } else { vec![p.clone(), format!("**/{p}")] };
            for g in globs {
                for g in [g.clone(), format!("{g}/**")] {
                    if let Ok(glob) = globset::GlobBuilder::new(&g).literal_separator(true).build() {
                        set.add(glob);
                    }
                }
            }
        }
        Mine {
            include: include.build().unwrap_or_else(|_| globset::GlobSet::empty()),
            exclude: exclude.build().unwrap_or_else(|_| globset::GlobSet::empty()),
            rules,
        }
    }

    pub fn matches(&self, path: &str) -> bool {
        self.include.is_match(path) && !self.exclude.is_match(path)
    }
}


/// What saving a workspace found.
pub struct Snapshot {
    pub tree: Id,
    /// The index to keep (None: the one given, unchanged).
    pub index: Option<Index>,
    /// Mine files in the folder (kept out of the tree): path, file ID, exec.
    pub mine: Vec<(String, Id, bool)>,
}

struct Seen {
    stat: Stat,
    kind: EntryKind,
    /// Not known to the index as it is now: its content is read.
    changed: bool,
}

/// Every file under `root` its .gitignore files leave in, never vcs's or
/// git's own folders, as "a/b" paths with their stats. Folders are read in
/// parallel, and each folder's files are handed over at once.
fn walk(root: &Path) -> Vec<(String, fs::Metadata)> {
    use ignore::gitignore::{Gitignore, GitignoreBuilder};
    use std::sync::{Arc, Mutex};
    type Rules = Arc<Vec<Arc<Gitignore>>>;
    type Found = Mutex<Vec<(String, fs::Metadata)>>;

    // The deepest .gitignore with something to say about a path decides.
    fn ignored(rules: &Rules, path: &Path, is_dir: bool) -> bool {
        for g in rules.iter().rev() {
            match g.matched(path, is_dir) {
                ignore::Match::Ignore(_) => return true,
                ignore::Match::Whitelist(_) => return false,
                ignore::Match::None => {}
            }
        }
        false
    }

    fn visit<'s>(scope: &rayon::Scope<'s>, dir: PathBuf, rel: String, rules: Rules, found: &'s Found) {
        let Ok(read) = fs::read_dir(&dir) else { return };
        let entries: Vec<fs::DirEntry> = read.flatten().collect();
        let rules = match entries.iter().find(|e| e.file_name() == ".gitignore" && e.file_type().is_ok_and(|t| t.is_file())) {
            Some(e) => {
                let mut builder = GitignoreBuilder::new(&dir);
                let _ = builder.add(e.path());
                match builder.build() {
                    Ok(g) => Arc::new(rules.iter().cloned().chain([Arc::new(g)]).collect()),
                    Err(_) => rules,
                }
            }
            None => rules,
        };
        let mut files = Vec::new();
        for entry in entries {
            let name = entry.file_name();
            // Names no project can hold, and vcs's and git's own folders at
            // any depth (a nested git checkout too).
            let Some(name) = name.to_str().filter(|n| tree::safe_name(n)) else { continue };
            let Ok(kind) = entry.file_type() else { continue };
            let child = if rel.is_empty() { name.to_string() } else { format!("{rel}/{name}") };
            let path = entry.path();
            if kind.is_dir() {
                if !ignored(&rules, &path, true) {
                    let rules = rules.clone();
                    scope.spawn(move |s| visit(s, path, child, rules, found));
                }
            } else if !ignored(&rules, &path, false)
                && let Ok(meta) = entry.metadata()
            {
                files.push((child, meta));
            }
        }
        if !files.is_empty() {
            found.lock().unwrap().extend(files);
        }
    }

    let found = Mutex::new(Vec::new());
    rayon::scope(|s| visit(s, root.to_path_buf(), String::new(), Arc::new(Vec::new()), &found));
    found.into_inner().unwrap()
}


fn in_scope(only: &[String], path: &str) -> bool {
    only.is_empty() || only.iter().any(|o| tree::under(path, o))
}

/// Saves the folder's content into the store and returns its tree. `base`
/// is the tree the change starts from: files it tracks are kept even when
/// .gitignore matches them, and with `only` set every path outside those
/// folders comes from it unchanged.
/// `main_paths` answers the files the change's base tracks (asked only when
/// a file looks like a mine file): rules for mine files never take a file
/// main already has.
pub fn snapshot(
    root: &Path,
    store: &Store,
    base: &Files,
    only: &[String],
    index: &Index,
    personal: &[String],
    main_paths: &dyn Fn() -> Result<std::collections::HashSet<String>>,
) -> Result<Snapshot> {
    let config = config(root);
    let mine = Mine::new(&config, personal);

    // What a file on disk is now, and whether it changed since the index.
    let look = |path: &str, meta: &fs::Metadata| -> Seen {
        let stat = Stat::of(meta);
        match index.entries.get(path) {
            Some(e) => {
                let kind = sys::kind_of(meta, Some(e.entry.kind));
                Seen { stat, kind, changed: !(index.unchanged(e, stat) && e.entry.kind == kind) }
            }
            None => Seen { stat, kind: sys::kind_of(meta, base.get(path).map(|e| e.kind)), changed: true },
        }
    };
    // One pass over the walk and the index, both sorted: each file's state,
    // the files the index doesn't have yet, and tracked files the walk
    // skipped (ignored but tracked, which still count).
    let mut walked = walk(root);
    walked.par_sort_unstable_by(|a, b| a.0.cmp(&b.0));
    let mut list: Vec<(String, Seen)> = Vec::with_capacity(walked.len());
    let mut fresh: Vec<String> = Vec::new();
    let mut skipped: Vec<&String> = Vec::new();
    let mut tracked = index.entries.iter().peekable();
    for (path, meta) in walked {
        while let Some((p, _)) = tracked.next_if(|(p, _)| p.as_str() < path.as_str()) {
            skipped.push(p);
        }
        let stat = Stat::of(&meta);
        let seen = match tracked.next_if(|(p, _)| **p == path) {
            Some((_, e)) => {
                let kind = sys::kind_of(&meta, Some(e.entry.kind));
                Seen { stat, kind, changed: !(index.unchanged(e, stat) && e.entry.kind == kind) }
            }
            None => {
                fresh.push(path.clone());
                look(&path, &meta)
            }
        };
        list.push((path, seen));
    }
    skipped.extend(tracked.map(|(p, _)| p));
    let mut seen: BTreeMap<String, Seen> = list.into_iter().collect();
    if !only.is_empty() {
        skipped.extend(base.keys().filter(|p| !seen.contains_key(*p) && !index.entries.contains_key(*p)));
    }
    for path in skipped {
        if !in_scope(only, path) {
            continue;
        }
        if let Ok(meta) = root.join(path).symlink_metadata()
            && !meta.is_dir()
        {
            seen.insert(path.clone(), look(path, &meta));
        }
    }

    // Submodules are pointers, not files here: they carry over as they are,
    // and nothing inside their folders is tracked.
    let links: Vec<(String, Entry)> = if only.is_empty() {
        index.entries.iter().filter(|(_, e)| e.entry.kind == EntryKind::Gitlink).map(|(p, e)| (p.clone(), e.entry)).collect()
    } else {
        base.iter().filter(|(p, e)| e.kind == EntryKind::Gitlink && in_scope(only, p)).map(|(p, e)| (p.clone(), *e)).collect()
    };
    if !links.is_empty() {
        seen.retain(|path, _| !links.iter().any(|(l, _)| tree::under(path, l)));
    }

    // Mine files stay out of trees unless main already tracks them. Files
    // the index has were sorted out when they came, so only new ones are
    // looked at, unless the rules changed since; main's paths are read only
    // when some file looks like one.
    let rules_file = root.join(DIR).join("mine-rules");
    let rules_changed = fs::read_to_string(&rules_file).ok().as_deref() != Some(mine.rules.as_str());
    let candidates: Vec<String> =
        if rules_changed { seen.keys().filter(|p| mine.matches(p)).cloned().collect() } else { fresh.into_iter().filter(|p| mine.matches(p)).collect() };
    let main_paths = if candidates.is_empty() { Default::default() } else { main_paths()? };
    let mut mine_paths = Vec::new();
    for path in candidates.into_iter().filter(|p| !main_paths.contains(p)) {
        if let Some(s) = seen.remove(&path) {
            mine_paths.push((root.join(&path), s.kind == EntryKind::Exec, path));
        }
    }
    if rules_changed {
        let _ = fs::write(&rules_file, &mine.rules);
    }
    if !only.is_empty() {
        seen.retain(|path, _| in_scope(only, path));
    }

    // Read and store what might have changed, in parallel. Objects are
    // files named by their IDs, so parallel writes are safe.
    let jobs: Vec<(&String, &Seen)> = seen.iter().filter(|(_, s)| s.changed).collect();
    let mut mine_out = Vec::new();
    for (abs, exec, path) in mine_paths {
        let bytes = read_content(&abs, EntryKind::File)?;
        mine_out.push((path, store.put_file(&bytes)?, exec));
    }
    // Nothing changed since the index (what a status or a land usually
    // finds): its tree stands, and nothing is rebuilt.
    if jobs.is_empty()
        && only.is_empty()
        && links.is_empty()
        && seen.len() == index.entries.len()
        && let Some(tree) = index.tree.filter(|t| store.has(t))
        && seen.keys().eq(index.entries.keys())
    {
        return Ok(Snapshot { tree, index: None, mine: mine_out });
    }
    let stored: BTreeMap<String, Id> = jobs
        .par_iter()
        .map(|(path, s)| -> Result<(String, Id)> {
            let bytes = read_content(&root.join(path), s.kind)?;
            Ok(((*path).clone(), store.put_file(&bytes)?))
        })
        .collect::<Result<BTreeMap<_, _>>>()?;

    let mut files = Files::new();
    for (path, entry) in base {
        if !in_scope(only, path) {
            files.insert(path.clone(), *entry);
        }
    }
    let mut next = Index::default();
    for (path, entry) in &links {
        files.insert(path.clone(), *entry);
        let stat = index.entries.get(path).map(|e| e.stat).unwrap_or(Stat { size: 0, mtime_ns: 0, ino: 0 });
        next.entries.insert(path.clone(), IndexEntry { entry: *entry, stat });
    }
    for (path, s) in &seen {
        let id = match stored.get(path) {
            Some(id) => *id,
            None => index.entries[path].entry.id,
        };
        let entry = Entry { kind: s.kind, id };
        files.insert(path.clone(), entry);
        next.entries.insert(path.clone(), IndexEntry { entry, stat: s.stat });
    }
    // Nothing new since the index's tree: it's still the tree.
    let unchanged = stored.is_empty() && only.is_empty() && next.entries.len() == index.entries.len() && next.entries.keys().eq(index.entries.keys());
    let tree = match index.tree {
        Some(t) if unchanged && store.has(&t) => t,
        _ => {
            let (tree, objects) = tree::build(&files);
            for o in &objects {
                store.put(o)?;
            }
            tree
        }
    };
    next.tree = Some(tree);
    Ok(Snapshot { tree, index: Some(next), mine: mine_out })
}

fn read_content(path: &Path, kind: EntryKind) -> Result<Vec<u8>> {
    if kind == EntryKind::Link {
        return Ok(sys::read_link(path)?);
    }
    let mut bytes = Vec::new();
    fs::File::open(path)
        .with_context(|| format!("reading {}", path.display()))?
        .read_to_end(&mut bytes)?;
    Ok(bytes)
}

/// Another workspace to copy unchanged files from.
pub struct CowSource<'a> {
    pub root: &'a Path,
    pub index: &'a Index,
}

/// Makes the folder hold exactly `target` (within `only`), changing only
/// what differs from `index`. Untracked and ignored files are left alone.
/// Returns the new index.
pub fn materialize(root: &Path, store: &Store, index: &Index, target: &Files, only: &[String], cow: Option<CowSource>) -> Result<Index> {
    let wanted: BTreeMap<&String, &Entry> = target.iter().filter(|(p, _)| in_scope(only, p)).collect();

    // Remove tracked files that the target doesn't have.
    let mut removed_dirs = Vec::new();
    for path in index.entries.keys() {
        if !wanted.contains_key(path) {
            let abs = root.join(path);
            if abs.symlink_metadata().is_ok_and(|m| !m.is_dir()) {
                fs::remove_file(&abs)?;
            }
            if let Some(parent) = abs.parent() {
                removed_dirs.push(parent.to_path_buf());
            }
        }
    }
    // Empty folders left behind go too (never the root).
    removed_dirs.sort_by_key(|d| std::cmp::Reverse(d.components().count()));
    removed_dirs.dedup();
    for dir in removed_dirs {
        let mut d = dir.as_path();
        while d != root && d.starts_with(root) {
            if fs::remove_dir(d).is_err() {
                break;
            }
            match d.parent() {
                Some(p) => d = p,
                None => break,
            }
        }
    }

    // Never vcs's or git's own folders, or outside the workspace, whatever a
    // host sends; and nothing this system can't have a file by.
    if let Some(bad) = wanted.keys().find(|p| !tree::valid_path(p) || !p.split('/').all(sys::fits)) {
        return Err(anyhow!(
            "{bad} can't be written here: {}",
            if tree::valid_path(bad) { "this system can't have a file by that name" } else { "it's vcs's or git's own folder, or not a safe path" }
        ));
    }

    // Write what's new or different.
    let mut keep = Index::default();
    let mut jobs = Vec::new();
    for (path, entry) in &wanted {
        if let Some(existing) = index.entries.get(*path)
            && existing.entry == **entry
            && let Ok(meta) = root.join(path).symlink_metadata()
            && Stat::of(&meta) == existing.stat
        {
            keep.entries.insert((*path).clone(), *existing);
            continue;
        }
        jobs.push(((*path).clone(), **entry));
    }
    // Links go last, so what they point at is there first (Windows makes a
    // link to a folder differently from one to a file).
    let (links, files): (Vec<(String, Entry)>, Vec<(String, Entry)>) = jobs.into_iter().partition(|(_, e)| e.kind == EntryKind::Link);
    let write = |(path, entry): &(String, Entry)| -> Result<(String, IndexEntry)> {
            let abs = root.join(path);
            if let Some(parent) = abs.parent() {
                make_dirs(root, parent)?;
            }
            if entry.kind == EntryKind::Gitlink {
                // A submodule's folder, left as it is (checked out or not), or
                // made empty, as git leaves one it hasn't checked out.
                if abs.symlink_metadata().is_ok_and(|m| !m.is_dir()) {
                    fs::remove_file(&abs)?;
                }
                make_dirs(root, &abs)?;
                let meta = abs.symlink_metadata()?;
                return Ok((path.clone(), IndexEntry { entry: *entry, stat: Stat::of(&meta) }));
            }
            if let Ok(meta) = abs.symlink_metadata() {
                if meta.is_dir() {
                    fs::remove_dir_all(&abs)?;
                } else {
                    fs::remove_file(&abs)?;
                }
            }
            let copied = match &cow {
                Some(src) if entry.kind != EntryKind::Link => copy_from(src, path, entry, &abs),
                _ => false,
            };
            if !copied {
                let content = store.read_file(&entry.id).with_context(|| format!("missing content for {path}"))?;
                if entry.kind == EntryKind::Link {
                    sys::write_link(&String::from_utf8_lossy(&content), &abs)?;
                } else {
                    fs::write(&abs, content)?;
                }
            }
            if entry.kind != EntryKind::Link {
                sys::set_exec(&abs, entry.kind == EntryKind::Exec)?;
            }
            let meta = abs.symlink_metadata()?;
            Ok((path.clone(), IndexEntry { entry: *entry, stat: Stat::of(&meta) }))
    };
    let mut written: Vec<(String, IndexEntry)> = files.par_iter().map(write).collect::<Result<Vec<_>>>()?;
    written.extend(links.par_iter().map(write).collect::<Result<Vec<_>>>()?);
    for (path, e) in written {
        keep.entries.insert(path, e);
    }
    Ok(keep)
}

/// Creates `dir` and its parents, replacing any file in the way.
fn make_dirs(root: &Path, dir: &Path) -> Result<()> {
    if dir.is_dir() {
        return Ok(());
    }
    let mut missing = Vec::new();
    let mut d = dir;
    while d != root && !d.is_dir() {
        missing.push(d.to_path_buf());
        match d.parent() {
            Some(p) => d = p,
            None => break,
        }
    }
    for m in missing.into_iter().rev() {
        if m.symlink_metadata().is_ok_and(|meta| !meta.is_dir()) {
            fs::remove_file(&m)?;
        }
        match fs::create_dir(&m) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(e) => return Err(e.into()),
        }
    }
    Ok(())
}

/// Copies one file from another workspace if it still has the same content.
fn copy_from(src: &CowSource, path: &str, entry: &Entry, dest: &Path) -> bool {
    let Some(theirs) = src.index.entries.get(path) else { return false };
    if theirs.entry != *entry {
        return false;
    }
    let from = src.root.join(path);
    let Ok(meta) = from.symlink_metadata() else { return false };
    if Stat::of(&meta) != theirs.stat {
        return false;
    }
    reflink_copy::reflink_or_copy(&from, dest).is_ok()
}

/// Copies a build folder (like node_modules) from another workspace,
/// copy-on-write where the disk supports it. Returns how many files.
pub fn copy_folder(from: &Path, to: &Path) -> Result<u64> {
    let mut count = 0;
    if !from.is_dir() || to.exists() {
        return Ok(0);
    }
    let mut stack = vec![from.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let rel = dir.strip_prefix(from)?;
        fs::create_dir_all(to.join(rel))?;
        for entry in fs::read_dir(&dir)? {
            let entry = entry?;
            let path = entry.path();
            let dest = to.join(path.strip_prefix(from)?);
            let kind = entry.file_type()?;
            if kind.is_dir() {
                stack.push(path);
            } else if kind.is_symlink() {
                sys::write_link(&String::from_utf8_lossy(&sys::read_link(&path)?), &dest)?;
                count += 1;
            } else {
                reflink_copy::reflink_or_copy(&path, &dest)?;
                count += 1;
            }
        }
    }
    Ok(count)
}

/// Whether text still holds conflict markers.
pub fn has_markers(bytes: &[u8]) -> bool {
    let text = String::from_utf8_lossy(bytes);
    let mut open = false;
    let mut middle = false;
    for line in text.lines() {
        if line.starts_with("<<<<<<< ") || line == "<<<<<<<" {
            open = true;
        } else if open && line == "=======" {
            middle = true;
        } else if open && middle && (line.starts_with(">>>>>>> ") || line == ">>>>>>>") {
            return true;
        }
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn markers() {
        assert!(has_markers(b"a\n<<<<<<< main 4\nx\n=======\ny\n>>>>>>> your change bcdf\nb\n"));
        assert!(!has_markers(b"a\n=======\nb\n"));
    }

    #[test]
    fn mine_defaults() {
        let mine = Mine::new(&Config::default(), &[]);
        assert!(mine.matches(".env"));
        assert!(mine.matches(".env.production"));
        assert!(mine.matches("apps/web/.env.local"));
        assert!(!mine.matches(".env.example"));
        assert!(!mine.matches("src/env.ts"));
    }

    #[test]
    fn snapshot_and_materialize() {
        let objects = tempfile::tempdir().unwrap();
        let store = Store::open(objects.path()).unwrap();
        let ws = tempfile::tempdir().unwrap();
        let root = ws.path();
        write_id(root, "w1").unwrap();
        fs::create_dir_all(root.join("src")).unwrap();
        fs::write(root.join("README.md"), "hi\n").unwrap();
        fs::write(root.join("src/a.rs"), "fn a() {}\n").unwrap();
        fs::write(root.join(".gitignore"), "target/\n").unwrap();
        fs::create_dir_all(root.join("target")).unwrap();
        fs::write(root.join("target/junk"), "x").unwrap();
        fs::write(root.join(".env"), "SECRET=1\n").unwrap();
        let snap = snapshot(root, &store, &Files::new(), &[], &Index::default(), &[], &|| Ok(Default::default())).unwrap();
        let files = tree::flatten(&store, &snap.tree).unwrap();
        assert!(files.contains_key("README.md"));
        assert!(files.contains_key("src/a.rs"));
        assert!(files.contains_key(".gitignore"));
        assert!(!files.contains_key("target/junk"));
        assert!(!files.contains_key(".env"));
        assert_eq!(snap.mine.len(), 1);
        let index = snap.index.unwrap();

        // Nothing new: the same tree, and the index stands.
        let again = snapshot(root, &store, &Files::new(), &[], &index, &[], &|| Ok(Default::default())).unwrap();
        assert_eq!(again.tree, snap.tree);

        let ws2 = tempfile::tempdir().unwrap();
        write_id(ws2.path(), "w2").unwrap();
        let index2 = materialize(ws2.path(), &store, &Index::default(), &files, &[], Some(CowSource { root, index: &index })).unwrap();
        assert_eq!(fs::read_to_string(ws2.path().join("src/a.rs")).unwrap(), "fn a() {}\n");
        let mut fewer = files.clone();
        fewer.remove("src/a.rs");
        materialize(ws2.path(), &store, &index2, &fewer, &[], None).unwrap();
        assert!(!ws2.path().join("src").exists());
    }

    /// An executable or a link stays one when it's edited, on a disk that
    /// can't mark executables or make links (Windows).
    #[test]
    fn kinds_survive_edits() {
        let objects = tempfile::tempdir().unwrap();
        let store = Store::open(objects.path()).unwrap();
        let ws = tempfile::tempdir().unwrap();
        let root = ws.path();
        write_id(root, "w1").unwrap();
        let mut files = Files::new();
        files.insert("run.sh".into(), Entry { kind: EntryKind::Exec, id: store.put_file(b"echo 1\n").unwrap() });
        files.insert("link".into(), Entry { kind: EntryKind::Link, id: store.put_file(b"run.sh").unwrap() });
        let index = materialize(root, &store, &Index::default(), &files, &[], None).unwrap();
        fs::write(root.join("run.sh"), "echo 2\n").unwrap();
        let snap = snapshot(root, &store, &Files::new(), &[], &index, &[], &|| Ok(Default::default())).unwrap();
        let got = tree::flatten(&store, &snap.tree).unwrap();
        assert_eq!(got["run.sh"].kind, EntryKind::Exec);
        assert_eq!(got["run.sh"].id, store.put_file(b"echo 2\n").unwrap());
        assert_eq!(got["link"], files["link"]);
    }
}